From ea114153b92686cd4abb96a52fa6fc00d889de12 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sat, 14 Feb 2026 14:17:32 +0300 Subject: [PATCH] =?UTF-8?q?fix(generator):=20=D0=BE=D0=B1=D1=80=D0=B0?= =?UTF-8?q?=D0=B1=D0=BE=D1=82=D0=BA=D0=B0=20findings=20=D0=B8=D0=B7=20?= =?UTF-8?q?=D1=84=D0=B8=D0=BD=D0=B0=D0=BB=D1=8C=D0=BD=D0=BE=D0=B3=D0=BE=20?= =?UTF-8?q?=D1=80=D0=B5=D0=B2=D1=8C=D1=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Изолированы ошибки history-канала: best-effort с логированием, не валят тик - Добавлено явное создание топика generator_batch_history при старте - Добавлена защита от пустого словаря (ValueError при загрузке) - Обновлена документация о структуре тестов - Исправлены тесты на пустой словарь Ревью: изоляция ошибок history, явное создание топика, защита от пустых данных --- generator/README.md | 10 ++- generator/generator.py | 120 +++++++++++++++++++++-------- generator/tests/test_generation.py | 24 +----- 3 files changed, 96 insertions(+), 58 deletions(-) diff --git a/generator/README.md b/generator/README.md index 192be01..a10b0fb 100644 --- a/generator/README.md +++ b/generator/README.md @@ -132,10 +132,12 @@ docker run --rm -v $(PWD):/workspace -w /workspace/generator generator:test pyte ``` generator/tests/ -├── conftest.py # Fixtures pytest -├── test_config.py # Тесты конфигурации -├── test_generation.py # Тесты генерации событий -└── test_history.py # Тесты истории батчей +├── conftest.py # Fixtures pytest +├── test_config.py # Тесты конфигурации +├── test_generation.py # Тесты генерации событий +├── test_history.py # Тесты структуры BatchRecord +├── test_kafka_history.py # Тесты KafkaBatchHistory +└── test_service.py # Тесты GeneratorService ``` ### Интеграционный тест diff --git a/generator/generator.py b/generator/generator.py index 4b7f571..0c630f6 100644 --- a/generator/generator.py +++ b/generator/generator.py @@ -174,11 +174,19 @@ class EventDictionary: logger.info(f" Loaded {len(events)} events from {filename}") return events + browser_events = load_jsonl("browser_events.jsonl") + location_events = load_jsonl("location_events.jsonl") + device_events = load_jsonl("device_events.jsonl") + geo_events = load_jsonl("geo_events.jsonl") + + if not browser_events: + raise ValueError("browser_events.jsonl is empty or missing") + return cls( - browser_events=load_jsonl("browser_events.jsonl"), - location_events=load_jsonl("location_events.jsonl"), - device_events=load_jsonl("device_events.jsonl"), - geo_events=load_jsonl("geo_events.jsonl"), + browser_events=browser_events, + location_events=location_events, + device_events=device_events, + geo_events=geo_events, ) @@ -249,6 +257,15 @@ class EventGenerator: Возвращает словарь {topic: [events]} """ + if not self.dictionary.browser_events: + logger.warning("Event dictionary is empty, skipping batch generation") + return { + "browser_events": [], + "location_events": [], + "device_events": [], + "geo_events": [], + } + batch = { "browser_events": [], "location_events": [], @@ -338,6 +355,30 @@ class BatchRecord: } +# --------------------------------------------------------------------------- +# Создание топика для истории +# --------------------------------------------------------------------------- +def ensure_history_topic(bootstrap_servers: str) -> None: + """Создаёт топик для истории батчей если он не существует.""" + from kafka import KafkaAdminClient + from kafka.admin import NewTopic + from kafka.errors import TopicAlreadyExistsError + + admin_client = KafkaAdminClient(bootstrap_servers=bootstrap_servers) + try: + new_topic = NewTopic( + name=KafkaBatchHistory.HISTORY_TOPIC, + num_partitions=1, + replication_factor=1, + ) + admin_client.create_topics([new_topic]) + logger.info(f"Created topic: {KafkaBatchHistory.HISTORY_TOPIC}") + except TopicAlreadyExistsError: + logger.debug(f"Topic already exists: {KafkaBatchHistory.HISTORY_TOPIC}") + finally: + admin_client.close() + + # --------------------------------------------------------------------------- # Kafka history - пишет историю в отдельный топик # --------------------------------------------------------------------------- @@ -486,6 +527,8 @@ class GeneratorService: f"jitter={self.config.jitter_pct}%") # Подключаемся к Kafka для публикации событий и истории + # Сначала создаём топик для истории если нужно + ensure_history_topic(self.config.kafka_bootstrap_servers) self.publisher = KafkaPublisher(self.config.kafka_bootstrap_servers) self.history = KafkaBatchHistory(self.config.kafka_bootstrap_servers) @@ -554,26 +597,30 @@ class GeneratorService: if status in ("success", "partial"): METRICS_LAST_SUCCESS.set_to_current_time() - # Сохраняем в историю (с fallback на in-memory при деградации Kafka) - batch_record = BatchRecord( - batch_id=batch_id, - started_at=datetime.fromtimestamp(tick_start, tz=timezone.utc), - finished_at=datetime.now(timezone.utc), - sent_total=total_sent, - sent_browser=sent_counts.get("browser_events", {}).get("sent", 0), - sent_location=sent_counts.get("location_events", {}).get("sent", 0), - sent_device=sent_counts.get("device_events", {}).get("sent", 0), - sent_geo=sent_counts.get("geo_events", {}).get("sent", 0), - status=status, - error_message=None if status == "success" else f"Errors: {total_errors}", - ) - self.history.add(batch_record) - - # Флашим публикацию и историю + # Флашим публикацию событий self.publisher.flush() - self.history.flush() pub_duration = time.time() - pub_start + # Сохраняем в историю (best-effort: ошибки не валят тик) + try: + batch_record = BatchRecord( + batch_id=batch_id, + started_at=datetime.fromtimestamp(tick_start, tz=timezone.utc), + finished_at=datetime.now(timezone.utc), + sent_total=total_sent, + sent_browser=sent_counts.get("browser_events", {}).get("sent", 0), + sent_location=sent_counts.get("location_events", {}).get("sent", 0), + sent_device=sent_counts.get("device_events", {}).get("sent", 0), + sent_geo=sent_counts.get("geo_events", {}).get("sent", 0), + status=status, + error_message=None if status == "success" else f"Errors: {total_errors}", + ) + self.history.add(batch_record) + self.history.flush() + except Exception as hist_err: + logger.warning(f"Failed to write batch history: {hist_err}") + METRICS_ERRORS_TOTAL.labels(topic="history").inc() + # Логируем результат tick_duration = time.time() - tick_start logger.info( @@ -589,20 +636,25 @@ class GeneratorService: except Exception as e: logger.exception(f"Error in tick {tick}: {e}") - self.history.add( - BatchRecord( - batch_id=batch_id, - started_at=datetime.fromtimestamp(tick_start, tz=timezone.utc), - finished_at=datetime.now(timezone.utc), - sent_total=0, - sent_browser=0, - sent_location=0, - sent_device=0, - sent_geo=0, - status="error", - error_message=str(e), + # Пытаемся записать ошибку в историю (best-effort) + try: + self.history.add( + BatchRecord( + batch_id=batch_id, + started_at=datetime.fromtimestamp(tick_start, tz=timezone.utc), + finished_at=datetime.now(timezone.utc), + sent_total=0, + sent_browser=0, + sent_location=0, + sent_device=0, + sent_geo=0, + status="error", + error_message=str(e), + ) ) - ) + self.history.flush() + except Exception as hist_err: + logger.warning(f"Failed to write error to history: {hist_err}") # Ждём до следующего тика elapsed = time.time() - tick_start diff --git a/generator/tests/test_generation.py b/generator/tests/test_generation.py index 7b750aa..5ad968d 100644 --- a/generator/tests/test_generation.py +++ b/generator/tests/test_generation.py @@ -168,23 +168,7 @@ class TestPoissonDistribution: class TestEmptyData: """Тесты обработки пустых данных.""" - def test_empty_jsonl_handled(self, empty_temp_dir): - """Пустые JSONL файлы обрабатываются корректно.""" - dictionary = EventDictionary.load(empty_temp_dir) - - assert len(dictionary.browser_events) == 0 - assert len(dictionary.location_events) == 0 - assert len(dictionary.device_events) == 0 - assert len(dictionary.geo_events) == 0 - - def test_generate_batch_with_empty_dict(self, empty_temp_dir, base_config): - """Генерация с пустым словарем (должна работать, но без событий).""" - from dataclasses import replace - dictionary = EventDictionary.load(empty_temp_dir) - config = replace(base_config, data_dir=empty_temp_dir) - generator = EventGenerator(dictionary, config) - - # С пустым словарем генерация упадет при choice() - # Это ожидаемое поведение — проверяем что падает с IndexError - with pytest.raises(IndexError): - generator.generate_batch(1) + def test_empty_jsonl_raises_error(self, empty_temp_dir): + """Пустые JSONL файлы вызывают ValueError при загрузке.""" + with pytest.raises(ValueError, match="browser_events.jsonl is empty"): + EventDictionary.load(empty_temp_dir)