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/AGENTS.md b/AGENTS.md index 5e9f478..3239c12 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -82,6 +82,11 @@ Airflow) и названия из кода. Если для понятия ес `generator/` — код генератора. Правишь даги или Superset — зови корневую, правишь генератор — зови из `generator/`. - Изменения держать минимальными и в границах задания. +- `make clean` полностью удаляет постоянное состояние стенда. Перед этой + командой назови состояние вне прикладного мира, которое требуется удалить, + и объясни, почему `make rebuild-storage` его не удаляет. При изменении DDL + или проверке на чистых данных по умолчанию используй адресную очистку либо + `make rebuild-storage`. - Куда класть новую проверку, что утверждает каждая цель `make` и почему смоук обязан оставаться быстрым — [`docs/architecture/testing.md`](docs/architecture/testing.md). - Секреты не коммитить: настройки — через `.env`, образец — `.env.example`. 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/README.md b/README.md index f2c3822..f1f0502 100644 --- a/README.md +++ b/README.md @@ -67,9 +67,10 @@ make check-services стенд и ждёт здорового состояния долгоживущих контейнеров. В образ Airflow добавлены закреплённые клиенты ClickHouse и Kafka. Одноразовые `airflow-init` и `superset-init` завершаются с кодом 0. Первый обновляет схему Airflow, -подготавливает администратора и подключение к `clickhouse-01`. Второй обновляет -Superset, создаёт администратора и импортирует подключение к `clickhouse-02`. -С нуля подъём занимает около трёх минут, на живом стенде — около минуты. +подготавливает администратора и два подключения к `clickhouse-01`. Второй +обновляет Superset, создаёт администратора и импортирует подключение к +`clickhouse-02`. На ноутбуке в режиме энергосбережения приёмочный подъём с +пустых томов занял около десяти минут, повторный на живом стенде — около трёх. `make smoke` за секунды спрашивает, собран ли стенд: зависимости машины, здоровье контейнеров, устройство keeper, ответ Kafka с машины через отображённый @@ -118,29 +119,33 @@ Kafka по той же причине спрашивают снаружи. Её поднятый стенд, сколько стоит прогон и куда класть новую — в [карте проверок](docs/architecture/testing.md). -Остановить контейнеры без удаления данных можно командой `make down`. Для -полного сброса с удалением всех именованных томов используйте `make clean`. -Повторный `make up` безопасен: одноразовая подготовка приложений идемпотентна. +Остановить контейнеры без удаления данных можно командой `make down`. Если +нужно вернуть прикладной мир к началу, `make rebuild-storage` запустит в +Airflow его явную пересборку: удалит прикладные базы ClickHouse, топики и +позиции потребителей Kafka, затем создаст схему и данные заново. Службы и их +тома останутся на месте. `make clean` удаляет весь стенд вместе с именованными +томами и для обычной разработки не нужен. Перед пересборкой поставьте +`world_live` на паузу и дайте уже начатому работнику закончить день. ## Стартовый мир -Стенд поднимается не пустым: разовая служба `world-init` играет в топик `hits` -первые восемь дней модельного времени — понедельник по понедельник, 401 185 -событий. Следом второй запуск того же генератора отправляет в топик `orders` -семь слепков — за дни 0…6. Дальше оба источника разбирает хранилище, и к концу -`make up` события лежат в `ods.event`, а заказы — в `ods.order_v`. Так у всякой -лабы есть данные, и всегда одни и те же. +Стенд поднимается не пустым: `make up` через REST API запускает даг Airflow +`world_initialize`. Тот создаёт прикладные топики и схему, собирает образ +генератора и играет в `hits` первые восемь дней модельного времени — понедельник +по понедельник, 401 185 событий. Следом второй запуск генератора отправляет в +`orders` семь слепков — за дни 0…6. Даг ждёт приёма обоих источников и только +тогда ставит позицию мира в `8`. Так у всякой лабы есть данные, и всегда одни и +те же. Последний день намеренно неполон: события дня 7 уже приехали, а его слепок уедет только с первым ходом мира. Поэтому на свежем стенде правый край графика выручки отстаёт от кликстрима и дозаполняется по мере движения мира — так виден разный темп потокового трекера и ночной выгрузки магазина. -Ждать приходится дольше, чем работает заливка. `make up` сначала опрашивает -ClickHouse, пока события не доедут, затем разово запускает даг `orders_ingest` -и ждёт его конца. Повторный `make up` заливает мир заново; это не ошибка, а -свойство: номера событий и версии заказов те же, и повторы схлопнут -`ReplacingMergeTree`. +Повторный `make up` безопасен: `world_initialize` видит готовую позицию, +сообщает в журнале, что мир уже существует, и не применяет DDL, не собирает +образ и не заливает данные повторно. Изменённую схему применяют адресно либо +явной командой `make rebuild-storage`. Сам мир в git не хранится — он чистая функция зерна, и держать его в репозитории значило бы держать там кэш. Вместо него лежит **опись мира**, @@ -202,9 +207,9 @@ make generate-live GENERATOR_DAY=3 GENERATOR_SPEED=1000 docker compose --profile generator build generator ``` -Тот же образ несёт заливка стартового мира, поэтому свежим его держит и -обычный `make up`: он собирает образы всего стенда, и генератор теперь среди -них. +Тот же образ несёт заливка стартового мира. Перед такой заливкой +`world_initialize` сам собирает его из текущего дерева; на готовом мире +повторный `make up` до сборки не доходит. Разведено это нарочно: собрать образ и запустить контейнер — разные действия, и цель запуска, молча пересобирающая образ, стирает между ними границу. Что образ @@ -231,11 +236,13 @@ uv run --project generator python -m clickstream_generator batch \ ### Пульт мира -Считать дни самому необязательно. В Airflow живёт **пульт мира** — три дага, -которые зовут тот же образ генератора и ведут позицию на оси за вас. +Считать дни самому необязательно. В Airflow живёт **пульт мира**: два дага +создают его начальное состояние, а три двигают ось дальше. | Даг | Что делает | | --- | --- | +| `world_initialize` | создаёт стартовый мир, если его ещё нет; повторный запуск ничего не меняет | +| `world_recreate` | удаляет только прикладной мир и запускает его создание заново | | `world_next_day` | играет следующие дни пачкой; сколько — параметр запуска, по умолчанию один | | `world_live_day` | играет следующий день в темпе модельного времени, около двадцати четырёх минут | | `world_live` | выключатель: пока снят с паузы, дёргает `world_live_day` день за днём | @@ -248,8 +255,10 @@ uv run --project generator python -m clickstream_generator batch \ Позицию хранит переменная Airflow `world_position` — номер первого несыгранного дня. Ставит её работник, сыгравший день, и только по успеху: оборванный прогон -позицию не двигает, и следующий запуск играет тот же день заново. Нет -переменной — мир в стартовом состоянии. +позицию не двигает, и следующий запуск играет тот же день заново. Отсутствующая +переменная означает, что мир ещё не создавали; `-1` — очистка не закончена, +`0` — очистка закончена, а начальная генерация ещё не завершена. Работники +начинают играть только с позиции `8` и больше. Выключатель создаётся на паузе. Снимите — мир поедет сам; поставите обратно — встанет на границе модельных суток, доиграв начатый день. Форма пульта и доводы @@ -286,10 +295,12 @@ uv run --project generator python -m clickstream_generator batch \ - Prometheus — `http://127.0.0.1:29090`; - Grafana — `http://127.0.0.1:23000`, пользователь `admin`, пароль `admin`. -В ClickHouse шесть пользователей и пять ролей: +В ClickHouse семь пользователей и шесть ролей: -- `etl` применяет DDL и подключает Airflow; роль `etl_writer` читает и пишет - слои хранилища; +- `etl` подключает рабочие даги Airflow; роль `etl_writer` читает и пишет слои + хранилища; +- `lifecycle` применяет канонический DDL и пересоздаёт прикладные базы; роль + `lifecycle_owner` не даёт эти административные права рабочим дагам; - `bi` подключает Superset; роль `bi_reader` читает справочники `dic` и будущие слои DDS и DM; - `analyst` предназначен для подключения человека и читает все слои и diff --git a/compose.yaml b/compose.yaml index 3abf49d..5f21f01 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 @@ -46,10 +46,6 @@ x-kafka-target: &kafka-target <<: *kafka-common KAFKA_TOPIC: hits -x-orders-kafka-target: &orders-kafka-target - <<: *kafka-common - KAFKA_TOPIC: *orders-topic - x-world-starting-days: &world-starting-days "8" x-airflow-common: &airflow-common @@ -73,9 +69,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 +82,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 +97,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 +220,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 +264,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 +304,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..492ec04 --- /dev/null +++ b/dags/world_lifecycle.py @@ -0,0 +1,416 @@ +"""Создание и пересоздание прикладного мира. + +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, + # `ON CLUSTER` ждёт хосты до 180 с. Транспорт живёт дольше, чтобы + # ClickHouse сам назвал незавершённый хост вместо сетевого тайм-аута. + send_receive_timeout=300, + ) + + +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", + # Временный путь Airflow существует внутри контейнера, но монтировать + # его пытается хостовый Docker. Генератору этот каталог не нужен. + 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}; " + "проверьте журнал send_initial_events и " + "SELECT * FROM system.kafka_consumers" + ) + + @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/docs/adr/0001-stand-services.md b/docs/adr/0001-stand-services.md index 0282380..4586a4d 100644 --- a/docs/adr/0001-stand-services.md +++ b/docs/adr/0001-stand-services.md @@ -10,6 +10,12 @@ обработчик DAG. Triggerer не запускается: в каркасе нет отложенных задач. Два интеграционных пробника написаны через публичный `airflow.sdk`. +Долгоживущие процессы Airflow зависят от `airflow-init`, а Superset — от +`superset-init`. Это ребро нужно самому `up --wait`: Docker Compose 2.40.3 +считает успешно завершившийся одноразовый сервис без зависимого упавшим и +возвращает единицу — поведение проверено при исполнении #37. Зависимость +включает результат подготовки в общий граф ожидания. + Для входа выбран SimpleAuthManager. У него нет команды создания пользователя, поэтому `airflow-init` записывает пароль администратора из окружения в JSON-файл отдельного тома `airflow_auth`. Новый именованный том принадлежит @@ -24,9 +30,11 @@ JSON-файл отдельного тома `airflow_auth`. Новый имен пользователи и строки подключения. Это экономит один контейнер, но сохраняет границу владения данными. -Airflow получает подготовленное общее подключение к `clickhouse-01`, а -Superset — подключение `clickhousedb://` к `clickhouse-02`. Провайдер -ClickHouse для Airflow не нужен: локальный образ содержит прямой клиент. +Airflow получает два подготовленных подключения к `clickhouse-01`: рабочее +`clickhouse_default` и административное `clickhouse_lifecycle` для дагов +жизненного цикла мира ([ADR 0013](0013-world-lifecycle.md)). Superset получает +подключение `clickhousedb://` к `clickhouse-02`. Провайдер ClickHouse для +Airflow не нужен: локальный образ содержит прямой клиент. Пробник создаёт служебные таблицы `ON CLUSTER`, пишет на первой ноде и через `remote` читает `Distributed` на второй. Разные ноды сохраняют учебную ловушку: забытый `ON CLUSTER` проявится в Superset, даже если операция Airflow diff --git a/docs/adr/0007-clickhouse-access.md b/docs/adr/0007-clickhouse-access.md index afb09c5..a9a2043 100644 --- a/docs/adr/0007-clickhouse-access.md +++ b/docs/adr/0007-clickhouse-access.md @@ -9,8 +9,11 @@ - `default` — с паролем, только служебный: проверки здоровья нод и работа изнутри контейнеров. Приложения им не ходят. -- `etl` с ролью `etl_writer` — чтение и запись во всех слоях. Им ходит - Airflow и применяется DDL при подъёме стенда. +- `etl` с ролью `etl_writer` — чтение и запись во всех слоях. Им ходят рабочие + даги Airflow. +- `lifecycle` с ролью `lifecycle_owner` — создание и удаление только + прикладных баз и объектов. Им даги жизненного цикла применяют DDL и + пересоздают мир ([ADR 0013](0013-world-lifecycle.md)). - `bi` с ролью `bi_reader` — чтение витрин DM, слоя DDS и справочников `dic`. Им ходит Superset. - `analyst` с ролью `analyst_reader` — чтение всех слоёв и справочников. Им @@ -127,10 +130,11 @@ DDS под ними, потому что так чаще всего и быва Единственное, что ему доступно сегодня, — справочники: с ними зона `dic` пришла раньше своих потребителей. -Владелец матвью приёма определяется тем, кто применил DDL. Весь DDL стенда -идёт через `IF NOT EXISTS`, поэтому на существующих томах три матвью -сохранят нынешнего владельца, а на собранном заново стенде им станет `etl`. -Работают оба — права есть у обоих, — но два стенда в этом расходятся. +DDL для матвью приёма выполняется от имени `lifecycle`. Это видно в журнале +запросов, а при создании ClickHouse проверяет у этой учётки чтение источника и +вставку в цель. Так отдельная учётка показывает границу между редким +управлением структурой и повседневной записью `etl`, не выдавая рабочим дагам +право удалять базы. Лаба «сломать права» вариантом с файлами не закрывается: создать через SQL новую роль и нового пользователя ничто не мешает, и новую роль разрешено @@ -182,8 +186,8 @@ ClickHouse через MCP Context7 (#114). - Секрет кластера принимается внутри `remote_servers` в виде ``. Запрос от `etl` через обе ноды выполнился на каждой под пользователем `etl`. -- `from_env` работает у элемента `password`: `etl`, `bi` и `analyst` вошли с - паролями из окружения и получили свои роли. +- `from_env` работает у элемента `password`: `etl`, `lifecycle`, `bi` и + `analyst` вошли с паролями из окружения и получили свои роли. - При запросе от `bi` через Distributed на второй ноде `system.query_log` показал `user = bi` и `initial_user = bi`. Значит, на соседнем шарде применяются права инициатора, а не общей учётной записи. diff --git a/docs/adr/0009-world-control.md b/docs/adr/0009-world-control.md index 2cdbbc7..8ad13b4 100644 --- a/docs/adr/0009-world-control.md +++ b/docs/adr/0009-world-control.md @@ -30,6 +30,11 @@ работник берёт из переменной, а не из логической даты прогона: календарь Airflow к оси мира отношения не имеет. +Позднее [ADR 0013](0013-world-lifecycle.md) отдал Airflow начальную позицию и +состояния обслуживания: переменной нет до первого создания мира, `-1` означает +незавершённую очистку, `0` — чистый мир до генерации, а работники принимают +только позицию `8` и больше. + Пакетная автоматика — второй выключатель, дёргающий `world_next_day`, — **отложена, а не отвергнута**. diff --git a/docs/adr/0013-world-lifecycle.md b/docs/adr/0013-world-lifecycle.md new file mode 100644 index 0000000..8b48a40 --- /dev/null +++ b/docs/adr/0013-world-lifecycle.md @@ -0,0 +1,67 @@ +# ADR 0013. Жизненный цикл мира ведёт Airflow + +Дата: 21 августа 2026 года. Статус: принято. Реализация — тикет #119. + +## Решение + +Compose поднимает службы стенда. Airflow владеет прикладным миром: топиками +Kafka, DDL прикладных баз ClickHouse, стартовыми данными и позицией модельного +времени. `make` поднимает инфраструктуру и через REST API запускает нужный даг, +дожидаясь его результата. + +Безопасный даг `world_initialize` создаёт мир, если тот ещё не создан. На +готовом мире он ничего не меняет и успешно сообщает об этом в журнале. +Разрушительный даг `world_recreate` очищает прикладные базы, топики и позиции +их потребителей, затем запускает `world_initialize` и ждёт его завершения. +Канонический DDL применяет отдельная учётная запись Airflow с правами только на +прикладную структуру стенда. + +`world_position` остаётся единственным состоянием пульта: + +- переменной нет — стенд ещё не инициализировали; +- `-1` — прежний мир признан недействительным, очистка не завершена; +- `0` — мир очищен и готов к начальной генерации; +- `8` и больше — первый несыгранный день готового мира. + +`world_recreate` ставит `-1` до первого удаления и `0` только после успешной +очистки. `world_initialize` принимает отсутствие позиции или `0`, а значение +`8` и больше считает готовым миром. Позиция становится `8` последней задачей, +после генерации стартовых дней и успешного приёма обоих источников. Обычные +даги движения мира работают только с готовой позицией. + +## Почему + +Прежняя цепочка делила один процесс между Compose, shell-скриптами и Airflow: +одноразовые службы создавали стартовые данные, отдельный сценарий ждал их +приёма, а пересборка сама исправляла внутреннюю переменную пульта. Из-за этой +границы успешная отправка в Kafka выдавалась за создание мира, повторный +`make up` смешивался с инициализацией данных, а узкая пересборка сохраняла в +Kafka сообщения и офсеты прежнего мира. + +Airflow уже ведёт последующие дни и приём заказов. Отдав ему начало той же оси, +получаем один наблюдаемый процесс и одну точку успеха: зелёный +`world_initialize` означает, что схема создана, стартовый мир принят и его +следующий день известен. + +## Что это решение уточняет + +- В [ADR 0007](0007-clickhouse-access.md) DDL больше не принадлежит `etl`: + прикладную структуру ведёт отдельный `lifecycle`. +- В [ADR 0009](0009-world-control.md) отсутствие `world_position` больше не + означает готовый стартовый мир. Начальную позицию ставит + `world_initialize`, а работники принимают только `8` и больше. +- Одноразовые службы Compose и shell-ожидание, описанные прежним устройством + хранилища, отменены. Их последовательность стала наблюдаемым графом Airflow. + +## Отклонено + +- Оставить создание мира в Compose, а позицию исправлять из + `make rebuild-storage`: работает технически, но сохраняет двух + оркестраторов одного процесса. +- Один безусловно разрушительный даг для первого и повторного запуска: обычный + `make up` мог бы стереть уже прожитые дни. +- Отдельная переменная состояния мира: она дублировала бы + `world_position` и могла разойтись с ней. Значения `-1` и `0` выражают + нужные переходы в одном месте. +- Выдать обычной учётной записи `etl` право удаления баз: рабочий конвейер не + должен получать административные права из-за редкой операции восстановления. diff --git a/docs/architecture/monitoring/cluster.md b/docs/architecture/monitoring/cluster.md index 5fe9547..345cdbf 100644 --- a/docs/architecture/monitoring/cluster.md +++ b/docs/architecture/monitoring/cluster.md @@ -144,8 +144,10 @@ Keeper — отдельный сервис, системных таблиц у вышло восемь: четыре таблицы, каждая обеими нодами. Таблиц ошибок среди них нет, и это правда состояния — частей у них пока не появилось. - Очередь DDL показала `CREATE MATERIALIZED VIEW ... ON CLUSTER` в состоянии - `Finished` и со временем по каждой ноде: `make generate-batch` поднимает - `clickhouse-init`, и тот заново прогоняет DDL. + `Finished` и со временем по каждой ноде. Новые безопасные записи для показа + оставляет `make check-clickhouse`: цель создаёт и удаляет временные таблицы + `ON CLUSTER`. Канонический DDL появляется в очереди при создании мира дагом + `world_initialize`. Подписи рядов (`${__field.labels.node}`) и правило, по которому пунктирным становится только ряд «отведено», сверены по документации Grafana через MCP diff --git a/docs/architecture/orders/start-world.md b/docs/architecture/orders/start-world.md index b1639b0..083cf11 100644 --- a/docs/architecture/orders/start-world.md +++ b/docs/architecture/orders/start-world.md @@ -6,13 +6,14 @@ Генератор играет два источника с разными темпами — поток Метрики и ночную выгрузку магазина; разные темпы дают всё остальное. -`world-init` играет восемь дней одним прогоном `batch --day 0 --days 8`; -слепки отправляет второй запуск — команда `snapshot` того же диапазона (два -источника — два запуска, [слепок и его доставка](snapshot.md)), и со сдвигом -отправки уезжают **слепки дней 0…6**. Слепок дня 7 снят на границе суток и -уедет первым же ходом мира. Особого режима у стартового мира нет: правило -отправки живёт в одном месте — в проигрывателе; генератору это решение не -стоит ничего. +Даг `world_initialize` играет восемь дней одним прогоном +`batch --day 0 --days 8`; слепки отправляет второй запуск — команда `snapshot` +того же диапазона (два источника — два запуска, +[слепок и его доставка](snapshot.md)). Со сдвигом отправки уезжают **слепки +дней 0…6**, затем даг `orders_ingest` принимает их в хранилище. Слепок дня 7 +снят на границе суток и уедет первым же ходом мира. Особого режима у стартового +мира нет: правило отправки живёт в одном месте — в проигрывателе; генератору +это решение не стоит ничего. Что видно снаружи: у последнего прожитого дня клики есть, а заказов нет; глубже — день 0 виден дожившим до конца окна, день 6 — только что родившимся. diff --git a/docs/architecture/storage.md b/docs/architecture/storage.md index 5702a25..551b199 100644 --- a/docs/architecture/storage.md +++ b/docs/architecture/storage.md @@ -480,44 +480,35 @@ Airflow читает и собирает эти файлы штатным шаб Порядок остальных задают два правила. Первое: матвью принадлежит слою своей цели, а не источника, — разбор из STG в ODS лежит среди файлов ODS, потому что -наполняет ODS. Второе: матвью приёма создаётся последней из всех, и потому нарушает -нумерацию слоёв. Kafka-движок начинает читать топик ровно тогда, когда к нему +наполняет ODS. Второе: матвью приёма создаётся последней из всех, и потому +нарушает нумерацию слоёв. Kafka-движок начинает читать топик, когда к нему привязывают первую матвью; создай её раньше разбора — и всё, что доедет в зазоре, ляжет в сырьё и не попадёт в ODS никуда, ни в событие, ни в ошибки. На пустом топике зазор безвреден, поэтому первый прогон о нём не скажет. Проснётся он, когда тома ClickHouse снесены, а данные Kafka целы, — то есть на обычной отладке. -Применение — двумя одноразовыми сервисами при `make up`, по образцу уже -работающих `airflow-init` и `superset-init`. Сначала `kafka-init` создаёт топики -— `hits` с двумя партициями и `orders` с одной ([ADR -0008](../adr/0008-order-ingestion.md)), — затем `clickhouse-init` дожидается его -завершения и применяет файлы с ноды 1: всё `ON CLUSTER`, кроме чтеца заказов — -он объявлен только на этой ноде. Порядок страхует `hits` от автосоздания с -одной партицией. Переключателей тут два, и путать их не надо: брокер -автосоздание разрешает, а потребитель librdkafka внутри ClickHouse его не -просит — оба конца измерены, см. «Что проверено». То есть стенд держится на -умолчании клиента, а урок «обе ноды читают топик» умирает тихо, поэтому топик и -создаётся явно, до применения DDL. +Файлы применяет даг `world_initialize` с ноды 1 отдельной учётной записью +`lifecycle`: всё `ON CLUSTER`, кроме чтеца заказов — он объявлен только на этой +ноде. Перед DDL даг создаёт `hits` с двумя партициями и `orders` с одной +([ADR 0008](../adr/0008-order-ingestion.md)). Этот порядок не даёт `hits` +автоматически создаться с одной партицией. Брокер автосоздание разрешает, а +потребитель librdkafka внутри ClickHouse его не просит. Оба конца измерены; +результаты приведены в разделе «Что проверено». Успех дага означает больше, +чем применённый DDL: оба стартовых +источника приняты, а позиция мира установлена последней задачей. Полный граф и +состояния отказа описаны в [спеке жизненного цикла](../specs/2026-08-21-world-lifecycle.md). -Образцы копируются не целиком, и в двух местах. `clickhouse-init` обязан ждать -готовности **обеих** нод: `ON CLUSTER` ждёт исполнения на всех хостах и по -таймауту бросает, а `airflow-init` ждёт только первую ноду, `superset-init` — -только вторую. И второе: оба образца переживают `make up --wait` лишь потому, что -от них зависят долгоживущие сервисы. Что делает `--wait` с одноразовым сервисом -без зависимых, проверено при исполнении #37 на Docker Compose 2.40.3: считает -его упавшим и возвращает единицу, хотя контейнер вышел с нулём. Поэтому -зависимые есть и у новой пары: `clickhouse-init` ждёт `kafka-init`, а -`airflow-init` — `clickhouse-init`, и цепочка упирается в долгоживущий Airflow. -Побочная выгода важнее обхода `--wait`: к моменту старта Airflow DDL заведомо -применён. - -Повторный `make up` поверх живого тома проходит зелёным: весь DDL идёт через -`CREATE ... IF NOT EXISTS`. Оборотная сторона — изменённый объект тем же -запуском не применяется, причём молча. Отдельного механизма для этого нет и не -нужно: правка существующего DDL случается, только пока стенд пишут, а лекарство -уже есть — `make clean && make up`. Мир регенерируется, сырьё живёт трое суток, -терять нечего. +Повторный `make up` поверх готового мира проходит зелёным без изменений: +`world_initialize` ветвится раньше DDL. Поэтому изменённый объект обычным +подъёмом не применяется. В разработке меняют или пересоздают только затронутый +объект, партицию либо базу слоя. Если нужен целиком канонический мир, +`make rebuild-storage` запускает `world_recreate`: тот удаляет прикладные базы +`stg`, `dm`, `dds`, `ods`, `dic`, позиции групп потребителей и прикладные топики, +затем вызывает `world_initialize`. Службы и их тома остаются на месте, а позиция +возвращается к концу стартовых дней. Полный `make clean` для изменения DDL не +нужен. Когда какая область очистки уместна перед проверкой, задаёт +[карта проверок](testing.md). ## Карта таблиц diff --git a/docs/architecture/testing.md b/docs/architecture/testing.md index e11cbce..cac78c3 100644 --- a/docs/architecture/testing.md +++ b/docs/architecture/testing.md @@ -71,9 +71,15 @@ тогда окружение окупится, и условие возврата — именно это, а не «стало неудобно». -Цена самого `make up` — 8 м 18 с с нуля и 3 м 35 с на живом стенде. В цену -целей она не входит, но записана здесь по той же причине: подъём пересчитывает -оба источника стартового мира и ждёт их приёма. +Цена самого `make up` с пустых томов — 8 м 18 с до переезда жизненного цикла в +Airflow. В цену целей она не входит. На готовом мире подъём больше не +пересчитывает данные: `world_initialize` зеленеет без изменений. + +`make rebuild-storage` тоже не проверка. Команда через Airflow удаляет +прикладные базы ClickHouse, топики и позиции групп Kafka, заново применяет DDL, +заливает стартовый мир и возвращает его пульт к концу стартовых дней. Службы и +их тома она сохраняет, поэтому возврат к каноническим данным не требует +холодного запуска всего стенда. ## Правило: смоук обязан оставаться быстрым @@ -96,11 +102,19 @@ ClickHouse отвечает сразу. Деление целей ценно ровно до тех пор, пока оно не превратилось в «гонять всегда всё». +Состояние перед проверкой восстанавливается в самой узкой достаточной области: +сначала объект или партиция, затем база слоя. Если нужен целиком чистый +прикладной мир, `make rebuild-storage` пересобирает его базы, топики, позиции +потребителей и данные. `make clean` нужен, только когда меняется или сломано +постоянное состояние вне этих баз либо проверяется сам запуск с пустых томов. +Требование к обоснованию полного сброса задаёт `AGENTS.md`. Неизвестную поломку +сначала исследуют: полный сброс уничтожает её следы. + | Когда | Что гонять | |---|---| | Правка в работе | `make config-test` и `make smoke` | | PR или задача | плюс цели, которых правка касалась: DAG-и, Superset или Kafka — `check-services`; DDL, кластер или данные — `check-clickhouse` | -| Приёмка этапа | `make up` с нуля и все три цели на стенде | +| Приёмка этапа | все три цели на стенде; запуск с пустых томов — только для изменений начальной сборки или постоянного состояния служб | Правки генератора добавляют к этому `make lint`, `make typecheck` и `make test` из каталога `generator/`: стенд им не нужен, а `make test` из них самая @@ -194,12 +208,12 @@ Kafka → STG → ODS, и живёт он в `make check-clickhouse`. | `make check-services` | `scripts/stand-services.sh` | | `make check-clickhouse` | `scripts/check-clickhouse.sh` | -Особняком — два сценария продолжения `make up`, оба не проверки. -`scripts/wait-for-world.sh` ждёт, пока события доедут из Kafka в ODS: -`docker compose up --wait` видит конец отправки, но приём идёт асинхронно. -Затем `scripts/ingest-starting-orders.sh` разово запускает единственный даг -приёма заказов `orders_ingest` и ждёт его конца. Оба ждут ограниченным циклом, -а не паузой наугад. +Особняком стоит `scripts/run-dag.sh`: это не проверка, а общий REST-переход от +`make` к Airflow. `make up` запускает через него `world_initialize`, а +`make rebuild-storage` — `world_recreate`; сценарий ждёт появления дага в API, +печатает идентификаторы дага и прогона и возвращает вызывающему его итог. +Ожидание доставки событий и запуск `orders_ingest` теперь видны задачами +`world_initialize`, а не спрятаны в shell. Общее у смоука и `check-services` — счёт проверок, обращение к Compose и две проверки — вынесено в `scripts/stand-common.sh`; сам он не запускается. @@ -227,9 +241,42 @@ Kafka → STG → ODS, и живёт он в `make check-clickhouse`. 19/2: назвал нездоровый сервис и отсутствие одной из четырёх целей. После возврата сервиса прогон снова дал 21/21. -**Перезамер 20 августа 2026 года при исполнении #96.** `make up` после -`make clean` прошёл за 8 м 18 с, повтор на живом стенде — за 3 м 35 с. Служба -семи слепков заняла около 28 секунд между концом `world-init` и своим концом. +После переноса #119 поверх этого состава повторный `make up` сохранил готовый +мир, смоук прошёл 21/21, проверка ClickHouse — 10/10, межсервисная — 7/7. + +**Приёмка логики #119 до переноса поверх мониторинга 21 августа 2026 года.** +`make rebuild-storage` через Airflow прошёл за 3 м 9 с: позиция прошла задачи +`-1` → `0` → `8`, топики получили новые идентификаторы, группы дочитали новые +топики без отставания, а ODS принял 401 185 событий без брака и 2 851 +актуальный заказ. Задача `build_generator_image` заняла по 1,5 с в двух +прогонах. +Повторный `make up` занял 2 м 41 с вместе со штатным повторным запуском служб +инициализации; `world_initialize` завершился за один цикл опроса и оставил +счётчики неизменными. Отдельно проверены красные отказы `world_initialize` при +`-1` и `world_next_day` при отсутствующей позиции, `0` и `-1`; `lifecycle` +отказано в чтении `system.users`, но разрешено чтение `system.settings`. + +Итоговый холодный `make up` из пустых томов занял 9 м 39 с после 25 с на +`make clean`. Даг создал отсутствующую позицию и закончил на `8`; ODS снова +получил 401 185 событий без брака и 2 851 заказ. У `hits` два раздела, у +`orders` один, отставание обеих групп равно нулю. На тогдашнем составе стенда +прошли смоук 20/20, проверка ClickHouse 10/10 и межсервисная проверка 7/7. + +Оборванную очистку проверили после записи `world_position=-1`: остановили +первую ноду ClickHouse, и `world_recreate` покраснел на удалении баз. Следующий +`make up` тоже покраснел и оставил `-1`. После возврата ноды повторный +`make rebuild-storage` восстановил мир. + +**Прототип 21 августа 2026 года при исследовании #110.** Предшествующая +shell-версия узкой пересборки за 2 м 25 с пересоздала прикладные базы, отправила +401 185 событий и 6 656 версий заказов и вернула позицию с 11 на 8. Замер +оставлен как ориентир цены, но не подтверждает нынешнюю реализацию Airflow: тот +вариант не очищал Kafka и был отклонён как второй оркестратор мира. + +**Перезамер 20 августа 2026 года при исполнении #96, до переезда в Airflow.** +`make up` после `make clean` прошёл за 8 м 18 с, повтор на живом стенде — за +3 м 35 с. Отправка семи слепков заняла около 28 секунд. + Прежний замер ниже — 2 м 50 с и 1 м 5 с до появления приёма заказов и пульта мира. Весь рост приписывать слепкам нельзя: на холодном прогоне основное время ушло ещё до сценариев ожидания: `airflow-init` завершился примерно на 320-й diff --git a/docs/specs/2026-07-30-stand-v2-realism.md b/docs/specs/2026-07-30-stand-v2-realism.md index ec2acd2..1b34496 100644 --- a/docs/specs/2026-07-30-stand-v2-realism.md +++ b/docs/specs/2026-07-30-stand-v2-realism.md @@ -365,11 +365,22 @@ CSV в репозитории (`data/catalog/products.csv`: `sku`, `name`, `cate — через `FINAL` или равносильный выбор последней версии; батчевая переобработка нижележащих объектов — по дневным партициям (`DROP/REPLACE PARTITION ON CLUSTER`); `TRUNCATE ... ON CLUSTER` в конвейере не - применяется вовсе — полный сброс стенда делается `make clean && make up`, то - есть вместе с томами. `DROP/REPLACE PARTITION` работает только по - **локальным** таблицам ON CLUSTER, не по Distributed; замена через - DROP+INSERT неатомарна — дашборд в середине прогона честно моргает (это - осознанная цена, не баг). + применяется вовсе: обычная переобработка ограничена партицией, а полное + восстановление данных — базами слоёв в следующем пункте. `DROP/REPLACE + PARTITION` работает только по **локальным** таблицам ON CLUSTER, не по + Distributed; замена через DROP+INSERT неатомарна — дашборд в середине прогона + честно моргает (это осознанная цена, не баг). +- **Область очистки следует за областью состояния**: обычная разработка не + требует пустых томов. Сначала пересоздаётся затронутый объект, партиция или + база слоя; если нужен целиком чистый мир, пересоздаются прикладные базы + ClickHouse, топики и позиции их потребителей Kafka, затем заново заливаются + данные. Сами службы и их тома сохраняются. `make clean` нужен, только когда + требуется удалить состояние вне прикладного мира либо проверить запуск именно + с пустых томов. Перед таким сбросом агент называет это состояние и объясняет, + почему более узкой очистки недостаточно. «Так проще и надёжнее» не считается + обоснованием. Обязательный `make clean` перед каждой приёмкой отвергнут: он + стирает следы неисправности и превращает редкий холодный запуск в постоянную + цену разработки. - **Поздние заказы поглощает ODS** как новую версию `order_id`. Как их подхватывают материализованные объекты DDS и DM, решается вместе с их моделью, а не при проектировании приёма. @@ -607,9 +618,11 @@ ADR 0002 с приёмочными совпадают почти целиком. Что гонять на приёмке этапа — строка в [карте проверок](../architecture/testing.md), раздел «Какую проверку когда -запускать»: стенд собирается с нуля и на нём зелены все три цели, которым -нужен стенд. Это гигиена, а не приёмка: она обещает, что этап не сломал стенд, -и молчит о его предмете — дашборд может быть не нарисован, а все три останутся +запускать»: на стенде зелены все три цели, которым он нужен, а состояние перед +проверкой восстанавливается в самой узкой достаточной области. Запуск с пустых +томов обязателен только для изменения начальной сборки или постоянного состояния +служб, а не как общая гигиена этапа. Эти цели обещают, что этап не сломал стенд, +и молчат о его предмете — дашборд может быть не нарисован, а все три останутся зелёными. Свои наблюдаемые критерии каждый этап получает при разбиении в /to-tickets. Документация правится в PR этапа (правило AGENTS.md). diff --git a/docs/specs/2026-08-01-generator.md b/docs/specs/2026-08-01-generator.md index c2434bf..6e780b5 100644 --- a/docs/specs/2026-08-01-generator.md +++ b/docs/specs/2026-08-01-generator.md @@ -326,10 +326,10 @@ - **Обрыв любого режима — переигровка дня целиком**; дедуп склеивает повторы: `WatchID` детерминированы, повтор — та же строка для ReplacingMergeTree. -- **Эталонный снимок при старте стенда — через Kafka, пакетным режимом - проигрывателя.** Отдельный механизм заливки не строится: каждый `make up` - бесплатно прогоняет весь конвейер на настоящих данных, и строгий приём - хранилища проверяет контракт тем же прогоном. +- **Эталонный снимок при создании мира — через Kafka, пакетным режимом + проигрывателя.** Отдельный механизм заливки не строится: `world_initialize` + прогоняет весь конвейер на настоящих данных при первом `make up` и при явном + `make rebuild-storage`. На готовом мире повторный `make up` заливку пропускает. Оговорка «если заливка уйдёт в десятки минут — вернуться к прямой загрузке» проверена при фиксации чисел: 14 × 50 тыс. ≈ 700 тыс. событий — расчётно минута-две, запас есть. @@ -361,10 +361,11 @@ окном K = 7, дышащая зона и две волны недельной сезонности. Удлинение до месяца — дешёвый ход (пересборка описи), если понадобится. - **В git — только опись, снимок не хранится.** Снимок генерируется при - `make up` и при проверках: артефакт — кэш чистой функции, кэш в git не - хранят. Опись несёт паспорт мира, счётчики и хеши по дням; проверки + создании мира и в проверках генератора: артефакт — кэш чистой функции, кэш в + git не хранят. Опись несёт паспорт мира, счётчики и хеши по дням; проверки «пустой git diff» и «пересгенерируй день N — сравни хеш» живут на ней. - Каждый `make up` — живая демонстрация детерминизма. Честная потеря — + Первый `make up` пустого стенда — живая демонстрация детерминизма; готовый мир + повторный подъём не пересчитывает. Честная потеря — страховка на случай платформенного бага: раньше менти с расходящимися байтами мог взять готовый снимок из git, теперь он упрётся в красный чек описи; смягчение — CI гоняет генерацию на amd64 и arm64. @@ -402,9 +403,10 @@ Наблюдаемость без порогов остаётся и делает всю работу: проигрыватель печатает тайминги генерации и доставки раздельно, лаг живого дня — в его -логе, а `make up` после #42 гоняет генератор по восемь дней при каждом -подъёме стенда: замедлись он на порядок — подъём стенда встанет колом, и -заметит это первый же человек, который его поднял. +логе, а `world_initialize` гоняет генератор по восемь дней при первом создании +мира и явной пересборке. Замедлись он на порядок — операция встанет колом, и +это увидит первый же человек, который создаёт мир; повторный `make up` готовый +мир не пересчитывает. Отклонено с доводами: @@ -480,12 +482,16 @@ - **Позиция на оси времени** (хвост отдан пульту мира раньше этапа 5 — [ADR 0009](../adr/0009-world-control.md))**.** Проигрыватель состояния не хранит (раздел 9), поэтому вести позицию — работа того, кто его зовёт. Живёт - она переменной Airflow: отсутствие переменной означает мир в стартовом - состоянии, а ставит её тот даг пульта, который сыграл день, — значением - «последний сыгранный + 1» и только по успеху. + она переменной Airflow: отсутствие переменной означает, что стартовый мир + ещё не создан. В обычной жизни её ставит даг пульта, который сыграл день, — + значением «последний сыгранный + 1» и только по успеху. Даги жизненного цикла + используют `-1` для незавершённой очистки, `0` для чистого мира до генерации + и ставят первый несыгранный день только после приёма стартового мира + ([ADR 0013](../adr/0013-world-lifecycle.md)). Довод — генератор отдельная и заменяемая сущность, привязывать его к - хранилищу незачем, а `make clean` сносит том метаданных Airflow вместе с - данными ClickHouse, так что позиция и мир чистятся одной командой. + хранилищу незачем. Полный `make clean` сносит метаданные Airflow вместе с + данными ClickHouse, а узкая пересборка явно сводит сохранённую позицию с + заново залитым миром. Расхождение переменной с данными (менти почистил партицию руками) лечится документацией, а не сторожем: постоянная сверка была бы проверкой, красной в норме ([ADR 0002](../adr/0002-monitoring-scope.md)). diff --git a/docs/specs/2026-08-21-world-lifecycle.md b/docs/specs/2026-08-21-world-lifecycle.md new file mode 100644 index 0000000..3673798 --- /dev/null +++ b/docs/specs/2026-08-21-world-lifecycle.md @@ -0,0 +1,248 @@ +# Жизненный цикл мира: создание и пересоздание ведёт Airflow + +Дата: 21 августа 2026 года. Решение и доводы — [ADR 0013](../adr/0013-world-lifecycle.md); +здесь устройство, которое из него следует. + +## Задача + +До этой работы начало оси мира собирали трое. Compose создавал топики, применял +DDL и двумя одноразовыми службами заливал восемь дней. Shell-сценарии ждали, +пока события доедут до ODS, и разово дёргали даг приёма заказов. Airflow вёл +мир после стартовой заливки. + +Граница владения не совпадала с процессом, и это видно по следствиям. Успешная +отправка в Kafka выдавалась за созданный мир: `world-init` завершался сразу +после записи в топик, до приёма сообщений хранилищем. +Повторный `make up` смешивал подъём служб с повторной заливкой данных. Узкая +пересборка хранилища сохраняла в Kafka сообщения и офсеты прежнего мира, потому +что чинила только ClickHouse. Позиция пульта правилась снаружи Airflow — тем, +кто миром не управляет. + +**Чему учит переезд.** Менти видит один наблюдаемый граф, которому принадлежат +прикладная схема, данные мира и позиция модельного времени, — и Compose, +который держит только инфраструктуру. Зелёный `world_initialize` значит ровно +одно: схема создана, оба источника приняты, следующий день известен. + +## Решение + +Граница владения одна: **Compose отвечает за инфраструктуру; Airflow — за +прикладной мир.** + +Compose держит долгоживущие службы, сети, тома, объявленные файлом учётные +записи ClickHouse, инициализацию метаданных Airflow и инициализацию Superset. + +Airflow держит прикладные топики Kafka, канонический DDL прикладных баз `stg`, +`ods`, `dds`, `dm`, `dic`, данные стартового мира, их приём и переменную +`world_position`. + +Одноразовых служб `kafka-init`, `clickhouse-init`, `world-init` и +`world-snapshots` в графе Compose не остаётся. Ни профиля, ни вызова через +`compose run` им не оставляют: спрятанная за профилем служба — тот же второй +оркестратор, только менее заметный. + +`make` инфраструктурой и ограничивается: поднимает стенд, зовёт нужный даг +через REST API Airflow и ждёт результата. Shell больше не применяет DDL, не +запускает генератор и не правит позицию. + +### Учётная запись жизненного цикла + +Прикладную структуру создаёт и удаляет отдельная учётная запись `lifecycle` с +ролью `lifecycle_owner`, объявленная тем же файлом, что остальные учётные записи +([ADR 0007](../adr/0007-clickhouse-access.md)). Ей разрешено администрировать +прикладные базы — заводить и удалять их и объекты внутри, — читать их и +выполнять кластерный DDL. Ей разрешено чтение `system.settings`: без него +`clickhouse-connect` не открывает соединение. Административные запросы к +системным базам ей запрещены. + +`etl` при этом не расширяется. Право удаления баз нужно редкой операции +восстановления и остаётся у `lifecycle`. Граница проверяема: при лишнем запросе +менти увидит в ошибке имя учётной записи и недостающее право. + +### Что Airflow получает от репозитория + +Каталог `/opt/airflow/world` — то, что нужно для создания мира: `generator/`, +`data/` и корневой `.dockerignore` смонтированы только для чтения. Из первого +собирается канонический образ генератора, из второго читается опись мира, а +последний не пускает в контекст сборки локальные окружения и кэши. Запросы +стенда приезжают прежним монтированием `sql/`. + +Меняющиеся API сверены через Context7 и затем с установленными пакетами +стенда. Выбраны публичный REST API v2 и ждущий `TriggerDagRunOperator` Airflow, +сборка образа через `DockerHook.get_conn().build()`, futures `AdminClient` для +Kafka и по одному `clickhouse-connect.command()` на выражение DDL. Последний +выбор требует явного разбора файлов: клиент сам многозапросный файл не делит, +а в каноническом SQL конец выражения уже задан точкой с запятой в конце строки. + +## Контракты дагов + +Оба дага публичные: без расписания, не на паузе, `max_active_runs = 1`. Их +запускают руками из интерфейса и командой `make` через REST API. + +### `world_initialize` — создать мир, если его нет + +Даг безопасен по построению: на готовом мире он ничего не меняет. + +1. Читает `world_position` и решает, что делать дальше: инициализировать, + успешно завершиться без изменений или покраснеть. Разбор состояний — в + таблице ниже. +2. Создаёт топики: `hits` с двумя партициями, `orders` с одной. Топики идут + **до** DDL: чтец ClickHouse начинает читать топик, когда над ним появляется + матвью, и топик, созданный брокером сам, приехал бы с одним разделом + ([ADR 0008](../adr/0008-order-ingestion.md)). +3. Применяет канонический DDL из `sql/ddl/` по порядку имён файлов, с ноды 1, + учётной записью `lifecycle`. Правило «матвью приёма создаётся последней» + остаётся за раскладкой файлов и не переезжает в даг. +4. Собирает канонический образ генератора. Иначе мир мог бы приехать из старого + кода: сборкой образа при подъёме больше никто не занимается. +5. Играет стартовые дни одним прогоном `batch` того же образа. +6. Ждёт, пока события доедут до ODS. Условие ожидания то же, что у прежнего + сценария, и оно точное: сумма строк `ods.event_dist` за даты описи и строк + `ods.event_errors_dist` не меньше суммы событий из + [`data/world-inventory.json`](../../data/world-inventory.json). «Что-то + приехало» условием не считается. +7. Отправляет слепки заказов вторым прогоном образа — со сдвигом, дни 0…6 + ([заказы в стартовом мире](../architecture/orders/start-world.md)). +8. Дёргает `orders_ingest` ждущим триггером и зеленеет только вслед за ним. +9. Последней успешной задачей ставит позицию в `WORLD_STARTING_DAYS`. + +Порядок задач связывает проверки: каждый шаг подтверждён предыдущим, а красная +задача называет место остановки. + +### `world_recreate` — снести мир и создать заново + +Сначала даг выполняет всю разрушительную часть, затем делегирует создание мира. + +1. Ставит позицию в `-1`: прежний мир объявлен недействительным раньше, чем с + ним что-то произошло. +2. Удаляет прикладные базы ClickHouse `stg`, `dm`, `dds`, `ods`, `dic` — первым + делом, чтобы чтецы Kafka перестали существовать вместе с ними. +3. Удаляет позиции групп потребителей `clickstream_hits` и + `clickstream_orders`, затем топики `hits` и `orders`. +4. Ставит позицию в `0` — только после успешной очистки. +5. Дёргает `world_initialize` ждущим триггером и ждёт его результата. + +**Общего модуля с шагами двух дагов нет.** Переиспользование здесь — это +делегирование: пересборка вызывает создание мира целиком. Такой модуль стал бы +третьим местом с той же последовательностью, а первое же расхождение между +дагами пришлось бы прятать в его параметрах. + +### Сторожа у работников мира + +`world_next_day` и `world_live_day` краснеют, если позиции нет либо она меньше +`WORLD_STARTING_DAYS`. Отказ называет прочитанное значение и следующий шаг: +поднять стенд или пересобрать хранилище. Выключатель `world_live` своего +сторожа не получает — он краснеет вслед за работником. + +`orders_ingest` сторожа не получает вовсе: его зовёт сама инициализация, когда +готового мира ещё нет. + +## Состояния позиции + +`world_position` остаётся единственным состоянием пульта. + +| Позиция | `world_initialize` | `world_recreate` | Работники | +|---|---|---|---| +| переменной нет | создаёт мир | очищает и создаёт | красный отказ | +| `-1` | красный отказ: очистка не завершена, повторите `make rebuild-storage` | очищает и создаёт | красный отказ | +| `0` | создаёт мир | очищает и создаёт | красный отказ | +| `WORLD_STARTING_DAYS` и больше | мир уже есть: пишет это в журнал и зеленеет | очищает и создаёт | работают | + +Значения между `1` и `WORLD_STARTING_DAYS` штатно не возникают: позиция +становится либо `-1`, либо `0`, либо номером сыгранного дня за стартовым миром. +Попав на такое значение, `world_initialize` краснеет тем же отказом, что на +`-1`, и называет прочитанное число. + +## Поведение при отказах + +**Пересборка упала на очистке.** Позиция остаётся `-1`. Обычный `make up` эту +поломку не заглаживает: `world_initialize` краснеет и говорит, что мир +недействителен и пересборку надо повторить. Стенд обязан назвать незавершённую +очистку вслух и не продолжать работу молча. + +**Очистка прошла, потомок не доработал.** Позиция остаётся `0`. Инициализация +детерминирована, поэтому следующий `make up` просто пробует её заново — с +того же места и с тем же результатом. + +**Позицию в `WORLD_STARTING_DAYS` ставит только успех обоих источников.** +События доехали, слепки приняты, `orders_ingest` зелёный — тогда и только тогда +мир считается созданным. + +**Порядок очистки Kafka входит в контракт.** Пока живы чтецы +ClickHouse, они состоят в группах потребителей, и удаление группы недопустимо. +Поэтому сначала удаляются базы, затем топики. + +**Позицию удалили руками на живом мире.** Даг честно зальёт мир заново: сырьё +в STG удвоится, ODS схлопнет повтор дедупликацией. Условие ожидания на непустом +ODS истинно сразу — этим оно и ограничено, обещание «доехала именно эта +заливка» ни один счёт дать не может. Сторожа на этот случай нет: ошибку менти +делает себе сам и тут же видит. + +**Слоты Airflow.** Их четыре, и вложенные ожидания в них укладываются, пока +пересборка не наложилась на движение мира: `world_recreate` ждёт +`world_initialize`, тот ждёт `orders_ingest` — три слота из четырёх. Одновременно +работающий выключатель добавляет свои и загоняет стенд в тупик ожиданий. +Поэтому пересборка — явная операция обслуживания: оператор ставит `world_live` +на паузу и даёт работникам доиграть. Предусловие остаётся в документации; +механизм блокировок стоил бы дороже этой строки. + +## Границы + +- **`make up` не мигратор схемы.** На готовом мире изменённый DDL не + применяется. Обновляет схему явная пересборка `world_recreate` либо + адресная правка объекта в разработке. Размен принят с открытыми глазами. +- **Ни блокировок, ни статуса, ни автоматической паузы.** Предусловие + обслуживания названо словами. +- **`etl` не расширяется**, `orders_ingest` не получает сторожа. +- **`make clean` сохраняет свой смысл**: полное удаление постоянного состояния + стенда. Ни в пересборку, ни в обычную проверку он не входит. +- **`make generate-batch` и `make generate-live` остаются** — с них снимаются + унаследованные зависимости от исчезающих одноразовых служб. +- Содержание DDL, код генератора и устройство слоёв не меняются. + +## Проверка + +Цели `make` — обычные: `config-test` и `lint` без стенда, `smoke`, +`check-clickhouse` и `check-services` на поднятом. Сверх них проверяются +сценарии, которых у стенда раньше не было. + +- **Повторный `make up` на живом мире** проходит быстро и ничего не меняет: + позиция та же, счётчики `check-clickhouse` те же, в журнале дага сказано, что + мир уже есть. +- **`make rebuild-storage`** возвращает канонический мир: позиция проходит + `-1` → `0` → `WORLD_STARTING_DAYS`, в ODS ровно даты описи, топики созданы + заново, офсеты групп идут с нуля. Цена прогона называется числом и сверяется + с прежними 2 м 25 с. +- **Красные пути проверяются своими отказами.** Сторож работника при + отсутствующей, нулевой и отрицательной позиции; `world_initialize` на `-1`; + оборванная очистка, оставляющая `-1`. Проверка, не видевшая своей поломки, + доказывает только то, что умеет печатать «ЗЕЛЁНО». +- **Отказ в правах читается.** Учётная запись `lifecycle` применяет весь DDL + без единой ошибки прав, а административный запрос к системной базе ей + отклоняют — и текст отказа называет учётку. +- **Один холодный `make clean && make up`** оправдан на итоговой приёмке и + только этой работой: она меняет самый первый подъём стенда, а его иначе не + увидеть. Обобщать этот случай не нужно — обычная проверка по-прежнему + обходится узкой областью очистки. + +Замеры холодного и повторного `make up` и пересборки записываются в +[карту проверок](../architecture/testing.md), раздел «Что проверено». + +## Отклонённые варианты + +- **Спрятать одноразовые службы за профилем или звать их `compose run`.** + Дешевле всего в правках, но оставляет двух оркестраторов одного процесса — + ровно то, ради чего затевался переезд. +- **Оставить применение DDL и заливку мира в shell, а Airflow отдать только + позицию.** Тогда «один наблюдаемый процесс» остаётся обещанием: половина + графа живёт в журнале терминала. +- **Вынести общие шаги двух дагов в отдельный модуль.** Третье место для той же + последовательности; расхождение дагов пряталось бы в его параметрах. +- **Применять изменённый DDL при каждом `make up`.** Это мигратор схемы; ему + нужен отдельный разговор, в граф создания мира он не помещается. +- **Выдать права на удаление баз учётной записи `etl`.** Отвергнуто ещё + ADR 0013: конвейер не получает административные права из-за редкой операции. +- **Завести блокировку или статус обслуживания, чтобы пересборка не + сталкивалась с движением мира.** Механизм на редкий ручной случай дороже + предусловия, названного словами. +- **Сторож на `orders_ingest`.** Даг зовут до того, как мир существует; сторож + запретил бы саму инициализацию. 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..9dde04e 100755 --- a/scripts/config-test.sh +++ b/scripts/config-test.sh @@ -57,6 +57,18 @@ 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["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 подключение, и она же одна читает топик.