diff --git a/.env.example b/.env.example index 1a35c39..6469fa9 100644 --- a/.env.example +++ b/.env.example @@ -7,6 +7,7 @@ CLICKHOUSE_02_TCP_PORT=29001 CLICKHOUSE_CLUSTER_SECRET=cluster-26-secret CLICKHOUSE_DEFAULT_PASSWORD=default-26-local CLICKHOUSE_ETL_PASSWORD=etl-26-writer +CLICKHOUSE_LIFECYCLE_PASSWORD=lifecycle-26-owner CLICKHOUSE_BI_PASSWORD=bi-26-reader CLICKHOUSE_ANALYST_PASSWORD=analyst-26-reader CLICKHOUSE_GRAFANA_PASSWORD=grafana-26-reader diff --git a/Makefile b/Makefile index f340831..6fe145f 100644 --- a/Makefile +++ b/Makefile @@ -4,23 +4,27 @@ GENERATOR_LIMIT ?= GENERATOR_SPEED ?= # Цели корня — про стенд; проверки генератора — в `generator/Makefile`. -.PHONY: up down clean ps logs smoke check-clickhouse check-services config-test lint generate-batch generate-live +.PHONY: up down clean rebuild-storage ps logs smoke check-clickhouse check-services config-test lint generate-batch generate-live # --- Жизнь стенда --- -# Второй шаг: `--wait` дожидается служб, а приём событий асинхронный — довод -# целиком в шапке скрипта. +# Compose дожидается инфраструктуры, Airflow — готового прикладного мира. up: - $(COMPOSE) up --detach --build --wait --wait-timeout 600 - COMPOSE_BIN="$(COMPOSE)" ./scripts/wait-for-world.sh - COMPOSE_BIN="$(COMPOSE)" ./scripts/ingest-starting-orders.sh + $(COMPOSE) up --detach --build --wait --wait-timeout 600 --remove-orphans + COMPOSE_BIN="$(COMPOSE)" ./scripts/run-dag.sh world_initialize down: $(COMPOSE) down --remove-orphans +# Полное удаление стенда, а не способ получить чистые данные перед проверкой. +# Граница операции и требование к обоснованию — в AGENTS.md. clean: $(COMPOSE) down --volumes --remove-orphans +# Возврат прикладного мира к началу без удаления инфраструктурных томов. +rebuild-storage: + COMPOSE_BIN="$(COMPOSE)" ./scripts/run-dag.sh world_recreate + ps: $(COMPOSE) ps diff --git a/compose.yaml b/compose.yaml index 3abf49d..0f58f01 100644 --- a/compose.yaml +++ b/compose.yaml @@ -17,6 +17,7 @@ x-clickhouse-common: &clickhouse-common CLICKHOUSE_CLUSTER_SECRET: ${CLICKHOUSE_CLUSTER_SECRET:?Скопируйте .env.example в .env} CLICKHOUSE_DEFAULT_PASSWORD: ${CLICKHOUSE_DEFAULT_PASSWORD:?Скопируйте .env.example в .env} CLICKHOUSE_ETL_PASSWORD: ${CLICKHOUSE_ETL_PASSWORD:?Скопируйте .env.example в .env} + CLICKHOUSE_LIFECYCLE_PASSWORD: ${CLICKHOUSE_LIFECYCLE_PASSWORD:?Скопируйте .env.example в .env} CLICKHOUSE_BI_PASSWORD: ${CLICKHOUSE_BI_PASSWORD:?Скопируйте .env.example в .env} CLICKHOUSE_ANALYST_PASSWORD: ${CLICKHOUSE_ANALYST_PASSWORD:?Скопируйте .env.example в .env} CLICKHOUSE_GRAFANA_PASSWORD: ${CLICKHOUSE_GRAFANA_PASSWORD:?Скопируйте .env.example в .env} @@ -33,9 +34,8 @@ x-clickhouse-common: &clickhouse-common retries: 30 start_period: 10s -# Факты стенда, у которых стало по два потребителя: разовая служба генератора и -# пульт мира. Названы по одному разу — иначе однажды разойдутся, и заметит это -# не проверка, а менти с пустым топиком. +# Факты генератора и пульта мира названы по одному разу — иначе они однажды +# разойдутся, и заметит это не проверка, а менти с пустым топиком. x-generator-image: &generator-image clickstream-generator:local x-orders-topic: &orders-topic orders @@ -73,9 +73,9 @@ x-airflow-common: &airflow-common AIRFLOW_ADMIN_USER: ${AIRFLOW_ADMIN_USER:?Скопируйте .env.example в .env} AIRFLOW_ADMIN_PASSWORD: ${AIRFLOW_ADMIN_PASSWORD:?Скопируйте .env.example в .env} CLICKHOUSE_ETL_PASSWORD: ${CLICKHOUSE_ETL_PASSWORD:?Скопируйте .env.example в .env} - # Пульт мира поднимает генератор сам, отдельным контейнером, — значит те же - # факты стенда, что compose даёт разовой службе генератора, нужны и дагам. - # Имя сети собирается из имени проекта: у второй копии стенда оно другое. + CLICKHOUSE_LIFECYCLE_PASSWORD: ${CLICKHOUSE_LIFECYCLE_PASSWORD:?Скопируйте .env.example в .env} + # Пульт мира поднимает генератор отдельным контейнером. Имя сети собирается + # из имени проекта: у второй копии стенда оно другое. <<: *kafka-target GENERATOR_IMAGE: *generator-image STAND_NETWORK: ${COMPOSE_PROJECT_NAME}_default @@ -86,6 +86,11 @@ x-airflow-common: &airflow-common # Запросы дагов: даг называет файл, а текст читает Airflow при исполнении # задачи. Каталог тот же, что применяет DDL, — вся SQL стенда живёт в sql/. - ./sql:/opt/airflow/sql:ro + # Инициализация мира собирает канонический образ из того же контекста, что + # ручной generator. Docker SDK архивирует эти файлы внутри задачи. + - ./generator:/opt/airflow/world/generator:ro + - ./data:/opt/airflow/world/data:ro + - ./.dockerignore:/opt/airflow/world/.dockerignore:ro - ./infra/airflow/init.sh:/opt/airflow/init.sh:ro - airflow_logs:/opt/airflow/logs - airflow_auth:/opt/airflow/auth @@ -96,15 +101,6 @@ x-generator-common: &generator-common context: . dockerfile: generator/Dockerfile restart: "no" - depends_on: - kafka-init: - condition: service_completed_successfully - # Читать топик движок Kafka начинает не при своём создании, а когда над ним - # появляется матвью приёма, и она создаётся последней по номеру файла - # (sql/ddl/40-stg-views.sql). Отправь генератор события раньше — они лягут в - # топик и молча минуют хранилище. - clickhouse-init: - condition: service_completed_successfully environment: # Адрес брокера и имя топика — факты стенда, и называет их стенд. # Остальное (день, зерно, число дней, предел пачки) приходит аргументами @@ -228,119 +224,6 @@ services: retries: 15 start_period: 10s - kafka-init: - image: apache/kafka:4.3.1 - restart: "no" - depends_on: - kafka: - condition: service_healthy - entrypoint: ["/bin/bash", "-ec"] - command: - - | - topic_partition_count() { - /opt/kafka/bin/kafka-topics.sh \ - --bootstrap-server kafka:9092 \ - --describe \ - --topic hits | \ - sed -n '1s/.*PartitionCount: \([0-9][0-9]*\).*/\1/p' - } - - /opt/kafka/bin/kafka-topics.sh \ - --bootstrap-server kafka:9092 \ - --create \ - --if-not-exists \ - --topic hits \ - --partitions 2 \ - --replication-factor 1 - - partition_count="$$(topic_partition_count)" - # Брокер разрешает автосоздание, хотя потребитель ClickHouse его не запрашивает. - # На сохранённом томе другой клиент мог уже создать hits с одним разделом. - if [[ "$$partition_count" == "1" ]]; then - /opt/kafka/bin/kafka-topics.sh \ - --bootstrap-server kafka:9092 \ - --alter \ - --topic hits \ - --partitions 2 - partition_count="$$(topic_partition_count)" - fi - if [[ "$$partition_count" != "2" ]]; then - printf 'ОШИБКА: у топика hits разделов: %s, ожидалось: 2.\n' \ - "$${partition_count:-неизвестно}" >&2 - exit 1 - fi - - # Топик заказов: одна партиция, читает его один чтец на ноде 1 — - # довод в ADR 0008. - /opt/kafka/bin/kafka-topics.sh \ - --bootstrap-server kafka:9092 \ - --create \ - --if-not-exists \ - --topic orders \ - --partitions 1 \ - --replication-factor 1 - - # DDL применяется с ноды 1 по порядку имён файлов после готовности всего кластера. - clickhouse-init: - image: clickhouse/clickhouse-server:26.3.17.56 - restart: "no" - depends_on: - kafka-init: - condition: service_completed_successfully - clickhouse-01: - condition: service_healthy - clickhouse-02: - condition: service_healthy - environment: - LC_ALL: C - CLICKHOUSE_ETL_PASSWORD: ${CLICKHOUSE_ETL_PASSWORD:?Скопируйте .env.example в .env} - entrypoint: ["/bin/bash", "-ec"] - command: - - | - set -- /ddl/*.sql - if [[ ! -e "$$1" ]]; then - printf 'ОШИБКА: в /ddl нет файлов SQL.\n' >&2 - exit 1 - fi - for ddl_file do - printf 'Применяем %s.\n' "$${ddl_file##*/}" - clickhouse-client \ - --host clickhouse-01 \ - --user etl \ - --password "$$CLICKHOUSE_ETL_PASSWORD" \ - --multiquery \ - --queries-file "$$ddl_file" - done - volumes: - - ./sql/ddl:/ddl:ro - - # Стартовый мир: восемь модельных дней уезжают в топик при каждом подъёме. - # Что это за дни и каким мир обязан выйти — data/world-inventory.json. - # - # Число дней стоит здесь числом: YAML не читает Python, и одно из двух мест - # (второе — `STARTING_DAYS` в inventory.py) лишнее по построению. Правя одно, - # правьте второе — на страже тут никто не стоит: залей эта служба лишний - # день, он лёг бы за рамкой дат описи и остался бы незамеченным. Внутри YAML - # число одно на всех: то же говорит дагам пульта, где кончается стартовый мир. - # - # Повторный `make up` заливает мир заново, и это не оплошность: `WatchID` у - # событий те же, ReplacingMergeTree схлопнет повтор в ODS. Сырьё в STG при - # этом честно удвоится — свойство слоя, описанное в storage.md. - world-init: - <<: *generator-common - command: ["batch", "--day", "0", "--days", *world-starting-days] - - # Второй источник стартового мира идёт после событий тем же генератором. - # Сдвиг команды snapshot оставляет день 7 первому ходу мира: за восемь - # дней уезжают семь слепков, дней 0…6. - world-snapshots: - <<: *generator-common - depends_on: - world-init: - condition: service_completed_successfully - environment: *orders-kafka-target - command: ["snapshot", "--day", "0", "--days", *world-starting-days] - # Тот же образ для ручных прогонов: `make generate-batch`, `make generate-live`. # Под профилем — чтобы обычный подъём стенда её не трогал. generator: @@ -385,14 +268,6 @@ services: condition: service_healthy clickhouse-01: condition: service_healthy - # Не убирать: без зависимого успешный clickhouse-init считается - # упавшим для --wait. - clickhouse-init: - condition: service_completed_successfully - # По той же причине. Слепки идут за событиями и замыкают цепь разовых - # служб; долгоживущие службы данных её не ждут, а `airflow-init` ждёт. - world-snapshots: - condition: service_completed_successfully entrypoint: ["/bin/bash"] command: ["/opt/airflow/init.sh"] @@ -433,6 +308,9 @@ services: - /var/run/docker.sock:/var/run/docker.sock - ./dags:/opt/airflow/dags:ro - ./sql:/opt/airflow/sql:ro + - ./generator:/opt/airflow/world/generator:ro + - ./data:/opt/airflow/world/data:ro + - ./.dockerignore:/opt/airflow/world/.dockerignore:ro - ./infra/airflow/init.sh:/opt/airflow/init.sh:ro - airflow_logs:/opt/airflow/logs - airflow_auth:/opt/airflow/auth diff --git a/dags/world_control.py b/dags/world_control.py index 0ecad54..47d23a8 100644 --- a/dags/world_control.py +++ b/dags/world_control.py @@ -27,6 +27,7 @@ from __future__ import annotations import datetime import os +from airflow.exceptions import AirflowException from airflow.providers.docker.operators.docker import DockerOperator from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator from airflow.sdk import Param, Variable, dag, get_current_context, task @@ -45,8 +46,8 @@ GENERATOR_ENVIRONMENT = { ORDERS_TOPIC = os.environ["KAFKA_ORDERS_TOPIC"] STARTING_DAYS = int(os.environ["WORLD_STARTING_DAYS"]) -# Позиция на оси: номер первого несыгранного дня. Переменной нет — мир в -# стартовом состоянии, и играть надо сразу за ним. +# Позиция на оси: номер первого несыгранного дня. До конца инициализации +# работники не имеют права угадывать её — состояние объяснено в ADR 0013. # # Позиция ставится, а не увеличивается. Наложись один прогон на другой, худшее # при таком правиле — сыгранный дважды день, а повтор схлопнет @@ -69,7 +70,24 @@ TAGS = ["пульт мира"] @task def first_unplayed_day() -> int: """Номер дня, с которого играть.""" - return int(Variable.get(WORLD_POSITION, default=STARTING_DAYS)) + raw_position = Variable.get(WORLD_POSITION, default=None) + if raw_position is None: + raise AirflowException( + "мир ещё не создан: выполните make up и дождитесь world_initialize" + ) + try: + position = int(raw_position) + except (TypeError, ValueError) as error: + raise AirflowException( + f"world_position={raw_position!r} не является номером дня; " + "выполните make rebuild-storage" + ) from error + if position < STARTING_DAYS: + action = ( + "выполните make up" if position == 0 else "выполните make rebuild-storage" + ) + raise AirflowException(f"мир не готов: world_position={position}; {action}") + return position def _generator(task_id: str, command: list[str]) -> DockerOperator: diff --git a/dags/world_lifecycle.py b/dags/world_lifecycle.py new file mode 100644 index 0000000..16f6d2b --- /dev/null +++ b/dags/world_lifecycle.py @@ -0,0 +1,409 @@ +"""Создание и пересоздание прикладного мира. + +Compose поднимает службы, а этот граф создаёт всё прикладное поверх них: +топики, схему ClickHouse, данные стартовых дней и позицию модельного времени. +`world_initialize` безопасен для обычного `make up`; `world_recreate` — явная +операция обслуживания, которая сначала объявляет прежний мир недействительным. + +Контракт состояний и доводы — ADR 0013 и спека жизненного цикла мира. +""" + +from __future__ import annotations + +import datetime +import json +import logging +import os +import time +from pathlib import Path + +from airflow.exceptions import AirflowException +from airflow.providers.docker.hooks.docker import DockerHook +from airflow.providers.docker.operators.docker import DockerOperator +from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator +from airflow.sdk import Connection, Variable, dag, task + +START_DATE = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) +TAGS = ["жизненный цикл мира"] + +WORLD_POSITION = "world_position" +STARTING_DAYS = int(os.environ["WORLD_STARTING_DAYS"]) +HITS_TOPIC = os.environ["KAFKA_TOPIC"] +ORDERS_TOPIC = os.environ["KAFKA_ORDERS_TOPIC"] +KAFKA_BOOTSTRAP_SERVERS = os.environ["KAFKA_BOOTSTRAP_SERVERS"] +GENERATOR_IMAGE = os.environ["GENERATOR_IMAGE"] +STAND_NETWORK = os.environ["STAND_NETWORK"] + +SQL_ROOT = Path("/opt/airflow/sql/ddl") +WORLD_ROOT = Path("/opt/airflow/world") +INVENTORY = WORLD_ROOT / "data/world-inventory.json" +CLICKHOUSE_CONNECTION = "clickhouse_lifecycle" +CLICKHOUSE_CLUSTER = "clickstream_cluster" +APPLICATION_DATABASES = ("stg", "dm", "dds", "ods", "dic") +CONSUMER_GROUPS = ("clickstream_hits", "clickstream_orders") +TOPICS = {HITS_TOPIC: 2, ORDERS_TOPIC: 1} + + +def _clickhouse_client(): + """Открыть соединение владельца жизненного цикла.""" + import clickhouse_connect + + connection = Connection.get(CLICKHOUSE_CONNECTION) + return clickhouse_connect.get_client( + host=connection.host, + port=connection.port, + username=connection.login, + password=connection.password, + database=connection.schema or "default", + connect_timeout=5, + send_receive_timeout=60, + ) + + +def _ddl_statements(text: str) -> list[str]: + """Разделить канонический DDL по его явному правилу файлов. + + ClickHouse Connect исполняет один запрос за вызов. В `sql/ddl` конец + выражения — точка с запятой в конце строки; это проверяемая конвенция + наших файлов, а не попытка написать общий SQL-парсер. + """ + statements: list[str] = [] + lines: list[str] = [] + for line in text.splitlines(): + lines.append(line) + stripped = line.rstrip() + if stripped.endswith(";") and not stripped.lstrip().startswith("--"): + statement = "\n".join(lines).strip() + statements.append(statement[:-1].rstrip()) + lines = [] + + remainder = "\n".join(lines).strip() + if remainder and any( + line.strip() and not line.lstrip().startswith("--") + for line in remainder.splitlines() + ): + raise AirflowException("DDL-файл оканчивается незавершённым выражением") + return statements + + +def _generator(task_id: str, command: list[str]) -> DockerOperator: + """Запустить канонический образ генератора в сети стенда.""" + return DockerOperator( + task_id=task_id, + image=GENERATOR_IMAGE, + command=command, + network_mode=STAND_NETWORK, + environment={ + "KAFKA_BOOTSTRAP_SERVERS": KAFKA_BOOTSTRAP_SERVERS, + "KAFKA_TOPIC": HITS_TOPIC, + }, + auto_remove="force", + mount_tmp_dir=False, + ) + + +def _kafka_error_code(error: Exception) -> int: + """Код ошибки из KafkaException.""" + return error.args[0].code() + + +@dag( + dag_id="world_initialize", + schedule=None, + start_date=START_DATE, + is_paused_upon_creation=False, + max_active_runs=1, + tags=TAGS, +) +def world_initialize(): + """Создать стартовый мир, если готового мира ещё нет.""" + + @task.branch + def choose_path() -> str: + raw_position = Variable.get(WORLD_POSITION, default=None) + if raw_position is None: + logging.info("world_position отсутствует: создаём первый мир") + return "create_topics" + try: + position = int(raw_position) + except (TypeError, ValueError) as error: + raise AirflowException( + f"world_position={raw_position!r} не является номером дня; " + "выполните make rebuild-storage" + ) from error + if position == 0: + logging.info("очистка завершена: создаём мир заново") + return "create_topics" + if position == -1: + raise AirflowException( + "world_position=-1: очистка мира не завершена; " + "повторите make rebuild-storage" + ) + if position >= STARTING_DAYS: + logging.info("мир уже существует: world_position=%s", position) + return "world_already_exists" + raise AirflowException( + f"world_position={position}: недопустимое незавершённое состояние; " + "выполните make rebuild-storage" + ) + + @task + def world_already_exists() -> None: + """Читаемый зелёный конец ветки без изменений.""" + logging.info("инициализация не требуется: готовый мир не изменён") + + @task + def create_topics() -> None: + """Создать топики до появления читающих их таблиц ClickHouse.""" + from confluent_kafka import KafkaError, KafkaException + from confluent_kafka.admin import AdminClient, NewTopic + + admin = AdminClient({"bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS}) + futures = admin.create_topics( + [ + NewTopic(name, num_partitions=partitions, replication_factor=1) + for name, partitions in TOPICS.items() + ], + request_timeout=30, + operation_timeout=15, + ) + for name, future in futures.items(): + try: + future.result() + logging.info("топик %s создан", name) + except KafkaException as error: + if _kafka_error_code(error) != KafkaError.TOPIC_ALREADY_EXISTS: + raise + logging.info("топик %s уже существует", name) + + metadata = admin.list_topics(timeout=15) + for name, expected_partitions in TOPICS.items(): + topic = metadata.topics.get(name) + actual_partitions = ( + len(topic.partitions) if topic and not topic.error else 0 + ) + if actual_partitions != expected_partitions: + raise AirflowException( + f"у топика {name} разделов {actual_partitions}, " + f"ожидалось {expected_partitions}; " + "выполните make rebuild-storage" + ) + + @task + def apply_ddl() -> None: + """Применить канонические файлы DDL с ноды 1 по порядку имён.""" + files = sorted(SQL_ROOT.glob("*.sql")) + if not files: + raise AirflowException(f"в {SQL_ROOT} нет файлов DDL") + + client = _clickhouse_client() + try: + for path in files: + statements = _ddl_statements(path.read_text(encoding="utf-8")) + logging.info("применяем %s: выражений %s", path.name, len(statements)) + for statement in statements: + client.command(statement) + finally: + client.close() + + @task + def build_generator_image() -> None: + """Собрать генератор из текущего дерева, а не из старого образа.""" + client = DockerHook( + docker_conn_id=None, + base_url="unix://var/run/docker.sock", + timeout=900, + ).get_conn() + output = client.build( + path=str(WORLD_ROOT), + dockerfile="generator/Dockerfile", + tag=GENERATOR_IMAGE, + rm=True, + decode=True, + ) + for item in output: + if error := item.get("error"): + raise AirflowException( + f"сборка генератора завершилась ошибкой: {error}" + ) + if stream := item.get("stream", "").strip(): + logging.info("docker build: %s", stream) + + @task + def wait_for_events() -> None: + """Дождаться приёма всех событий описи, включая строки брака.""" + inventory = json.loads(INVENTORY.read_text(encoding="utf-8")) + expected = sum(day["events"] for day in inventory["days"]) + first_date = inventory["days"][0]["date"] + last_date = inventory["days"][-1]["date"] + query = f""" + SELECT + (SELECT count() FROM ods.event_dist + WHERE EventDate BETWEEN '{first_date}' AND '{last_date}') + + (SELECT count() FROM ods.event_errors_dist) + """ + + client = _clickhouse_client() + try: + arrived = 0 + for attempt in range(1, 101): + row = client.query(query).first_row + if row is None: + raise AirflowException("ClickHouse не вернул счётчик приёма") + arrived = int(row[0]) + if arrived >= expected: + logging.info("стартовый мир принят: строк %s", arrived) + return + if attempt % 5 == 0: + logging.info("принято %s из %s строк", arrived, expected) + time.sleep(3) + finally: + client.close() + raise AirflowException( + f"стартовый мир не принят за 300 с: строк {arrived} из {expected}" + ) + + @task + def remember_starting_position() -> None: + """Объявить мир готовым только после приёма обоих источников.""" + Variable.set(WORLD_POSITION, str(STARTING_DAYS)) + logging.info("мир готов: world_position=%s", STARTING_DAYS) + + route = choose_path() + ready = world_already_exists() + topics = create_topics() + ddl = apply_ddl() + image = build_generator_image() + events = _generator( + "send_initial_events", + ["batch", "--day", "0", "--days", str(STARTING_DAYS)], + ) + arrived = wait_for_events() + snapshots = _generator( + "send_initial_snapshots", + [ + "snapshot", + "--day", + "0", + "--days", + str(STARTING_DAYS), + "--topic", + ORDERS_TOPIC, + ], + ) + orders = TriggerDagRunOperator( + task_id="trigger_orders_ingest", + trigger_dag_id="orders_ingest", + wait_for_completion=True, + poke_interval=10, + ) + position = remember_starting_position() + + route >> [ready, topics] + topics >> ddl >> image >> events >> arrived >> snapshots >> orders >> position + + +@dag( + dag_id="world_recreate", + schedule=None, + start_date=START_DATE, + is_paused_upon_creation=False, + max_active_runs=1, + tags=TAGS, +) +def world_recreate(): + """Удалить только прикладной мир и делегировать его создание.""" + + @task + def invalidate_world() -> None: + Variable.set(WORLD_POSITION, "-1") + logging.info("прежний мир недействителен: world_position=-1") + + @task + def drop_application_databases() -> None: + client = _clickhouse_client() + try: + for database in APPLICATION_DATABASES: + logging.info("удаляем базу %s", database) + client.command( + f"DROP DATABASE IF EXISTS {database} " + f"ON CLUSTER {CLICKHOUSE_CLUSTER} SYNC" + ) + finally: + client.close() + + @task + def delete_consumer_groups() -> None: + from confluent_kafka import KafkaError, KafkaException + from confluent_kafka.admin import AdminClient + + admin = AdminClient({"bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS}) + remaining = set(CONSUMER_GROUPS) + for attempt in range(1, 31): + futures = admin.delete_consumer_groups( + sorted(remaining), request_timeout=15 + ) + for group, future in futures.items(): + try: + future.result() + remaining.discard(group) + logging.info("группа %s удалена", group) + except KafkaException as error: + code = _kafka_error_code(error) + if code == KafkaError.GROUP_ID_NOT_FOUND: + remaining.discard(group) + elif code != KafkaError.NON_EMPTY_GROUP: + raise + if not remaining: + return + if attempt % 5 == 0: + logging.info("ждём выхода потребителей из групп: %s", sorted(remaining)) + time.sleep(2) + raise AirflowException( + f"за 60 с не освободились группы потребителей: {sorted(remaining)}" + ) + + @task + def delete_topics() -> None: + from confluent_kafka import KafkaError, KafkaException + from confluent_kafka.admin import AdminClient + + admin = AdminClient({"bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS}) + futures = admin.delete_topics(list(TOPICS), request_timeout=30) + for name, future in futures.items(): + try: + future.result() + logging.info("топик %s удалён", name) + except KafkaException as error: + if _kafka_error_code(error) != KafkaError.UNKNOWN_TOPIC_OR_PART: + raise + + for _ in range(30): + existing = set(admin.list_topics(timeout=10).topics) & set(TOPICS) + if not existing: + return + time.sleep(1) + raise AirflowException(f"топики не исчезли за 30 с: {sorted(existing)}") + + @task + def mark_world_clean() -> None: + Variable.set(WORLD_POSITION, "0") + logging.info("очистка завершена: world_position=0") + + invalid = invalidate_world() + databases = drop_application_databases() + groups = delete_consumer_groups() + topics = delete_topics() + clean = mark_world_clean() + initialize = TriggerDagRunOperator( + task_id="trigger_world_initialize", + trigger_dag_id="world_initialize", + wait_for_completion=True, + poke_interval=10, + ) + + invalid >> databases >> groups >> topics >> clean >> initialize + + +world_initialize() +world_recreate() diff --git a/generator/src/clickstream_generator/inventory.py b/generator/src/clickstream_generator/inventory.py index 9e3aba7..9811183 100644 --- a/generator/src/clickstream_generator/inventory.py +++ b/generator/src/clickstream_generator/inventory.py @@ -49,9 +49,9 @@ from clickstream_generator import serialize, world from clickstream_generator.catalog import CATALOG_PATH from clickstream_generator.seeds import CANONICAL_SEED -# Сколько дней оси заливается в стенд при подъёме. То же число стоит у службы -# `world-init` в compose.yaml: YAML не читает Python, и одно из двух мест — -# лишнее по построению. Расхождение поймают счётчики make check-clickhouse. +# Сколько дней оси описывает стартовый мир. То же число Airflow получает из +# compose.yaml: YAML не читает Python, и одно из двух мест лишнее по построению. +# Расхождение поймают счётчики make check-clickhouse. STARTING_DAYS = 8 diff --git a/infra/airflow/init.sh b/infra/airflow/init.sh index 74ec46b..4281d8f 100755 --- a/infra/airflow/init.sh +++ b/infra/airflow/init.sh @@ -30,3 +30,13 @@ runuser -u airflow -- airflow connections add clickhouse_default \ --conn-password "$CLICKHOUSE_ETL_PASSWORD" \ --conn-schema default \ --conn-description "ClickHouse, нода 1; типизированный провайдер появится на этапе ETL" + +runuser -u airflow -- airflow connections delete clickhouse_lifecycle >/dev/null 2>&1 || true +runuser -u airflow -- airflow connections add clickhouse_lifecycle \ + --conn-type generic \ + --conn-host clickhouse-01 \ + --conn-port 8123 \ + --conn-login lifecycle \ + --conn-password "$CLICKHOUSE_LIFECYCLE_PASSWORD" \ + --conn-schema default \ + --conn-description "ClickHouse, нода 1; создание и пересоздание прикладного мира" diff --git a/infra/clickhouse/users.d/access.xml b/infra/clickhouse/users.d/access.xml index 88cf8b8..d28b995 100644 --- a/infra/clickhouse/users.d/access.xml +++ b/infra/clickhouse/users.d/access.xml @@ -25,6 +25,22 @@ GRANT FILE ON *.* + + + + GRANT SELECT, INSERT, ALTER, CREATE TABLE, CREATE VIEW, DROP TABLE, DROP VIEW, CREATE DATABASE, DROP DATABASE ON stg.* + GRANT SELECT, INSERT, ALTER, CREATE TABLE, CREATE VIEW, DROP TABLE, DROP VIEW, CREATE DATABASE, DROP DATABASE ON ods.* + GRANT SELECT, INSERT, ALTER, CREATE TABLE, CREATE VIEW, DROP TABLE, DROP VIEW, CREATE DATABASE, DROP DATABASE ON dds.* + GRANT SELECT, INSERT, ALTER, CREATE TABLE, CREATE VIEW, DROP TABLE, DROP VIEW, CREATE DATABASE, DROP DATABASE ON dm.* + GRANT SELECT, CREATE TABLE, CREATE DICTIONARY, dictGet, DROP TABLE, DROP DICTIONARY, CREATE DATABASE, DROP DATABASE ON dic.* + + GRANT SELECT ON system.settings + GRANT CLUSTER ON *.* + GRANT REMOTE ON *.* + GRANT KAFKA ON *.* + GRANT FILE ON *.* + + @@ -75,6 +91,13 @@ GRANT etl_writer + + + default + + GRANT lifecycle_owner + + default diff --git a/scripts/config-test.sh b/scripts/config-test.sh index b8f71fe..b4209e7 100755 --- a/scripts/config-test.sh +++ b/scripts/config-test.sh @@ -57,6 +57,23 @@ jq -e ' ((.environment // {}) | has("_PIP_ADDITIONAL_REQUIREMENTS") | not) ) ' >/dev/null <<<"$config_json" +jq -e ' + .services as $services | + ["kafka-init", "clickhouse-init", "world-init", "world-snapshots"] | + all(. as $name | $services | has($name) | not) +' >/dev/null <<<"$config_json" +jq -e ' + .services.generator as $generator | + ((($generator.depends_on // {}) | has("kafka-init")) | not) and + ((($generator.depends_on // {}) | has("clickhouse-init")) | not) +' >/dev/null <<<"$config_json" +jq -e ' + .services["airflow-scheduler"] as $service | + ($service.environment.CLICKHOUSE_LIFECYCLE_PASSWORD | length > 0) and + any($service.volumes[]; .target == "/opt/airflow/world/generator" and .read_only) and + any($service.volumes[]; .target == "/opt/airflow/world/data" and .read_only) and + any($service.volumes[]; .target == "/opt/airflow/world/.dockerignore" and .read_only) +' >/dev/null <<<"$config_json" jq -e ' .services["airflow-init"] as $service | ($service.environment.AIRFLOW__CORE__SIMPLE_AUTH_MANAGER_PASSWORDS_FILE == diff --git a/scripts/ingest-starting-orders.sh b/scripts/ingest-starting-orders.sh deleted file mode 100755 index b816914..0000000 --- a/scripts/ingest-starting-orders.sh +++ /dev/null @@ -1,94 +0,0 @@ -#!/usr/bin/env bash -# Последний шаг `make up`: запускает единственный путь приёма заказов и ждёт -# его конца. Слепки к этому моменту уже лежат в Kafka, а события доехали до ODS. -set -Eeuo pipefail - -readonly ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -readonly DAG_ID='orders_ingest' -# Потолок каждого ожидания — 300 с: больше чем всемеро от замеренных 40 с -# между запуском сценария и концом дага (#96). Это предел, а не бюджет времени. -readonly ATTEMPTS=100 -readonly PAUSE_SECONDS=3 -readonly REPORT_EVERY=5 - -read -r -a COMPOSE_CMD <<<"${COMPOSE_BIN:-docker compose}" - -fail() { - printf 'ОШИБКА: %s\n' "$1" >&2 - exit 1 -} - -compose() { - "${COMPOSE_CMD[@]}" --project-directory "$ROOT_DIR" "$@" -} - -config="$(compose config --format json)" -user="$(jq -r '.services["airflow-apiserver"].environment.AIRFLOW_ADMIN_USER // empty' <<<"$config")" -password="$(jq -r '.services["airflow-apiserver"].environment.AIRFLOW_ADMIN_PASSWORD // empty' <<<"$config")" -binding="$(compose port airflow-apiserver 8080)" -port="${binding##*:}" -[[ -n "$port" ]] || fail 'не найден отображённый порт API Airflow' - -response="$(curl -sf --max-time 10 -X POST \ - -H 'Content-Type: application/json' \ - -d "$(jq -cn --arg username "$user" --arg password "$password" \ - '{username: $username, password: $password}')" \ - "http://127.0.0.1:${port}/auth/token" 2>/dev/null || true)" -token="$(jq -r '.access_token // empty' <<<"$response")" -[[ -n "$token" ]] || fail 'Airflow не принял учётные данные администратора' - -# После `up --wait` обработчик DAG здоров, но файл мог ещё не попасть в список. -# Ограниченный опрос закрывает эту гонку без паузы наугад. -dag='' -for ((attempt = 1; attempt <= ATTEMPTS; attempt++)); do - dag="$(curl -sf --max-time 10 \ - -H "Authorization: Bearer ${token}" \ - "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}" \ - 2>/dev/null || true)" - jq -e --arg dag_id "$DAG_ID" '.dag_id == $dag_id' \ - >/dev/null 2>&1 <<<"$dag" && break - sleep "$PAUSE_SECONDS" -done -jq -e --arg dag_id "$DAG_ID" '.dag_id == $dag_id' \ - >/dev/null 2>&1 <<<"$dag" || \ - fail "даг ${DAG_ID} не появился в Airflow за $((ATTEMPTS * PAUSE_SECONDS)) с" - -# В API v2 Airflow 3.3 logical_date=null означает событийный ручной запуск. -# Это уместно для дага без расписания: календарь Airflow не подменяет дату -# слепка, которая уже записана в сообщениях. -response="$(curl -sf --max-time 10 -X POST \ - -H "Authorization: Bearer ${token}" \ - -H 'Content-Type: application/json' \ - -d '{"logical_date":null}' \ - "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}/dagRuns" \ - 2>/dev/null || true)" -run_id="$(jq -r '.dag_run_id // empty' <<<"$response")" -[[ -n "$run_id" ]] || fail "Airflow не запустил даг ${DAG_ID}" - -encoded_run_id="$(jq -rn --arg value "$run_id" '$value | @uri')" -printf 'Ждём приём стартовых слепков дагом %s...\n' "$DAG_ID" -state='' -for ((attempt = 1; attempt <= ATTEMPTS; attempt++)); do - response="$(curl -sf --max-time 10 \ - -H "Authorization: Bearer ${token}" \ - "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}/dagRuns/${encoded_run_id}" \ - 2>/dev/null || true)" - state="$(jq -r '.state // empty' <<<"$response")" - case "$state" in - success) - printf 'ЗЕЛЁНО: стартовые слепки приняты в ODS.\n' - exit 0 - ;; - failed) - fail "даг ${DAG_ID} завершился с ошибкой; запуск ${run_id}" - ;; - esac - if ((attempt % REPORT_EVERY == 0)); then - printf 'Приём ещё не завершён: состояние %s, ждём дальше (%d с)...\n' \ - "${state:-неизвестно}" "$((attempt * PAUSE_SECONDS))" - fi - sleep "$PAUSE_SECONDS" -done - -fail "даг ${DAG_ID} не завершился за $((ATTEMPTS * PAUSE_SECONDS)) с: - состояние ${state:-неизвестно}, запуск ${run_id}" diff --git a/scripts/run-dag.sh b/scripts/run-dag.sh new file mode 100755 index 0000000..62bcc40 --- /dev/null +++ b/scripts/run-dag.sh @@ -0,0 +1,108 @@ +#!/usr/bin/env bash +# Запускает публичный даг Airflow и возвращает его итог вызывающему make. +set -Eeuo pipefail + +readonly ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +readonly DAG_ID="${1:-}" +# Каждый цикл ограничен двадцатью минутами: успешная пересборка заняла 3 м 9 с. +readonly ATTEMPTS=400 +readonly PAUSE_SECONDS=3 +readonly REPORT_EVERY=5 + +read -r -a COMPOSE_CMD <<<"${COMPOSE_BIN:-docker compose}" + +fail() { + printf 'ОШИБКА: %s\n' "$1" >&2 + exit 1 +} + +compose() { + "${COMPOSE_CMD[@]}" --project-directory "$ROOT_DIR" "$@" +} + +next_step() { + case "$DAG_ID" in + world_initialize) + printf 'посмотрите журнал; повторите make up, а для недействительного мира — make rebuild-storage' + ;; + world_recreate) printf 'устраните причину и повторите make rebuild-storage' ;; + *) printf 'проверьте журнал дага и повторите команду' ;; + esac +} + +[[ -n "$DAG_ID" ]] || fail 'укажите идентификатор дага' +[[ $# -eq 1 ]] || fail 'ожидался один аргумент: идентификатор дага' + +config="$(compose config --format json)" +user="$(jq -r '.services["airflow-apiserver"].environment.AIRFLOW_ADMIN_USER // empty' <<<"$config")" +password="$(jq -r '.services["airflow-apiserver"].environment.AIRFLOW_ADMIN_PASSWORD // empty' <<<"$config")" +binding="$(compose port airflow-apiserver 8080 2>/dev/null || true)" +port="${binding##*:}" +[[ -n "$user" && -n "$password" ]] || fail 'не найдены учётные данные Airflow в Compose' +[[ -n "$port" ]] || fail 'API Airflow не запущен; сначала поднимите инфраструктуру' + +# В API v2 logical_date=null означает событийный ручной запуск. Даты мира уже +# записаны в данных, календарь Airflow здесь не должен придумывать свою. +if ! response="$(curl -sS --fail-with-body --max-time 10 -X POST \ + -H 'Content-Type: application/json' \ + -d "$(jq -cn --arg username "$user" --arg password "$password" \ + '{username: $username, password: $password}')" \ + "http://127.0.0.1:${port}/auth/token" 2>/dev/null)"; then + fail 'Airflow не принял учётные данные администратора' +fi +token="$(jq -r '.access_token // empty' <<<"$response")" +[[ -n "$token" ]] || fail 'Airflow не вернул токен доступа' + +# Здоровый обработчик DAG ещё не означает, что новый файл уже появился в API. +dag='' +for ((attempt = 1; attempt <= ATTEMPTS; attempt++)); do + dag="$(curl -sf --max-time 10 \ + -H "Authorization: Bearer ${token}" \ + "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}" \ + 2>/dev/null || true)" + jq -e --arg dag_id "$DAG_ID" '.dag_id == $dag_id' \ + >/dev/null 2>&1 <<<"$dag" && break + sleep "$PAUSE_SECONDS" +done +jq -e --arg dag_id "$DAG_ID" '.dag_id == $dag_id' \ + >/dev/null 2>&1 <<<"$dag" || \ + fail "даг ${DAG_ID} не появился за $((ATTEMPTS * PAUSE_SECONDS)) с; $(next_step)" + +if ! response="$(curl -sS --fail-with-body --max-time 10 -X POST \ + -H "Authorization: Bearer ${token}" \ + -H 'Content-Type: application/json' \ + -d '{"logical_date":null}' \ + "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}/dagRuns" 2>/dev/null)"; then + detail="$(jq -r '.detail // .message // empty' <<<"$response" 2>/dev/null || true)" + fail "Airflow не запустил даг ${DAG_ID}${detail:+: ${detail}}; $(next_step)" +fi +run_id="$(jq -r '.dag_run_id // empty' <<<"$response")" +state="$(jq -r '.state // "queued"' <<<"$response")" +[[ -n "$run_id" ]] || fail "Airflow не вернул идентификатор запуска ${DAG_ID}" + +encoded_run_id="$(jq -rn --arg value "$run_id" '$value | @uri')" +printf 'Airflow: даг %s, запуск %s, состояние %s.\n' "$DAG_ID" "$run_id" "$state" + +for ((attempt = 1; attempt <= ATTEMPTS; attempt++)); do + response="$(curl -sf --max-time 10 \ + -H "Authorization: Bearer ${token}" \ + "http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}/dagRuns/${encoded_run_id}" \ + 2>/dev/null || true)" + state="$(jq -r '.state // empty' <<<"$response" 2>/dev/null || true)" + case "$state" in + success) + printf 'ЗЕЛЁНО: даг %s, запуск %s, состояние success.\n' "$DAG_ID" "$run_id" + exit 0 + ;; + failed) + fail "даг ${DAG_ID}, запуск ${run_id}, состояние ${state}; $(next_step)" + ;; + esac + if ((attempt % REPORT_EVERY == 0)); then + printf 'Airflow: даг %s, запуск %s, состояние %s; ждём (%d с).\n' \ + "$DAG_ID" "$run_id" "${state:-неизвестно}" "$((attempt * PAUSE_SECONDS))" + fi + sleep "$PAUSE_SECONDS" +done + +fail "даг ${DAG_ID}, запуск ${run_id}, состояние ${state:-неизвестно} не завершился за $((ATTEMPTS * PAUSE_SECONDS)) с; $(next_step)" diff --git a/scripts/wait-for-world.sh b/scripts/wait-for-world.sh deleted file mode 100755 index de81076..0000000 --- a/scripts/wait-for-world.sh +++ /dev/null @@ -1,93 +0,0 @@ -#!/usr/bin/env bash -# Ждёт, пока стартовый мир доедет до ODS. Второй шаг `make up`, и вот зачем он. -# -# `docker compose up --wait` дожидается служб, в том числе успешно отработавшей -# заливки. Но «заливка кончилась» — это «события лежат в топике», а не «события -# в хранилище»: приём асинхронный. Движок Kafka копит блок и отдаёт его по -# размеру либо по `stream_flush_interval_ms` (умолчание 7,5 с), а стартовый мир — -# около 400 тыс. сообщений. Верни `make up` управление сразу — и `make -# check-clickhouse` следом покраснел бы по устройству, а не по поломке. -# -# Ждём ограниченным циклом с потолком, а не паузой наугад: пауза либо коротка на -# медленной машине, либо ворует минуту на быстрой. -# -# Условие ожидания — «всё доехало», а не «всё разобралось»: сумма событий и -# брака. Сломайся разбор — события уедут в `*_errors`, сумма сойдётся, ожидание -# кончится, и поломку назовёт `make check-clickhouse`. Ждать здесь одних годных -# событий значило бы висеть пять минут вместо внятного ответа. -# -# Обещание у скрипта ровно одно и только на пустом стенде: «мир доехал». На -# живом стенде мир уже лежит в ODS, условие истинно на первой же попытке, и -# ждать тут нечего — повторная заливка ничего не добавляет, её схлопнет -# дедупликация. Утверждать после неё «доехала именно эта заливка» скрипт не -# может и не пытается: у события нет отметки, которым прогоном оно приехало. -set -Eeuo pipefail - -readonly ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -readonly INVENTORY="${ROOT_DIR}/data/world-inventory.json" -# Потолок ожидания — 300 с: больше чем вдесятеро от замеренных 22,5 с, за -# которые заливка отдаёт восемь дней в Kafka (карта проверок, «Что проверено»). -# Это не бюджет, а предел, за которым ждать бессмысленно и надо читать журнал. -readonly ATTEMPTS=100 -readonly PAUSE_SECONDS=3 -# Как часто отчитываться о ходе: молчащая минуту команда выглядит зависшей. -readonly REPORT_EVERY=5 - -read -r -a COMPOSE_CMD <<<"${COMPOSE_BIN:-docker compose}" - -fail() { - printf 'ОШИБКА: %s\n' "$1" >&2 - exit 1 -} - -query() { - "${COMPOSE_CMD[@]}" --project-directory "$ROOT_DIR" exec -T clickhouse-01 \ - sh -c ' - clickhouse-client \ - --user default \ - --password "$CLICKHOUSE_DEFAULT_PASSWORD" \ - --query "$1" - ' _ "$1" /dev/null || true)" - if [[ "$arrived" =~ ^[0-9]+$ ]] && ((arrived >= expected)); then - if ((arrived > expected)); then - # Повторный `make up` заливает мир заново. Промолчи мы об этом — - # удвоенное число под «ЗЕЛЁНО» выглядело бы поломкой. - printf 'ЗЕЛЁНО: стартовый мир доехал; строк %s при ожидаемых %s —' \ - "$arrived" "$expected" - printf ' мир заливали повторно, повтор схлопнет ReplacingMergeTree.\n' - else - printf 'ЗЕЛЁНО: стартовый мир доехал: %s строк.\n' "$arrived" - fi - exit 0 - fi - if ((attempt % REPORT_EVERY == 0)); then - printf 'Доехало %s из %s, ждём дальше (%d с)...\n' \ - "${arrived:-0}" "$expected" "$((attempt * PAUSE_SECONDS))" - fi - sleep "$PAUSE_SECONDS" -done - -fail "стартовый мир не доехал за $((ATTEMPTS * PAUSE_SECONDS)) с: - доехало ${arrived:-0} из ${expected}. Смотрите журнал заливки - (docker compose logs world-init) и чтеца топика: - SELECT * FROM system.kafka_consumers" diff --git a/sql/ddl/00-databases.sql b/sql/ddl/00-databases.sql index a567a12..8be3528 100644 --- a/sql/ddl/00-databases.sql +++ b/sql/ddl/00-databases.sql @@ -5,8 +5,8 @@ -- окажется лицом половины кластера. Имя кластера — clickstream_cluster, -- задано в infra/clickhouse/config.d/cluster.xml. -- --- Идемпотентность везде через IF NOT EXISTS: make up применяет эти файлы и --- поверх живого тома. +-- Идемпотентность везде через IF NOT EXISTS: world_initialize может повторить +-- применение после оборванного первого запуска. CREATE DATABASE IF NOT EXISTS stg ON CLUSTER clickstream_cluster; diff --git a/sql/ddl/10-stg-tables.sql b/sql/ddl/10-stg-tables.sql index 6836cae..c040356 100644 --- a/sql/ddl/10-stg-tables.sql +++ b/sql/ddl/10-stg-tables.sql @@ -94,7 +94,7 @@ ENGINE = Distributed('clickstream_cluster', 'stg', 'hits_raw_rep', cityHash64(ra -- Чтец топика orders. Матвью к нему не привязана: слепок забирает прямым -- SELECT даг orders_ingest. Почему пулл, почему без матвью и почему у топика --- одна партиция — ADR 0008; сам топик создаёт kafka-init в compose.yaml. +-- одна партиция — ADR 0008; сам топик заранее создаёт world_initialize. -- -- Без ON CLUSTER: таблица нужна только на clickhouse-01 — той ноде, к которой -- у Airflow подключение, и она же одна читает топик.