From e8d1c9efa90a7774aa3472b48daf42a0cabc932a Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Sat, 14 Feb 2026 12:47:01 +0300 Subject: [PATCH] =?UTF-8?q?feat(infra):=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=20=D0=B0=D0=B2=D1=82=D0=BE=D0=BD=D0=BE=D0=BC?= =?UTF-8?q?=D0=BD=D1=8B=D0=B9=20=D0=B3=D0=B5=D0=BD=D0=B5=D1=80=D0=B0=D1=82?= =?UTF-8?q?=D0=BE=D1=80=20=D1=81=D0=BE=D0=B1=D1=8B=D1=82=D0=B8=D0=B9=20?= =?UTF-8?q?=D0=B4=D0=BB=D1=8F=20Kafka?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - нужен постоянный поток данных для демонстрации работы стека - текущий batch-загрузчик не позволяет показать streaming-сценарии - Что: - добавлен сервис generator с режимом steady (Poisson-интенсивность) - генератор публикует в 4 топика: browser/location/device/geo_events - сохраняются связи event_id и click_id между событиями - сборка через uv для скорости и компактности образа - добавлены команды generator-* в Makefile - комплексные тесты: валидация, статистика, формат сообщений - Проверка: - `docker run --rm -v $(pwd)/..:/workspace -w /workspace/generator generator:test python test_comprehensive.py` — 8/8 тестов - `make generator-up` — 3 тика без ошибок, отправлено 2904 сообщения --- Makefile | 23 +- docker-compose.yml | 31 ++ generator/Dockerfile | 26 ++ generator/README.md | 108 +++++++ generator/generator.py | 542 ++++++++++++++++++++++++++++++++ generator/requirements.txt | 8 + generator/test_comprehensive.py | 445 ++++++++++++++++++++++++++ generator/test_local.py | 88 ++++++ generator/verify_kafka.py | 62 ++++ 9 files changed, 1332 insertions(+), 1 deletion(-) create mode 100644 generator/Dockerfile create mode 100644 generator/README.md create mode 100644 generator/generator.py create mode 100644 generator/requirements.txt create mode 100644 generator/test_comprehensive.py create mode 100644 generator/test_local.py create mode 100644 generator/verify_kafka.py diff --git a/Makefile b/Makefile index 3c7b388..feb6375 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,7 @@ .PHONY: up down clean ddl data transform logs \ reload-monitoring recover-monitoring \ - superset-init superset-dashboard superset-ui superset-restart + superset-init superset-dashboard superset-ui superset-restart \ + generator-up generator-down generator-logs generator-restart COMPOSE ?= docker compose @@ -85,3 +86,23 @@ superset-ui: # Перезапуск Superset superset-restart: $(COMPOSE) restart superset + +# ============================================================================ +# Генератор событий (автономный стриминг) +# ============================================================================ + +# Запустить только генератор (полезно для отладки) +generator-up: + $(COMPOSE) up -d --build generator + +# Остановить генератор +generator-down: + $(COMPOSE) stop generator + +# Логи генератора +generator-logs: + $(COMPOSE) logs -f --tail=100 generator + +# Перезапуск генератора с пересборкой +generator-restart: + $(COMPOSE) up -d --build --force-recreate generator diff --git a/docker-compose.yml b/docker-compose.yml index 223b9b6..da340e5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -284,6 +284,37 @@ services: postgres-metadata: condition: service_healthy + # Генератор событий (автономный стриминг в Kafka) + generator: + build: + context: ./generator + dockerfile: Dockerfile + environment: + KAFKA_BOOTSTRAP_SERVERS: kafka:29092 + GEN_TICK_SECONDS: "60" + GEN_LAMBDA_BASE_PER_MIN: "200" + GEN_JITTER_PCT: "20" + GEN_MIN_EVENTS_PER_TICK: "50" + GEN_MAX_EVENTS_PER_TICK: "500" + GEN_DATA_DIR: /data + GEN_ENABLED: "true" + CLICKHOUSE_HOST: clickhouse + CLICKHOUSE_PORT: "9000" + # PYTHONUNBUFFERED для сразу видеть логи + PYTHONUNBUFFERED: "1" + volumes: + - ./data:/data:ro + networks: + - cs_dwh + depends_on: + - kafka + restart: unless-stopped + healthcheck: + test: ["CMD", "python", "-c", "import sys; sys.exit(0)"] + interval: 30s + timeout: 10s + retries: 3 + # Kafka Exporter для мониторинга через Prometheus kafka-exporter: image: danielqsj/kafka-exporter:v1.9.0 diff --git a/generator/Dockerfile b/generator/Dockerfile new file mode 100644 index 0000000..1cb0d88 --- /dev/null +++ b/generator/Dockerfile @@ -0,0 +1,26 @@ +FROM python:3.12-slim + +# Установка uv +COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/ + +WORKDIR /app + +# Копируем зависимости отдельно для кэширования слоёв +COPY requirements.txt . + +# Установка зависимостей через uv (компактный venv) +RUN uv venv /app/.venv && \ + uv pip install --python /app/.venv/bin/python -r requirements.txt + +# Копируем код +COPY generator.py . + +# Не-root пользователь для безопасности +RUN useradd -m -u 1000 generator && chown -R generator:generator /app +USER generator + +# Активируем venv через PATH +ENV PATH="/app/.venv/bin:$PATH" + +# Запуск генератора +CMD ["python", "generator.py"] diff --git a/generator/README.md b/generator/README.md new file mode 100644 index 0000000..a956ebc --- /dev/null +++ b/generator/README.md @@ -0,0 +1,108 @@ +# Генератор событий (MVP) + +Автономный сервис для стриминга событий в Kafka. + +## Архитектура + +``` +generator-service -> Kafka topics -> (потребители отдельно) +``` + +Генератор работает автономно и не зависит от потребителей (Airflow, ClickHouse). + +## Режим работы: `steady` + +- Каждую минуту публикуем переменный объём событий (Poisson + jitter) +- Распределяем события по 4 топикам: + - `browser_events` + - `location_events` + - `device_events` + - `geo_events` +- Сохраняем связи `event_id <-> location`, `click_id <-> device/geo` + +## Конфигурация (env) + +| Переменная | Описание | По умолчанию | +|------------|----------|--------------| +| `KAFKA_BOOTSTRAP_SERVERS` | Адрес Kafka | `kafka:29092` | +| `GEN_TICK_SECONDS` | Интервал между тиками | `60` | +| `GEN_LAMBDA_BASE_PER_MIN` | Базовая интенсивность (событий/мин) | `200` | +| `GEN_JITTER_PCT` | Процент вариативности | `20` | +| `GEN_MIN_EVENTS_PER_TICK` | Минимум событий за тик | `50` | +| `GEN_MAX_EVENTS_PER_TICK` | Максимум событий за тик | `500` | +| `GEN_DATA_DIR` | Путь к JSONL файлам | `/data` | +| `GEN_SEED` | Сид для воспроизводимости | — | +| `GEN_ENABLED` | Включить генерацию | `true` | + +## Управление через Makefile + +```bash +# Запустить только генератор +make generator-up + +# Остановить генератор +make generator-down + +# Смотреть логи +make generator-logs + +# Перезапуск с пересборкой +make generator-restart +``` + +## Логи и метрики + +### Логи + +``` +=== Tick 1 (batch_id=a1b2c3d4) === +Generating ~156 base events +Batch a1b2c3d4 completed: sent=624, errors=0, gen_time=0.012s, pub_time=0.234s, total_time=0.247s + browser_events: 156 sent + location_events: 156 sent + device_events: 156 sent + geo_events: 156 sent +Sleeping for 59.8s until next tick +``` + +### Метрики (в коде) + +- `generator_events_total` — всего отправлено событий +- `generator_publish_errors_total` — ошибки публикации +- `generator_tick_duration_seconds` — длительность тика + +## Тестирование + +### Локальные тесты + +```bash +# Базовые тесты +docker run --rm -v $(pwd)/..:/workspace -w /workspace/generator generator:test python test_local.py + +# Комплексные тесты (проверка граничных случаев, статистики, формата) +docker run --rm -v $(pwd)/..:/workspace -w /workspace/generator generator:test python test_comprehensive.py +``` + +### Интеграционный тест + +```bash +# Запустить стек с генератором +make generator-up + +# Проверить логи +make generator-logs + +# Проверить сообщения в Kafka +docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh \ + --bootstrap-server kafka:29092 --topic browser_events --from-beginning +``` + +## История batch + +Хранится в памяти (последние 1000 записей). Поля: + +- `batch_id` — идентификатор батча +- `started_at` / `finished_at` — время начала/окончания +- `sent_total` — всего отправлено +- `sent_browser/location/device/geo` — по топикам +- `status` — success/partial/error diff --git a/generator/generator.py b/generator/generator.py new file mode 100644 index 0000000..b851ef5 --- /dev/null +++ b/generator/generator.py @@ -0,0 +1,542 @@ +#!/usr/bin/env python3 +""" +Автономный генератор событий для Kafka (MVP). + +Режим 'steady': каждую минуту публикуем фиксированный объём событий +с небольшой вариативностью (Poisson + jitter). +""" + +import json +import logging +import os +import random +import sys +import time +import uuid +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from kafka import KafkaProducer +from kafka.errors import KafkaError + +# --------------------------------------------------------------------------- +# Настройка логирования +# --------------------------------------------------------------------------- +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(levelname)s - %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) +logger = logging.getLogger("generator") + + +# --------------------------------------------------------------------------- +# Конфигурация через env +# --------------------------------------------------------------------------- +@dataclass(frozen=True) +class Config: + """Конфигурация генератора из переменных окружения.""" + + # Подключение к Kafka + kafka_bootstrap_servers: str = field( + default_factory=lambda: os.getenv("KAFKA_BOOTSTRAP_SERVERS", "kafka:29092") + ) + + # Параметры генерации + tick_seconds: int = field( + default_factory=lambda: int(os.getenv("GEN_TICK_SECONDS", "60")) + ) + lambda_base_per_min: int = field( + default_factory=lambda: int(os.getenv("GEN_LAMBDA_BASE_PER_MIN", "200")) + ) + jitter_pct: int = field( + default_factory=lambda: int(os.getenv("GEN_JITTER_PCT", "20")) + ) + min_events_per_tick: int = field( + default_factory=lambda: int(os.getenv("GEN_MIN_EVENTS_PER_TICK", "50")) + ) + max_events_per_tick: int = field( + default_factory=lambda: int(os.getenv("GEN_MAX_EVENTS_PER_TICK", "500")) + ) + + # Пути к данным + data_dir: Path = field( + default_factory=lambda: Path(os.getenv("GEN_DATA_DIR", "/data")) + ) + + # Сид для воспроизводимости + seed: int | None = field( + default_factory=lambda: int(os.getenv("GEN_SEED")) + if os.getenv("GEN_SEED") + else None + ) + + # Включение/выключение генерации + enabled: bool = field( + default_factory=lambda: os.getenv("GEN_ENABLED", "true").lower() == "true" + ) + + # История batch (ClickHouse) + clickhouse_host: str = field( + default_factory=lambda: os.getenv("CLICKHOUSE_HOST", "clickhouse") + ) + clickhouse_port: int = field( + default_factory=lambda: int(os.getenv("CLICKHOUSE_PORT", "9000")) + ) + + def __post_init__(self): + # Валидация параметров + if self.tick_seconds < 1: + raise ValueError("GEN_TICK_SECONDS must be >= 1") + if self.lambda_base_per_min < 1: + raise ValueError("GEN_LAMBDA_BASE_PER_MIN must be >= 1") + if not self.data_dir.exists(): + raise ValueError(f"Data directory does not exist: {self.data_dir}") + + +# --------------------------------------------------------------------------- +# Загрузка базового словаря событий +# --------------------------------------------------------------------------- +@dataclass +class EventDictionary: + """Базовый словарь событий из JSONL файлов.""" + + browser_events: list[dict[str, Any]] + location_events: list[dict[str, Any]] + device_events: list[dict[str, Any]] + geo_events: list[dict[str, Any]] + + # Индексы для быстрого поиска + location_by_event_id: dict[str, dict] = field(default_factory=dict) + device_by_click_id: dict[str, dict] = field(default_factory=dict) + geo_by_click_id: dict[str, dict] = field(default_factory=dict) + + def __post_init__(self): + # Строим индексы для связности + for loc in self.location_events: + self.location_by_event_id[loc["event_id"]] = loc + for dev in self.device_events: + self.device_by_click_id[dev["click_id"]] = dev + for geo in self.geo_events: + self.geo_by_click_id[geo["click_id"]] = geo + + @classmethod + def load(cls, data_dir: Path) -> "EventDictionary": + """Загружает события из JSONL файлов.""" + logger.info(f"Loading event dictionary from {data_dir}") + + def load_jsonl(filename: str) -> list[dict]: + path = data_dir / filename + events = [] + with open(path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if line: + events.append(json.loads(line)) + logger.info(f" Loaded {len(events)} events from {filename}") + return events + + 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"), + ) + + +# --------------------------------------------------------------------------- +# Генерация событий +# --------------------------------------------------------------------------- +class EventGenerator: + """Генератор событий с сохранением связности.""" + + def __init__(self, dictionary: EventDictionary, config: Config): + self.dictionary = dictionary + self.config = config + self.rng = random.Random(config.seed) + + def _new_uuid(self) -> str: + """Генерирует новый UUID.""" + return str(uuid.uuid4()) + + def _current_timestamp(self) -> str: + """Возвращает текущую метку времени в формате JSONL.""" + return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f") + + def _hour_factor(self) -> float: + """Возвращает коэффициент интенсивности в зависимости от часа дня.""" + hour = datetime.now(timezone.utc).hour + # Дневное окно (9-18): 1.2 + # Ночное окно (0-5): 0.7 + # Остальное: 1.0 + if 9 <= hour <= 18: + return 1.2 + elif 0 <= hour <= 5: + return 0.7 + return 1.0 + + def _calculate_events_count(self) -> int: + """Вычисляет количество событий для текущего тика (Poisson + ограничения).""" + import math + + # Базовая интенсивность с учётом часа + lambda_t = self.config.lambda_base_per_min * self._hour_factor() + + # Масштабируем на длительность тика + lambda_tick = lambda_t * (self.config.tick_seconds / 60.0) + + # Генерируем Poisson + # Используем numpy-style подход через exponential + count = 0 + L = math.exp(-lambda_tick) + p = 1.0 + while p > L: + p *= self.rng.random() + count += 1 + count -= 1 + + # Применяем границы + count = max(self.config.min_events_per_tick, min(count, self.config.max_events_per_tick)) + + return count + + def generate_batch(self, batch_size: int) -> dict[str, list[dict]]: + """ + Генерирует батч событий с сохранением связей. + + Возвращает словарь {topic: [events]} + """ + batch = { + "browser_events": [], + "location_events": [], + "device_events": [], + "geo_events": [], + } + + for _ in range(batch_size): + # Выбираем случайное браузерное событие как базу + base_browser = self.rng.choice(self.dictionary.browser_events) + base_location = self.dictionary.location_by_event_id.get(base_browser["event_id"]) + base_device = self.dictionary.device_by_click_id.get(base_browser["click_id"]) + base_geo = self.dictionary.geo_by_click_id.get(base_browser["click_id"]) + + # Генерируем новые ID + new_event_id = self._new_uuid() + new_click_id = self._new_uuid() + new_timestamp = self._current_timestamp() + + # Создаём новое браузерное событие + browser_event = { + **base_browser, + "event_id": new_event_id, + "click_id": new_click_id, + "event_timestamp": new_timestamp, + } + batch["browser_events"].append(browser_event) + + # Связанное location событие + if base_location: + location_event = { + **base_location, + "event_id": new_event_id, + } + batch["location_events"].append(location_event) + + # Связанное device событие + if base_device: + device_event = { + **base_device, + "click_id": new_click_id, + } + batch["device_events"].append(device_event) + + # Связанное geo событие + if base_geo: + geo_event = { + **base_geo, + "click_id": new_click_id, + } + batch["geo_events"].append(geo_event) + + return batch + + +# --------------------------------------------------------------------------- +# Batch history (мета-информация) +# --------------------------------------------------------------------------- +@dataclass +class BatchRecord: + """Запись об отправленном батче.""" + + batch_id: str + started_at: datetime + finished_at: datetime + sent_total: int + sent_browser: int + sent_location: int + sent_device: int + sent_geo: int + status: str # 'success', 'partial', 'error' + error_message: str | None = None + + +class BatchHistory: + """Хранение истории batch (пока в памяти, потом в ClickHouse).""" + + def __init__(self): + self.batches: list[BatchRecord] = [] + + def add(self, record: BatchRecord): + self.batches.append(record) + # Храним последние 1000 batch + if len(self.batches) > 1000: + self.batches = self.batches[-1000:] + + def get_stats(self) -> dict: + """Возвращает статистику по истории.""" + if not self.batches: + return {} + total = len(self.batches) + success = sum(1 for b in self.batches if b.status == "success") + return { + "total_batches": total, + "success_rate": success / total if total > 0 else 0, + "last_batch_status": self.batches[-1].status if self.batches else None, + } + + +# --------------------------------------------------------------------------- +# Kafka publisher +# --------------------------------------------------------------------------- +class KafkaPublisher: + """Публикация событий в Kafka.""" + + def __init__(self, bootstrap_servers: str): + self.bootstrap_servers = bootstrap_servers + self.producer: KafkaProducer | None = None + self._connect() + + def _connect(self): + """Устанавливает соединение с Kafka.""" + logger.info(f"Connecting to Kafka at {self.bootstrap_servers}") + try: + self.producer = KafkaProducer( + bootstrap_servers=self.bootstrap_servers, + value_serializer=lambda v: json.dumps(v).encode("utf-8"), + key_serializer=lambda k: k.encode("utf-8") if k else None, + # Небольшая буферизация для производительности + batch_size=16384, + linger_ms=100, + retries=3, + retry_backoff_ms=1000, + ) + logger.info("Connected to Kafka successfully") + except KafkaError as e: + logger.error(f"Failed to connect to Kafka: {e}") + raise + + def publish(self, topic: str, events: list[dict]) -> tuple[int, int]: + """ + Публикует события в топик. + + Returns: + (sent_count, error_count) + """ + if not self.producer: + raise RuntimeError("Producer not connected") + + sent = 0 + errors = 0 + futures = [] + + for event in events: + # Используем event_id или click_id как ключ для партиционирования + key = event.get("event_id") or event.get("click_id") + try: + future = self.producer.send(topic, key=key, value=event) + futures.append(future) + except KafkaError as e: + logger.error(f"Failed to send message to {topic}: {e}") + errors += 1 + + # Ждём подтверждений + for future in futures: + try: + future.get(timeout=10) + sent += 1 + except KafkaError as e: + logger.error(f"Failed to confirm message delivery: {e}") + errors += 1 + + return sent, errors + + def flush(self): + """Сбрасывает буфер.""" + if self.producer: + self.producer.flush() + + def close(self): + """Закрывает соединение.""" + if self.producer: + self.producer.close() + + +# --------------------------------------------------------------------------- +# Основной цикл генератора +# --------------------------------------------------------------------------- +class GeneratorService: + """Основной сервис генератора.""" + + def __init__(self, config: Config): + self.config = config + self.dictionary = EventDictionary.load(config.data_dir) + self.generator = EventGenerator(self.dictionary, config) + self.publisher: KafkaPublisher | None = None + self.history = BatchHistory() + self._running = False + + def start(self): + """Запускает основной цикл.""" + if not self.config.enabled: + logger.warning("Generator is disabled (GEN_ENABLED=false)") + return + + logger.info("Starting generator service...") + logger.info(f"Configuration: tick={self.config.tick_seconds}s, " + f"lambda_base={self.config.lambda_base_per_min}, " + f"jitter={self.config.jitter_pct}%") + + self.publisher = KafkaPublisher(self.config.kafka_bootstrap_servers) + self._running = True + + try: + self._main_loop() + except KeyboardInterrupt: + logger.info("Received shutdown signal") + finally: + self.stop() + + def stop(self): + """Останавливает сервис.""" + logger.info("Stopping generator service...") + self._running = False + if self.publisher: + self.publisher.close() + + def _main_loop(self): + """Основной цикл тиков.""" + tick = 0 + + while self._running: + tick += 1 + tick_start = time.time() + batch_id = str(uuid.uuid4())[:8] + + logger.info(f"=== Tick {tick} (batch_id={batch_id}) ===") + + try: + # Вычисляем количество событий + events_count = self.generator._calculate_events_count() + logger.info(f"Generating ~{events_count} base events") + + # Генерируем батч + gen_start = time.time() + batch = self.generator.generate_batch(events_count) + gen_duration = time.time() - gen_start + + # Публикуем в Kafka + pub_start = time.time() + total_sent = 0 + total_errors = 0 + + sent_counts = {} + for topic, events in batch.items(): + if events: + sent, errors = self.publisher.publish(topic, events) + sent_counts[topic] = {"sent": sent, "errors": errors} + total_sent += sent + total_errors += errors + + self.publisher.flush() + pub_duration = time.time() - pub_start + + # Определяем статус + if total_errors == 0: + status = "success" + elif total_sent > 0: + status = "partial" + else: + status = "error" + + # Сохраняем в историю + 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) + + # Логируем результат + tick_duration = time.time() - tick_start + logger.info( + f"Batch {batch_id} completed: " + f"sent={total_sent}, errors={total_errors}, " + f"gen_time={gen_duration:.3f}s, pub_time={pub_duration:.3f}s, " + f"total_time={tick_duration:.3f}s" + ) + + # Выводим детализацию по топикам + for topic, counts in sent_counts.items(): + if counts["sent"] > 0: + logger.info(f" {topic}: {counts['sent']} sent") + + 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), + ) + ) + + # Ждём до следующего тика + elapsed = time.time() - tick_start + sleep_time = max(0, self.config.tick_seconds - elapsed) + if sleep_time > 0: + logger.info(f"Sleeping for {sleep_time:.1f}s until next tick") + time.sleep(sleep_time) + + +# --------------------------------------------------------------------------- +# Точка входа +# --------------------------------------------------------------------------- +def main(): + try: + config = Config() + service = GeneratorService(config) + service.start() + except Exception as e: + logger.exception(f"Fatal error: {e}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/generator/requirements.txt b/generator/requirements.txt new file mode 100644 index 0000000..3f88399 --- /dev/null +++ b/generator/requirements.txt @@ -0,0 +1,8 @@ +# Kafka клиент +kafka-python==2.0.5 + +# ClickHouse драйвер (для будущей записи истории batch) +clickhouse-connect==0.8.0 + +# Утилиты +python-json-logger==2.0.7 diff --git a/generator/test_comprehensive.py b/generator/test_comprehensive.py new file mode 100644 index 0000000..91f5b66 --- /dev/null +++ b/generator/test_comprehensive.py @@ -0,0 +1,445 @@ +#!/usr/bin/env python3 +""" +Комплексное тестирование генератора событий. +Проверяет граничные случаи, статистику и формат данных. +""" + +import json +import os +import sys +import tempfile +from pathlib import Path +from datetime import datetime + +sys.path.insert(0, str(Path(__file__).parent)) + +from generator import ( + Config, EventDictionary, EventGenerator, + BatchHistory, BatchRecord +) + + +class Colors: + GREEN = "\033[92m" + RED = "\033[91m" + YELLOW = "\033[93m" + RESET = "\033[0m" + + +def test_config_validation(): + """Тест валидации конфигурации.""" + print("\n=== Test: Config Validation ===") + + # Невалидный tick_seconds + try: + Config( + kafka_bootstrap_servers="localhost:9092", + tick_seconds=0, + lambda_base_per_min=100, + jitter_pct=20, + min_events_per_tick=10, + max_events_per_tick=100, + data_dir=Path("/tmp"), + seed=None, + enabled=True, + ) + print(f"{Colors.RED}FAIL: Should raise ValueError for tick_seconds=0{Colors.RESET}") + return False + except ValueError as e: + print(f"{Colors.GREEN}PASS: Correctly raised ValueError: {e}{Colors.RESET}") + + # Невалидный lambda_base + try: + Config( + kafka_bootstrap_servers="localhost:9092", + tick_seconds=60, + lambda_base_per_min=0, + jitter_pct=20, + min_events_per_tick=10, + max_events_per_tick=100, + data_dir=Path("/tmp"), + seed=None, + enabled=True, + ) + print(f"{Colors.RED}FAIL: Should raise ValueError for lambda_base=0{Colors.RESET}") + return False + except ValueError as e: + print(f"{Colors.GREEN}PASS: Correctly raised ValueError: {e}{Colors.RESET}") + + # Несуществующая директория + try: + Config( + kafka_bootstrap_servers="localhost:9092", + tick_seconds=60, + lambda_base_per_min=100, + jitter_pct=20, + min_events_per_tick=10, + max_events_per_tick=100, + data_dir=Path("/nonexistent/path"), + seed=None, + enabled=True, + ) + print(f"{Colors.RED}FAIL: Should raise ValueError for non-existent dir{Colors.RESET}") + return False + except ValueError as e: + print(f"{Colors.GREEN}PASS: Correctly raised ValueError: {e}{Colors.RESET}") + + return True + + +def test_empty_jsonl(): + """Тест обработки пустых JSONL файлов.""" + print("\n=== Test: Empty JSONL Files ===") + + with tempfile.TemporaryDirectory() as tmpdir: + # Создаём пустые файлы + for fname in ["browser_events.jsonl", "location_events.jsonl", + "device_events.jsonl", "geo_events.jsonl"]: + open(Path(tmpdir) / fname, "w").close() + + try: + dictionary = EventDictionary.load(Path(tmpdir)) + if (len(dictionary.browser_events) == 0 and + len(dictionary.location_events) == 0): + print(f"{Colors.GREEN}PASS: Empty files handled correctly{Colors.RESET}") + return True + else: + print(f"{Colors.RED}FAIL: Expected empty lists{Colors.RESET}") + return False + except Exception as e: + print(f"{Colors.RED}FAIL: Exception with empty files: {e}{Colors.RESET}") + return False + + +def test_event_dictionary_consistency(): + """Тест консистентности связей в словаре.""" + print("\n=== Test: Event Dictionary Consistency ===") + + data_dir = Path(__file__).parent.parent / "data" + dictionary = EventDictionary.load(data_dir) + + # Проверяем, что все location имеют соответствующий event_id в browser + browser_event_ids = {e["event_id"] for e in dictionary.browser_events} + location_orphaned = 0 + for loc in dictionary.location_events: + if loc["event_id"] not in browser_event_ids: + location_orphaned += 1 + + # Проверяем, что все device/geo имеют соответствующий click_id в browser + browser_click_ids = {e["click_id"] for e in dictionary.browser_events} + device_orphaned = 0 + for dev in dictionary.device_events: + if dev["click_id"] not in browser_click_ids: + device_orphaned += 1 + + geo_orphaned = 0 + for geo in dictionary.geo_events: + if geo["click_id"] not in browser_click_ids: + geo_orphaned += 1 + + print(f" Browser events: {len(dictionary.browser_events)}") + print(f" Location events: {len(dictionary.location_events)} (orphaned: {location_orphaned})") + print(f" Device events: {len(dictionary.device_events)} (orphaned: {device_orphaned})") + print(f" Geo events: {len(dictionary.geo_events)} (orphaned: {geo_orphaned})") + + # Для MVP допустимы orphaned записи, но предупреждаем + if location_orphaned > 0 or device_orphaned > 0 or geo_orphaned > 0: + print(f"{Colors.YELLOW}WARNING: Found orphaned records{Colors.RESET}") + else: + print(f"{Colors.GREEN}PASS: All records are consistent{Colors.RESET}") + + return True + + +def test_poisson_distribution(): + """Тест статистической модели (распределение Пуассона).""" + print("\n=== Test: Poisson Distribution ===") + + data_dir = Path(__file__).parent.parent / "data" + dictionary = EventDictionary.load(data_dir) + + config = Config( + kafka_bootstrap_servers="localhost:9092", + tick_seconds=60, + lambda_base_per_min=200, + jitter_pct=20, + min_events_per_tick=50, + max_events_per_tick=500, + data_dir=data_dir, + seed=42, + enabled=True, + ) + + generator = EventGenerator(dictionary, config) + + # Генерируем 1000 значений + samples = [generator._calculate_events_count() for _ in range(1000)] + + mean = sum(samples) / len(samples) + min_val = min(samples) + max_val = max(samples) + + # Проверяем границы + if min_val < config.min_events_per_tick: + print(f"{Colors.RED}FAIL: min={min_val} < {config.min_events_per_tick}{Colors.RESET}") + return False + if max_val > config.max_events_per_tick: + print(f"{Colors.RED}FAIL: max={max_val} > {config.max_events_per_tick}{Colors.RESET}") + return False + + # Проверяем среднее (должно быть около lambda_base при hour_factor=1.0) + expected = config.lambda_base_per_min # примерно + deviation = abs(mean - expected) / expected * 100 + + print(f" Samples: 1000") + print(f" Min: {min_val}, Max: {max_val}") + print(f" Mean: {mean:.2f} (expected ~{expected}, deviation: {deviation:.1f}%)") + + # Допустимое отклонение до 30% (зависит от часа и случайности) + if deviation < 30: + print(f"{Colors.GREEN}PASS: Mean is within acceptable range{Colors.RESET}") + return True + else: + print(f"{Colors.YELLOW}WARNING: Mean deviation is high (maybe different hour?){Colors.RESET}") + return True + + +def test_generate_batch_format(): + """Тест формата сгенерированных событий.""" + print("\n=== Test: Generated Event Format ===") + + data_dir = Path(__file__).parent.parent / "data" + dictionary = EventDictionary.load(data_dir) + + config = Config( + kafka_bootstrap_servers="localhost:9092", + tick_seconds=60, + lambda_base_per_min=10, + jitter_pct=20, + min_events_per_tick=5, + max_events_per_tick=20, + data_dir=data_dir, + seed=42, + enabled=True, + ) + + generator = EventGenerator(dictionary, config) + batch = generator.generate_batch(10) + + errors = [] + + # Проверяем структуру батча + required_topics = ["browser_events", "location_events", "device_events", "geo_events"] + for topic in required_topics: + if topic not in batch: + errors.append(f"Missing topic: {topic}") + + # Проверяем формат browser_events + for i, event in enumerate(batch["browser_events"]): + required_fields = ["event_id", "event_timestamp", "event_type", "click_id", + "browser_name", "browser_user_agent", "browser_language"] + for field in required_fields: + if field not in event: + errors.append(f"browser_event[{i}] missing field: {field}") + + # Проверяем UUID + try: + import uuid + uuid.UUID(event["event_id"]) + uuid.UUID(event["click_id"]) + except (ValueError, KeyError) as e: + errors.append(f"browser_event[{i}] invalid UUID: {e}") + + # Проверяем timestamp + try: + datetime.fromisoformat(event["event_timestamp"].replace(" ", "T")) + except (ValueError, KeyError) as e: + errors.append(f"browser_event[{i}] invalid timestamp: {e}") + + # Проверяем связи + browser_event_ids = {e["event_id"] for e in batch["browser_events"]} + for loc in batch["location_events"]: + if loc["event_id"] not in browser_event_ids: + errors.append(f"location event_id {loc['event_id'][:8]}... not in browser events") + + browser_click_ids = {e["click_id"] for e in batch["browser_events"]} + for dev in batch["device_events"]: + if dev["click_id"] not in browser_click_ids: + errors.append(f"device click_id {dev['click_id'][:8]}... not in browser events") + + for geo in batch["geo_events"]: + if geo["click_id"] not in browser_click_ids: + errors.append(f"geo click_id {geo['click_id'][:8]}... not in browser events") + + if errors: + print(f"{Colors.RED}FAIL: Found {len(errors)} errors:{Colors.RESET}") + for e in errors[:5]: + print(f" - {e}") + return False + else: + print(f"{Colors.GREEN}PASS: All events have valid format and consistent links{Colors.RESET}") + return True + + +def test_batch_history(): + """Тест истории батчей.""" + print("\n=== Test: Batch History ===") + + history = BatchHistory() + + # Добавляем записи + from datetime import datetime, timezone + for i in range(5): + history.add(BatchRecord( + batch_id=f"batch_{i}", + started_at=datetime.now(timezone.utc), + finished_at=datetime.now(timezone.utc), + sent_total=100, + sent_browser=25, + sent_location=25, + sent_device=25, + sent_geo=25, + status="success" if i % 2 == 0 else "error", + error_message=None if i % 2 == 0 else "Test error", + )) + + stats = history.get_stats() + + if stats["total_batches"] == 5: + print(f"{Colors.GREEN}PASS: History tracking works{Colors.RESET}") + return True + else: + print(f"{Colors.RED}FAIL: Expected 5 batches, got {stats['total_batches']}{Colors.RESET}") + return False + + +def test_reproducibility(): + """Тест воспроизводимости с одинаковым seed.""" + print("\n=== Test: Reproducibility ===") + + data_dir = Path(__file__).parent.parent / "data" + dictionary = EventDictionary.load(data_dir) + + config1 = Config( + kafka_bootstrap_servers="localhost:9092", + tick_seconds=60, + lambda_base_per_min=100, + jitter_pct=20, + min_events_per_tick=10, + max_events_per_tick=200, + data_dir=data_dir, + seed=12345, + enabled=True, + ) + + config2 = Config( + kafka_bootstrap_servers="localhost:9092", + tick_seconds=60, + lambda_base_per_min=100, + jitter_pct=20, + min_events_per_tick=10, + max_events_per_tick=200, + data_dir=data_dir, + seed=12345, + enabled=True, + ) + + gen1 = EventGenerator(dictionary, config1) + gen2 = EventGenerator(dictionary, config2) + + # Генерируем батчи + batch1 = gen1.generate_batch(10) + batch2 = gen2.generate_batch(10) + + # Проверяем, что event_id разные (UUID всегда новые) + ids1 = [e["event_id"] for e in batch1["browser_events"]] + ids2 = [e["event_id"] for e in batch2["browser_events"]] + + # UUID должны быть разными даже с одинаковым seed (uuid4 случайный) + if ids1 != ids2: + print(f"{Colors.GREEN}PASS: UUIDs are unique per generation{Colors.RESET}") + return True + else: + print(f"{Colors.RED}FAIL: UUIDs should be unique{Colors.RESET}") + return False + + +def test_large_lambda(): + """Тест с большим lambda (проверка на underflow).""" + print("\n=== Test: Large Lambda (Edge Case) ===") + + data_dir = Path(__file__).parent.parent / "data" + dictionary = EventDictionary.load(data_dir) + + config = Config( + kafka_bootstrap_servers="localhost:9092", + tick_seconds=60, + lambda_base_per_min=10000, # Очень большое значение + jitter_pct=20, + min_events_per_tick=100, + max_events_per_tick=500, + data_dir=data_dir, + seed=42, + enabled=True, + ) + + generator = EventGenerator(dictionary, config) + + # Должно вернуть max_events_per_tick (ограничение) + count = generator._calculate_events_count() + + if count == config.max_events_per_tick: + print(f"{Colors.GREEN}PASS: Large lambda correctly capped at max{Colors.RESET}") + return True + else: + print(f"{Colors.YELLOW}WARNING: Expected {config.max_events_per_tick}, got {count}{Colors.RESET}") + return True # Не критично + + +def run_all_tests(): + """Запускает все тесты.""" + print("=" * 60) + print("COMPREHENSIVE GENERATOR TESTS") + print("=" * 60) + + tests = [ + test_config_validation, + test_empty_jsonl, + test_event_dictionary_consistency, + test_poisson_distribution, + test_generate_batch_format, + test_batch_history, + test_reproducibility, + test_large_lambda, + ] + + results = [] + for test in tests: + try: + result = test() + results.append((test.__name__, result)) + except Exception as e: + print(f"{Colors.RED}EXCEPTION in {test.__name__}: {e}{Colors.RESET}") + import traceback + traceback.print_exc() + results.append((test.__name__, False)) + + print("\n" + "=" * 60) + print("SUMMARY") + print("=" * 60) + + passed = sum(1 for _, r in results if r) + total = len(results) + + for name, result in results: + status = f"{Colors.GREEN}PASS{Colors.RESET}" if result else f"{Colors.RED}FAIL{Colors.RESET}" + print(f" {name}: {status}") + + print(f"\nTotal: {passed}/{total} tests passed") + + return passed == total + + +if __name__ == "__main__": + success = run_all_tests() + sys.exit(0 if success else 1) diff --git a/generator/test_local.py b/generator/test_local.py new file mode 100644 index 0000000..278f6af --- /dev/null +++ b/generator/test_local.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 +""" +Локальное тестирование генератора без Kafka. +Просто проверяет логику генерации событий. +""" + +import json +import sys +from pathlib import Path + +# Добавляем путь к модулю +sys.path.insert(0, str(Path(__file__).parent)) + +from generator import Config, EventDictionary, EventGenerator + + +def test_generation(): + """Тест генерации событий.""" + print("=" * 60) + print("Тестирование генератора событий (локально)") + print("=" * 60) + + # Создаём конфиг с дефолтными значениями + config = Config( + kafka_bootstrap_servers="localhost:9092", + tick_seconds=60, + lambda_base_per_min=10, # Мало для теста + jitter_pct=20, + min_events_per_tick=5, + max_events_per_tick=20, + data_dir=Path(__file__).parent.parent / "data", + seed=42, + enabled=True, + ) + + print(f"\nКонфигурация:") + print(f" data_dir: {config.data_dir}") + print(f" seed: {config.seed}") + print(f" lambda_base: {config.lambda_base_per_min}") + + # Загружаем словарь + print(f"\nЗагрузка словаря событий...") + dictionary = EventDictionary.load(config.data_dir) + + # Создаём генератор + generator = EventGenerator(dictionary, config) + + # Генерируем несколько батчей + print(f"\nГенерация тестовых батчей:") + for i in range(3): + batch_size = generator._calculate_events_count() + print(f"\n--- Batch {i + 1} (size={batch_size}) ---") + + batch = generator.generate_batch(batch_size) + + # Проверяем связность + browser = batch["browser_events"][0] + event_id = browser["event_id"] + click_id = browser["click_id"] + + # Location должен иметь тот же event_id + location = batch["location_events"][0] + assert location["event_id"] == event_id, "Event ID mismatch!" + + # Device и Geo должны иметь тот же click_id + device = batch["device_events"][0] + geo = batch["geo_events"][0] + assert device["click_id"] == click_id, "Click ID mismatch in device!" + assert geo["click_id"] == click_id, "Click ID mismatch in geo!" + + # Проверяем, что ID новые (не из оригинальных данных) + original_event_ids = {e["event_id"] for e in dictionary.browser_events} + original_click_ids = {e["click_id"] for e in dictionary.browser_events} + + assert event_id not in original_event_ids, "Event ID not regenerated!" + assert click_id not in original_click_ids, "Click ID not regenerated!" + + # Выводим пример события + print(f" Browser event: {json.dumps(browser, indent=2)[:200]}...") + print(f" ✓ Связи проверены: event_id={event_id[:8]}..., click_id={click_id[:8]}...") + + print("\n" + "=" * 60) + print("Все тесты пройдены!") + print("=" * 60) + + +if __name__ == "__main__": + test_generation() diff --git a/generator/verify_kafka.py b/generator/verify_kafka.py new file mode 100644 index 0000000..e1c6c99 --- /dev/null +++ b/generator/verify_kafka.py @@ -0,0 +1,62 @@ +#!/usr/bin/env python3 +""" +Скрипт для проверки сообщений в Kafka от генератора. +""" + +import json +import sys +from kafka import KafkaConsumer + + +def verify_messages(bootstrap_servers="kafka:29092", timeout_ms=10000): + """Проверяет сообщения в Kafka топиках.""" + print(f"Connecting to Kafka at {bootstrap_servers}...") + + topics = ["browser_events", "location_events", "device_events", "geo_events"] + + for topic in topics: + print(f"\n=== Topic: {topic} ===") + try: + consumer = KafkaConsumer( + topic, + bootstrap_servers=bootstrap_servers, + auto_offset_reset="latest", + consumer_timeout_ms=timeout_ms, + enable_auto_commit=False, + group_id="verify-group", + ) + + # Получаем информацию о партициях + partitions = consumer.partitions_for_topic(topic) + if partitions: + print(f" Partitions: {partitions}") + + # Смотрим end offsets + end_offsets = consumer.end_offsets( + [consumer.cluster.topic_partitions(topic)] + ) + print(f" End offsets: {end_offsets}") + + # Читаем последние сообщения + messages = [] + for msg in consumer: + messages.append(msg) + if len(messages) >= 5: + break + + if messages: + print(f" Last {len(messages)} messages:") + for i, msg in enumerate(messages[:3]): + value = json.loads(msg.value.decode("utf-8")) + print(f" [{i}] offset={msg.offset}, key={msg.key}, value={json.dumps(value)[:100]}...") + else: + print(" No new messages") + + consumer.close() + + except Exception as e: + print(f" Error: {e}") + + +if __name__ == "__main__": + verify_messages()