feat(airflow): пробные DAG test_clickhouse и test_kafka вместо демонстрационного

Зачем. Демонстрационный DAG example_clickstream_hello ничего не проверял: он не обращался ни к ClickHouse, ни к Kafka, поэтому его зелёный результат ничего не говорил о стенде. Пробники проверяют связи по-настоящему — и тем же клиентом, каким будут ходить рабочие DAG. Что. - test_clickhouse: пишет строку в ReplicatedMergeTree на ноде 1 и читает её с ноды 2 через Distributed. Данные проходят путь «нода 2 → все шарды → шард ноды 1», то есть проверяется межшардовое чтение, а не одна нода. - test_kafka: пишет в постоянный топик сообщение с меткой прогона и вычитывает его обратно. - infra/airflow/Dockerfile: clickhouse-connect 1.6.0 и confluent-kafka 2.15.0 вшиты в образ, импорт проверяется на сборке — при запуске контейнера пакеты не доустанавливаются. - Проверки: scripts/stand-smoke.sh гоняет оба пробника через API Airflow, scripts/config-test.sh разбирает DAG без стенда, tests/stand-smoke-guards.sh проверяет красный путь, tests/dag-probes-unit.py — модульные проверки разбора. - README и ADR 0001 обновлены тем же изменением. - Удалён dags/example_clickstream_hello.py. Проверка. make config-test — пройдено 3, 3 и 6, ошибок 0. make clean; cp .env.example .env; make up — 116 с на чистых томах. make smoke — пройдено 25, ошибок 0; стенд занимает 2244,0 MiB. make smoke-cluster — все 8 проверок кластера. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-31 14:04:19 +03:00
parent d3bc2f7c18
commit 64f3418378
12 changed files with 799 additions and 109 deletions
@@ -1,34 +0,0 @@
-"""Пример независимого DAG для проверки Airflow 3."""
-
-from __future__ import annotations
-
-import datetime
-
-from airflow.sdk import dag, task
-
-
-@dag(
-    dag_id="example_clickstream_hello",
-    schedule=None,
-    start_date=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
-    catchup=False,
-    tags=["пример"],
-    doc_md=__doc__,
-)
-def example_clickstream_hello():
-    @task
-    def extract() -> dict[str, int]:
-        return {"clicks": 3, "views": 10}
-
-    @task
-    def calculate_ctr(counters: dict[str, int]) -> float:
-        return round(counters["clicks"] / counters["views"], 3)
-
-    @task
-    def show_result(ctr: float) -> None:
-        print(f"CTR учебного примера: {ctr}")
-
-    show_result(calculate_ctr(extract()))
-
-
-example_clickstream_hello()
@@ -0,0 +1,215 @@
+"""Сквозная проверка подключения Airflow к кластеру ClickHouse."""
+
+from __future__ import annotations
+
+import datetime
+import uuid
+
+from airflow.sdk import Connection, dag, get_current_context, task
+
+CLUSTER = "clickstream_cluster"
+LOCAL_TABLE = "airflow_probe_local"
+DISTRIBUTED_TABLE = "airflow_probe_distributed"
+
+
+def _table_engines(client, *, query_node_2: bool) -> list[tuple[str, str]]:
+    source = (
+        "remote('clickhouse-02:9000', system.tables)"
+        if query_node_2
+        else "system.tables"
+    )
+    result = client.query(
+        f"""
+        SELECT name, engine
+        FROM {source}
+        WHERE database = 'default'
+          AND name IN ('{LOCAL_TABLE}', '{DISTRIBUTED_TABLE}')
+        ORDER BY name
+        """
+    )
+    return result.result_rows
+
+
+def _drop_tables(client) -> None:
+    client.command(
+        f"DROP TABLE IF EXISTS default.{DISTRIBUTED_TABLE} "
+        f"ON CLUSTER {CLUSTER} SYNC"
+    )
+    client.command(
+        f"DROP TABLE IF EXISTS default.{LOCAL_TABLE} ON CLUSTER {CLUSTER} SYNC"
+    )
+
+
+def _assert_tables_absent(client) -> None:
+    for query_node_2, node_name in ((False, "ноде 1"), (True, "ноде 2")):
+        remaining = _table_engines(client, query_node_2=query_node_2)
+        if remaining:
+            raise RuntimeError(
+                f"служебные таблицы остались на {node_name}: {remaining}"
+            )
+
+
+def _assert_marker_path(
+    *,
+    local_rows: list[tuple[str, str]],
+    distributed_rows: list[tuple[int, str, str]],
+    node_2_hostname: str,
+    marker: str,
+) -> None:
+    if len(local_rows) != 1 or local_rows[0][1] != marker:
+        raise RuntimeError(f"маркер не найден в локальной таблице: {marker}")
+    node_1_hostname = local_rows[0][0]
+    if node_1_hostname == node_2_hostname:
+        raise RuntimeError("запись и чтение маркера должны выполняться с разных нод")
+    expected_rows = [(1, node_1_hostname, marker)]
+    if distributed_rows != expected_rows:
+        raise RuntimeError(
+            "нода 2 не прочитала маркер первого шарда через Distributed: "
+            f"{marker}, получено {distributed_rows}"
+        )
+
+
+def _cleanup_clickhouse_client(client, original_error: BaseException | None) -> None:
+    cleanup_errors: list[Exception] = []
+    try:
+        _drop_tables(client)
+        _assert_tables_absent(client)
+    except Exception as error:
+        cleanup_errors.append(error)
+    try:
+        client.close()
+    except Exception as error:
+        cleanup_errors.append(error)
+
+    if original_error is not None:
+        for error in cleanup_errors:
+            original_error.add_note(
+                f"Дополнительная ошибка очистки ClickHouse: {error}"
+            )
+        return
+    if len(cleanup_errors) == 1:
+        raise cleanup_errors[0]
+    if cleanup_errors:
+        raise ExceptionGroup("ошибки очистки ClickHouse", cleanup_errors)
+
+
+@dag(
+    dag_id="test_clickhouse",
+    schedule=None,
+    start_date=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
+    catchup=False,
+    tags=["проверка"],
+    doc_md=__doc__,
+)
+def test_clickhouse():
+    @task
+    def check_cluster_path() -> None:
+        import clickhouse_connect
+
+        connection = Connection.get("clickhouse_default")
+        client = clickhouse_connect.get_client(
+            host=connection.host,
+            port=connection.port,
+            username=connection.login or "default",
+            password=connection.password or "",
+            database=connection.schema or "default",
+            connect_timeout=5,
+            send_receive_timeout=30,
+        )
+        marker = f"{get_current_context()['run_id']}:{uuid.uuid4()}"
+        expected_tables = [
+            (DISTRIBUTED_TABLE, "Distributed"),
+            (LOCAL_TABLE, "ReplicatedMergeTree"),
+        ]
+        probe_error = None
+
+        try:
+            _drop_tables(client)
+            _assert_tables_absent(client)
+            client.command(
+                f"""
+                CREATE TABLE default.{LOCAL_TABLE} ON CLUSTER {CLUSTER}
+                (
+                    marker String
+                )
+                ENGINE = ReplicatedMergeTree(
+                    '/clickhouse/tables/{{shard}}/{LOCAL_TABLE}',
+                    '{{replica}}'
+                )
+                ORDER BY marker
+                """
+            )
+            client.command(
+                f"""
+                CREATE TABLE default.{DISTRIBUTED_TABLE} ON CLUSTER {CLUSTER}
+                AS default.{LOCAL_TABLE}
+                ENGINE = Distributed(
+                    '{CLUSTER}',
+                    'default',
+                    '{LOCAL_TABLE}',
+                    cityHash64(marker)
+                )
+                """
+            )
+
+            for query_node_2, node_name in ((False, "ноде 1"), (True, "ноде 2")):
+                actual_tables = _table_engines(
+                    client,
+                    query_node_2=query_node_2,
+                )
+                if actual_tables != expected_tables:
+                    raise RuntimeError(
+                        f"неверный набор таблиц на {node_name}: {actual_tables}"
+                    )
+
+            client.insert(
+                f"default.{LOCAL_TABLE}",
+                [[marker]],
+                column_names=["marker"],
+            )
+            local_rows = client.query(
+                f"""
+                SELECT hostName(), marker
+                FROM default.{LOCAL_TABLE}
+                WHERE marker = {{marker:String}}
+                """,
+                parameters={"marker": marker},
+            ).result_rows
+            node_2_rows = client.query(
+                """
+                SELECT hostName()
+                FROM remote('clickhouse-02:9000', system.one)
+                """
+            ).result_rows
+            if len(node_2_rows) != 1:
+                raise RuntimeError(
+                    f"не удалось определить имя ноды 2: {node_2_rows}"
+                )
+            distributed_rows = client.query(
+                f"""
+                SELECT _shard_num, hostName(), marker
+                FROM remote(
+                    'clickhouse-02:9000',
+                    'default',
+                    '{DISTRIBUTED_TABLE}'
+                )
+                WHERE marker = {{marker:String}}
+                """,
+                parameters={"marker": marker},
+            ).result_rows
+            _assert_marker_path(
+                local_rows=local_rows,
+                distributed_rows=distributed_rows,
+                node_2_hostname=node_2_rows[0][0],
+                marker=marker,
+            )
+        except BaseException as error:
+            probe_error = error
+            raise
+        finally:
+            _cleanup_clickhouse_client(client, probe_error)
+
+    check_cluster_path()
+
+
+test_clickhouse()
@@ -0,0 +1,98 @@
+"""Проверка записи и чтения сообщения Airflow через Kafka."""
+
+from __future__ import annotations
+
+import datetime
+import time
+import uuid
+
+from airflow.sdk import dag, get_current_context, task
+
+BROKER = "kafka:9092"
+TOPIC = "airflow_integration_probe"
+
+
+@dag(
+    dag_id="test_kafka",
+    schedule=None,
+    start_date=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
+    catchup=False,
+    tags=["проверка"],
+    doc_md=__doc__,
+)
+def test_kafka():
+    @task
+    def check_round_trip() -> None:
+        from confluent_kafka import Consumer, Producer, TopicPartition
+
+        marker = f"{get_current_context()['run_id']}:{uuid.uuid4()}"
+        marker_bytes = marker.encode()
+        group_id = f"airflow-probe-{uuid.uuid4()}"
+        producer = None
+        consumer = None
+
+        try:
+            # Стенд опирается на автосоздание: срок хранения чистит записи, а
+            # не топик; в бою автосоздание обычно выключают.
+            delivery_errors: list[str] = []
+            delivered_offsets: list[tuple[int, int]] = []
+
+            def on_delivery(error, message) -> None:
+                if error is not None:
+                    delivery_errors.append(str(error))
+                else:
+                    delivered_offsets.append((message.partition(), message.offset()))
+
+            producer = Producer(
+                {
+                    "bootstrap.servers": BROKER,
+                    "client.id": "airflow-test-kafka",
+                    "message.timeout.ms": 10_000,
+                    "socket.timeout.ms": 5_000,
+                }
+            )
+            producer.produce(
+                TOPIC,
+                key=marker_bytes,
+                value=marker_bytes,
+                on_delivery=on_delivery,
+            )
+            undelivered = producer.flush(10)
+            if undelivered or delivery_errors or len(delivered_offsets) != 1:
+                raise RuntimeError(
+                    "Kafka не подтвердила запись маркера: "
+                    f"не доставлено {undelivered}, ошибки {delivery_errors}, "
+                    f"смещения {delivered_offsets}"
+                )
+
+            consumer = Consumer(
+                {
+                    "bootstrap.servers": BROKER,
+                    "group.id": group_id,
+                    "enable.auto.commit": False,
+                    "session.timeout.ms": 6_000,
+                    "socket.timeout.ms": 5_000,
+                }
+            )
+            partition, offset = delivered_offsets[0]
+            consumer.assign([TopicPartition(TOPIC, partition, offset)])
+            read_deadline = time.monotonic() + 20
+            while time.monotonic() < read_deadline:
+                message = consumer.poll(0.5)
+                if message is None:
+                    continue
+                if message.error():
+                    raise RuntimeError(f"Kafka вернула ошибку чтения: {message.error()}")
+                if message.value() == marker_bytes:
+                    return
+            raise RuntimeError(f"Kafka не вернула свой маркер за 20 секунд: {marker}")
+        finally:
+            if producer is not None:
+                producer.flush(1)
+            if consumer is not None:
+                consumer.close()
+
+    check_round_trip()
+
+
+test_kafka()