diff --git a/AGENTS.md b/AGENTS.md index 99ac974..84512a5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -8,6 +8,11 @@ Стенд строится по спеке [«Боевой реализм стенда (v2)»](docs/specs/2026-07-30-stand-v2-realism.md). +Отсюда вопрос к любой доработке — коду, проверке, документу: **чему на ней +научится менти?** Ответ называется одной фразой до того, как писать. Не +складывается фраза — это вопрос владельцу, а не строчка кода: проверки в этом +репозитории однажды уже переросли продукт. + ## Кому что поручать Читатель здесь не побочный потребитель, а тот, ради кого стенд существует: diff --git a/Makefile b/Makefile index 6ce846a..c2b32f7 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ COMPOSE ?= docker compose -.PHONY: up down clean ps logs config-test lint typecheck test docs smoke smoke-cluster smoke-guards +.PHONY: up down clean ps logs config-test lint typecheck test docs smoke smoke-cluster up: $(COMPOSE) up --detach --build --wait --wait-timeout 600 @@ -19,7 +19,6 @@ logs: config-test: COMPOSE_BIN="$(COMPOSE)" ./scripts/config-test.sh - ./tests/stand-smoke-static.sh lint: cd generator && uv run ruff check && uv run ruff format --check @@ -39,7 +38,3 @@ smoke: smoke-cluster: ./scripts/clickhouse-smoke.sh - -smoke-guards: - ./tests/smoke-guards.sh - COMPOSE_BIN="$(COMPOSE)" ./tests/stand-smoke-guards.sh diff --git a/README.md b/README.md index 64a3285..8af8a43 100644 --- a/README.md +++ b/README.md @@ -44,6 +44,12 @@ make up make smoke ``` +`.env.example` — справочник, а не настройка: в нём перечислены все переменные, +которые читает `compose.yaml`, с теми же значениями по умолчанию. Стенд его не +читает и на согласованность не проверяет, поэтому расхождение с `compose.yaml` +обнаружит только читатель. Меняя подстановку `${VAR:-значение}` в +`compose.yaml`, поправьте образец тем же коммитом. + Учётные данные Postgres и Grafana применяются при создании их томов. После первого запуска меняйте их только вместе с `make clean`: команда удалит все локальные данные стенда, а следующий `make up` создаст их с новыми @@ -56,10 +62,10 @@ make smoke подготавливает администратора и подключение к `clickhouse-01`. Второй обновляет Superset, создаёт администратора и импортирует подключение к `clickhouse-02`. -`make smoke` проверяет согласованность `.env.example` с Compose, зависимости -машины, здоровье контейнеров, Kafka через порт машины, три цели Prometheus, -источник Grafana, компоненты Airflow, ручной запуск пробников `test_clickhouse` -и `test_kafka`, метаданные и подключение Superset. Первый пробник создаёт +`make smoke` проверяет зависимости машины, здоровье контейнеров, устройство +keeper, Kafka через порт машины, три цели Prometheus, источник Grafana, +компоненты Airflow, ручной запуск пробников `test_clickhouse` и `test_kafka`, +метаданные и подключение Superset. Первый пробник создаёт таблицы на обеих нодах и читает через `Distributed` на ноде 2 строку из локальной таблицы ноды 1. Второй пишет в Kafka и читает свой маркер. В конце проверка спрашивает у Docker, не убивало ли ядро что-нибудь в долгоживущих @@ -68,17 +74,29 @@ Superset, создаёт администратора и импортирует Временный топик проверки с машины и запуски DAG удаляются; постоянный топик пробника сохраняется, а старые записи чистит Kafka. +Одиннадцать проверок здоровья сразу после `make up --wait` повторяют то, чего +Compose уже дождался: у каждой долгоживущей службы есть своя `healthcheck`. +Оставлены они потому, что первый вопрос к стенду всё равно «всё ли живо», а +ответ на него стоит меньше секунды. Устройство keeper — другое дело: он должен +работать от пользователя `clickhouse`, с пределом в 262144 открытых файла и со +своим томом под данные. Всё это объявлено в `compose.yaml`, но здоровым keeper +выглядит и без этого, поэтому смоук спрашивает у живого контейнера, дошли ли +объявленные настройки до процесса. + `make smoke-cluster` запускает отдельную глубокую проверку ClickHouse: описание кластера, макросы, связь с keeper, `ReplicatedMergeTree`, `Distributed`, очередь распределённых DDL и очистку временных таблиц. -`make config-test` проверяет Compose, синтаксис Bash и Python, малые проверки -логики пробников и пробельные ошибки в diff без запуска стенда. +`make config-test` проверяет Compose, синтаксис Bash и Python и пробельные +ошибки в diff без запуска стенда. -`make smoke-guards` сначала проверяет аварийную семантику кластерной проверки, -а затем удаляет служебную таблицу пробника только на второй ноде и -останавливает Prometheus с Kafka. Общая проверка должна назвать Prometheus и -оба пробника, после чего завершиться с ошибкой. В конце стенд восстанавливается. +Правило, которое стоит держать в голове, правя любую из этих проверок: +**проверка, которая не умеет краснеть, бесполезна.** Проверка, никогда не +видевшая своей поломки, доказывает только то, что она умеет печатать «ЗЕЛЁНО». +Убедиться дешевле всего руками: сломайте то, что она стережёт — остановите +`prometheus`, удалите служебную таблицу пробника на второй ноде, — и посмотрите, +покраснеет ли прогон и назовёт ли виновника. Не покраснел — проверка не +работает, и чинить надо её, а не стенд. ### Какую проверку когда запускать @@ -94,15 +112,8 @@ Superset, создаёт администратора и импортирует между службами, а не отдельные файлы: это интеграционная проверка. - `make smoke-cluster` — около минуты. Одна связь, зато до дна: межнодовое устройство ClickHouse. -- `make smoke-guards` — около пяти минут, и отвечает на другой вопрос. Не - «работает ли стенд», а «умеют ли проверки падать»: она намеренно ломает стенд - и смотрит, покраснеет ли `make smoke` и назовёт ли виновника, потом чинит и - убеждается, что стенд снова зелёный. Отсюда и три прогона `make smoke` - внутри — до поломки, во время неё и после починки. -Обычный рабочий цикл — `make config-test` и `make smoke`. `make smoke-guards` -нужна тому, кто правит сами проверки или пробники: без неё легко завести -проверку, которая зелена всегда. +Обычный рабочий цикл — `make config-test` и `make smoke`. Остановить контейнеры без удаления данных можно командой `make down`. Для полного сброса с удалением всех именованных томов используйте `make clean`. diff --git a/dags/test_clickhouse.py b/dags/test_clickhouse.py index 17d9808..89e3f03 100644 --- a/dags/test_clickhouse.py +++ b/dags/test_clickhouse.py @@ -42,9 +42,10 @@ NODES = ( def _clickhouse_client(): - # clickhouse_connect стоит только в образе Airflow, а малые проверки грузят - # этот модуль обычным интерпретатором, где пакета нет. Импорт верхнего - # уровня красит make config-test, поэтому он живёт здесь. + # clickhouse_connect импортируется внутри функции, а не наверху файла: + # обработчик DAG разбирает этот файл снова и снова, и импорт наверху + # оплачивался бы каждым разбором. Тяжёлые импорты Airflow советует + # держать внутри задач. import clickhouse_connect connection = Connection.get("clickhouse_default") @@ -82,6 +83,8 @@ def _drop_tables(client) -> None: ) +# Отсутствие таблиц проверяют обе задачи — перед созданием и после уборки, +# — поэтому у этой проверки своё имя, а остальные живут прямо в теле задач. def _assert_tables_absent(client) -> None: for node_name, source in NODES: remaining = _table_engines(client, source) @@ -91,32 +94,6 @@ def _assert_tables_absent(client) -> None: ) -def _assert_tables_created(client) -> None: - for node_name, source in NODES: - actual_tables = _table_engines(client, source) - if actual_tables != EXPECTED_TABLES: - raise RuntimeError( - f"неверный набор таблиц на {node_name}: {actual_tables}" - ) - - -def _assert_marker_path( - *, - distributed_rows: list[tuple[int, str, str]], - write_hostname: str, - node_2_hostname: str, - marker: str, -) -> None: - if write_hostname == node_2_hostname: - raise RuntimeError("запись и чтение маркера должны выполняться с разных нод") - expected_rows = [(1, write_hostname, marker)] - if distributed_rows != expected_rows: - raise RuntimeError( - "нода 2 не прочитала маркер первого шарда через Distributed: " - f"{marker}, получено {distributed_rows}" - ) - - @dag( dag_id="test_clickhouse", schedule=None, @@ -157,7 +134,12 @@ def test_clickhouse(): ) """ ) - _assert_tables_created(client) + for node_name, source in NODES: + actual_tables = _table_engines(client, source) + if actual_tables != EXPECTED_TABLES: + raise RuntimeError( + f"неверный набор таблиц на {node_name}: {actual_tables}" + ) finally: client.close() @@ -211,12 +193,16 @@ def test_clickhouse(): """, parameters={"marker": written["marker"]}, ).result_rows - _assert_marker_path( - distributed_rows=distributed_rows, - write_hostname=written["hostname"], - node_2_hostname=node_2_rows[0][0], - marker=written["marker"], - ) + if written["hostname"] == node_2_rows[0][0]: + raise RuntimeError( + "запись и чтение маркера должны выполняться с разных нод" + ) + expected_rows = [(1, written["hostname"], written["marker"])] + if distributed_rows != expected_rows: + raise RuntimeError( + "нода 2 не прочитала маркер первого шарда через Distributed: " + f"{written['marker']}, получено {distributed_rows}" + ) finally: client.close() diff --git a/dags/test_kafka.py b/dags/test_kafka.py index 812c2cc..7f081fd 100644 --- a/dags/test_kafka.py +++ b/dags/test_kafka.py @@ -74,48 +74,6 @@ class RecordAddress(NamedTuple): offset: int -def _assert_all_delivered(undelivered: int, marker: str) -> None: - if undelivered: - raise RuntimeError( - f"Kafka не приняла маркер за {FLUSH_TIMEOUT_SEC} с, " - f"не доставлено сообщений {undelivered}: {marker}" - ) - - -def _assert_no_delivery_errors(delivery_errors: list[str], marker: str) -> None: - if delivery_errors: - raise RuntimeError( - f"Kafka отказалась принять маркер {marker}: {delivery_errors}" - ) - - -def _assert_confirmed_once(addresses: list[RecordAddress], marker: str) -> None: - if len(addresses) != 1: - raise RuntimeError( - f"Kafka подтвердила запись маркера {marker} " - f"не одним сообщением: {addresses}" - ) - - -def _assert_message_arrived(message, marker: str) -> None: - if message is None: - raise RuntimeError( - f"Kafka молчала {READ_DEADLINE_SEC} с и не вернула маркер: {marker}" - ) - - -def _assert_no_read_error(message) -> None: - if message.error(): - raise RuntimeError(f"Kafka вернула ошибку чтения: {message.error()}") - - -def _assert_marker_matches(message, marker: str) -> None: - if message.value() != marker.encode(): - raise RuntimeError( - f"по адресу записи лежит не маркер запуска {marker}: {message.value()!r}" - ) - - @dag( dag_id="test_kafka", schedule=None, @@ -127,6 +85,10 @@ def _assert_marker_matches(message, marker: str) -> None: def test_kafka(): @task def write_marker() -> dict[str, str | int]: + # confluent_kafka импортируется внутри задачи, а не наверху файла: + # обработчик DAG разбирает этот файл снова и снова, и импорт наверху + # оплачивался бы каждым разбором. Тяжёлые импорты Airflow советует + # держать внутри задач. from confluent_kafka import Producer marker = f"{get_current_context()['run_id']}:{uuid.uuid4()}" @@ -158,9 +120,20 @@ def test_kafka(): finally: undelivered = producer.flush(FLUSH_TIMEOUT_SEC) - _assert_all_delivered(undelivered, marker) - _assert_no_delivery_errors(delivery_errors, marker) - _assert_confirmed_once(addresses, marker) + if undelivered: + raise RuntimeError( + f"Kafka не приняла маркер за {FLUSH_TIMEOUT_SEC} с, " + f"не доставлено сообщений {undelivered}: {marker}" + ) + if delivery_errors: + raise RuntimeError( + f"Kafka отказалась принять маркер {marker}: {delivery_errors}" + ) + if len(addresses) != 1: + raise RuntimeError( + f"Kafka подтвердила запись маркера {marker} " + f"не одним сообщением: {addresses}" + ) return { "marker": marker, "partition": addresses[0].partition, @@ -190,9 +163,17 @@ def test_kafka(): # другого нечем, поэтому у чтения обязан быть крайний срок. while message is None and time.monotonic() < deadline: message = consumer.poll(POLL_TIMEOUT_SEC) - _assert_message_arrived(message, marker) - _assert_no_read_error(message) - _assert_marker_matches(message, marker) + if message is None: + raise RuntimeError( + f"Kafka молчала {READ_DEADLINE_SEC} с и не вернула маркер: {marker}" + ) + if message.error(): + raise RuntimeError(f"Kafka вернула ошибку чтения: {message.error()}") + if message.value() != marker.encode(): + raise RuntimeError( + f"по адресу записи лежит не маркер запуска {marker}: " + f"{message.value()!r}" + ) finally: consumer.close() diff --git a/scripts/config-test.sh b/scripts/config-test.sh index 3f44108..fb4f1a5 100755 --- a/scripts/config-test.sh +++ b/scripts/config-test.sh @@ -48,20 +48,6 @@ if [[ "${#shell_files[@]}" -eq 0 ]]; then fi bash -n "${shell_files[@]}" PYTHONPYCACHEPREFIX="$CACHE_DIR" uv run --no-project python -m compileall -q "$ROOT_DIR/dags" -unit_status=0 -unit_output="$( - PYTHONPYCACHEPREFIX="$CACHE_DIR" \ - uv run --no-project python "$ROOT_DIR/tests/dag-probes-unit.py" 2>&1 -)" || unit_status=$? -printf '%s\n' "$unit_output" -if [[ "$unit_status" -ne 0 ]]; then - exit "$unit_status" -fi -if grep -Eq '^(Ran [0-9]+ tests|OK|FAILED)' <<<"$unit_output" || - ! grep -Eq '^ИТОГ: пройдено [0-9]+, ошибок 0$' <<<"$unit_output"; then - printf 'ОШИБКА: малые проверки пробников вывели итог не на русском языке.\n' >&2 - exit 1 -fi git -C "$ROOT_DIR" diff --check printf 'ЗЕЛЁНО: Compose, Bash, Python и пробельные ошибки diff проверены.\n' diff --git a/scripts/stand-smoke.sh b/scripts/stand-smoke.sh index 3d13b16..355689e 100755 --- a/scripts/stand-smoke.sh +++ b/scripts/stand-smoke.sh @@ -2,8 +2,6 @@ set -uo pipefail readonly ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -readonly COMPOSE_FILE="$ROOT_DIR/compose.yaml" -readonly ENV_EXAMPLE="$ROOT_DIR/.env.example" read -r -a COMPOSE_CMD <<<"${COMPOSE_BIN:-docker compose}" readonly -a LONG_LIVED_SERVICES=( clickhouse-keeper clickhouse-01 clickhouse-02 kafka postgres-metadata @@ -36,114 +34,6 @@ fail() { printf 'ОШИБКА: %s.\n' "$1" >&2 } -check_env_consistency() { - local LC_ALL=C - local content - local default - local depth - local env_value - local expression - local found_closing - local i - local inner - local joined - local j - local length - local line - local nested - local variable - local -A env_count=() - local -A env_values=() - local -a problems=() - local -A used=() - local -a expressions=() - - if [[ ! -r "$COMPOSE_FILE" || ! -r "$ENV_EXAMPLE" ]]; then - fail 'compose.yaml или .env.example недоступны для чтения' - return - fi - - # Разбираем только подстановки Compose и отдельно пропускаем $$ для команд контейнера. - while IFS= read -r line || [[ -n "$line" ]]; do - line="${line%$'\r'}" - if [[ "$line" =~ ^([A-Za-z_][A-Za-z0-9_]*)=(.*)$ ]]; then - variable="${BASH_REMATCH[1]}" - env_count["$variable"]=$(( ${env_count[$variable]:-0} + 1 )) - env_values["$variable"]="${BASH_REMATCH[2]}" - fi - done <"$ENV_EXAMPLE" - - content="$(<"$COMPOSE_FILE")" - length="${#content}" - for ((i = 0; i < length - 1; i++)); do - if [[ "${content:i:2}" == '$$' ]]; then - i=$((i + 1)) - continue - fi - if [[ "${content:i:2}" != '${' ]]; then - continue - fi - - depth=1 - found_closing=0 - nested=0 - for ((j = i + 2; j < length; j++)); do - if [[ "${content:j:2}" == '${' ]]; then - nested=1 - depth=$((depth + 1)) - j=$((j + 1)) - elif [[ "${content:j:1}" == '}' ]]; then - depth=$((depth - 1)) - if [[ "$depth" -eq 0 ]]; then - expressions+=("${content:i:j-i+1}") - found_closing=1 - i="$j" - break - fi - fi - done - if [[ "$found_closing" -eq 0 ]]; then - problems+=("незакрытая подстановка у позиции ${i}") - break - fi - if [[ "$nested" -eq 1 ]]; then - problems+=("${expressions[-1]}: вложенные подстановки запрещены, используйте простой вид \${VAR:-значение}") - fi - done - - for expression in "${expressions[@]}"; do - inner="${expression:2:${#expression}-3}" - if [[ "$inner" =~ ^([A-Za-z_][A-Za-z0-9_]*):-(.*)$ ]]; then - variable="${BASH_REMATCH[1]}" - default="${BASH_REMATCH[2]}" - used["$variable"]=1 - if [[ "${env_count[$variable]:-0}" -ne 1 ]]; then - problems+=("${variable}: нужна ровно одна строка в .env.example") - continue - fi - env_value="${env_values[$variable]}" - if [[ "$env_value" != "$default" ]]; then - problems+=("${variable}: значение '${env_value}' не равно '${default}'") - fi - else - problems+=("${expression}: нет значения по умолчанию вида :-") - fi - done - - for variable in "${!env_count[@]}"; do - if [[ -z "${used[$variable]+x}" ]]; then - problems+=("${variable}: не используется в compose.yaml") - fi - done - - if [[ "${#problems[@]}" -eq 0 ]]; then - pass '.env.example совпадает со всеми значениями по умолчанию compose.yaml' - else - printf -v joined '%s; ' "${problems[@]}" - fail "расхождение .env.example и compose.yaml: ${joined%; }" - fi -} - check_host_dependencies() { local command local -a missing=() @@ -184,6 +74,34 @@ check_container_health() { fi } +keeper_exec() { + timeout 20s "${COMPOSE_CMD[@]}" --project-directory "$ROOT_DIR" \ + exec -T clickhouse-keeper "$@" 2>/dev/null +} + +# Три настройки keeper объявлены в compose.yaml: свой пользователь, свой предел +# на открытые файлы и свой том под /var/lib/clickhouse. Проверка спрашивает у +# живого контейнера, дошли ли они до процесса — здоровым keeper выглядит и без +# них, а каталог координации, однажды созданный от чужого пользователя, +# следующий запуск уже не откроет. +check_keeper_runtime() { + local keeper_nofile + local keeper_owner + local keeper_user + + keeper_user="$(keeper_exec id -un)" + keeper_nofile="$(keeper_exec \ + awk '$1 == "Max" && $2 == "open" && $3 == "files" {print $4}' /proc/1/limits)" + keeper_owner="$(keeper_exec stat -c '%U:%G' /var/lib/clickhouse/coordination)" + if [[ "$keeper_user" == 'clickhouse' ]] && \ + [[ "$keeper_nofile" -ge 262144 ]] && \ + [[ "$keeper_owner" == 'clickhouse:clickhouse' ]]; then + pass 'keeper работает от clickhouse с nofile 262144 и своим каталогом данных' + else + fail "неверное окружение keeper: пользователь=${keeper_user:-нет ответа}, nofile=${keeper_nofile:-нет ответа}, владелец каталога=${keeper_owner:-нет ответа}" + fi +} + published_port() { local binding local service="$1" @@ -627,11 +545,11 @@ check_containers_survived() { fi } -check_env_consistency if check_host_dependencies; then for service in "${LONG_LIVED_SERVICES[@]}"; do check_container_health "$service" done + check_keeper_runtime check_kafka_from_host check_prometheus_targets check_grafana_datasource diff --git a/tests/dag-probes-unit.py b/tests/dag-probes-unit.py deleted file mode 100644 index 56b3031..0000000 --- a/tests/dag-probes-unit.py +++ /dev/null @@ -1,307 +0,0 @@ -"""Малые проверки логики пробников ClickHouse и Kafka без запуска Airflow.""" - -from __future__ import annotations - -import importlib.util -import sys -import types -import unittest -from pathlib import Path - - -class DeclaredTask: - """Заглушка объявленной задачи: держит только цепочку через `>>`.""" - - def __rshift__(self, other): - return other - - -def load_clickhouse_dag_tasks(): - airflow_module = types.ModuleType("airflow") - sdk_module = types.ModuleType("airflow.sdk") - captured_tasks = {} - - def dag(**_kwargs): - def decorate(function): - return function - - return decorate - - def task(function=None, **_kwargs): - def capture(target): - captured_tasks[target.__name__] = target - - def declare_task(*_args, **_kwargs): - return DeclaredTask() - - return declare_task - - return capture(function) if function is not None else capture - - sdk_module.Connection = object - sdk_module.dag = dag - sdk_module.get_current_context = lambda: {} - sdk_module.task = task - airflow_module.sdk = sdk_module - sys.modules["airflow"] = airflow_module - sys.modules["airflow.sdk"] = sdk_module - - dag_path = Path(__file__).resolve().parents[1] / "dags" / "test_clickhouse.py" - spec = importlib.util.spec_from_file_location("test_clickhouse_dag", dag_path) - if spec is None or spec.loader is None: - raise RuntimeError("не удалось загрузить модуль пробника ClickHouse") - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return module, captured_tasks - - -def load_kafka_dag_tasks(): - airflow_module = types.ModuleType("airflow") - sdk_module = types.ModuleType("airflow.sdk") - captured_tasks = {} - - def dag(**_kwargs): - def decorate(function): - return function - - return decorate - - def task(function): - captured_tasks[function.__name__] = function - - def declare_task(*_args, **_kwargs): - return None - - return declare_task - - sdk_module.dag = dag - sdk_module.get_current_context = lambda: {"run_id": "unit-test"} - sdk_module.task = task - airflow_module.sdk = sdk_module - sys.modules["airflow"] = airflow_module - sys.modules["airflow.sdk"] = sdk_module - - dag_path = Path(__file__).resolve().parents[1] / "dags" / "test_kafka.py" - spec = importlib.util.spec_from_file_location("test_kafka_dag", dag_path) - if spec is None or spec.loader is None: - raise RuntimeError("не удалось загрузить модуль пробника Kafka") - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return captured_tasks - - -class QueryResult: - def __init__(self, result_rows: list[tuple]) -> None: - self.result_rows = result_rows - - -class RecordingClient: - """Клиент ClickHouse, который запоминает запросы и отвечает заготовкой.""" - - def __init__(self, remaining_tables: list[tuple[str, str]] | None = None) -> None: - self.commands: list[str] = [] - self.queries: list[str] = [] - self.closed = False - self._remaining_tables = remaining_tables or [] - - def command(self, sql: str) -> None: - self.commands.append(" ".join(sql.split())) - - def query(self, sql: str, parameters=None) -> QueryResult: - self.queries.append(" ".join(sql.split())) - return QueryResult(list(self._remaining_tables)) - - def close(self) -> None: - self.closed = True - - -class ClickHouseProbeTests(unittest.TestCase): - @classmethod - def setUpClass(cls) -> None: - cls.module, cls.tasks = load_clickhouse_dag_tasks() - - def run_cleanup_task(self, client: RecordingClient) -> None: - original_client_factory = self.module._clickhouse_client - self.module._clickhouse_client = lambda: client - try: - self.tasks["cleanup_tables"]() - finally: - self.module._clickhouse_client = original_client_factory - - def test_marker_path_requires_different_nodes_and_first_shard(self) -> None: - marker = "свой-маркер" - self.module._assert_marker_path( - distributed_rows=[(1, "clickhouse-01-host", marker)], - write_hostname="clickhouse-01-host", - node_2_hostname="clickhouse-02-host", - marker=marker, - ) - - with self.assertRaisesRegex(RuntimeError, "разных нод"): - self.module._assert_marker_path( - distributed_rows=[(2, "clickhouse-02-host", marker)], - write_hostname="clickhouse-02-host", - node_2_hostname="clickhouse-02-host", - marker=marker, - ) - with self.assertRaisesRegex(RuntimeError, "первого шарда"): - self.module._assert_marker_path( - distributed_rows=[(2, "clickhouse-01-host", marker)], - write_hostname="clickhouse-01-host", - node_2_hostname="clickhouse-02-host", - marker=marker, - ) - - def test_cleanup_drops_tables_and_checks_both_nodes(self) -> None: - client = RecordingClient() - - self.run_cleanup_task(client) - - dropped = { - table - for table in (self.module.LOCAL_TABLE, self.module.DISTRIBUTED_TABLE) - if any( - command.startswith(f"DROP TABLE IF EXISTS default.{table} ") - for command in client.commands - ) - } - self.assertEqual( - dropped, {self.module.LOCAL_TABLE, self.module.DISTRIBUTED_TABLE} - ) - self.assertEqual(len(client.queries), len(self.module.NODES)) - self.assertTrue(client.closed) - - def test_cleanup_closes_client_when_tables_survive(self) -> None: - client = RecordingClient(remaining_tables=[("airflow_probe_local", "Log")]) - - with self.assertRaisesRegex(RuntimeError, "служебные таблицы остались"): - self.run_cleanup_task(client) - - self.assertTrue(client.closed) - - -DELIVERED_PARTITION = 3 -DELIVERED_OFFSET = 42 - - -class StubMessage: - """Сообщение Kafka в том объёме, в каком его читает пробник.""" - - def __init__(self, value: bytes) -> None: - self._value = value - - def partition(self) -> int: - return DELIVERED_PARTITION - - def offset(self) -> int: - return DELIVERED_OFFSET - - def error(self): - return None - - def value(self) -> bytes: - return self._value - - -class StubTopicPartition: - def __init__(self, topic: str, partition: int, offset: int) -> None: - self.topic = topic - self.partition = partition - self.offset = offset - - -def install_kafka_stub(produce_error: str | None = None): - """Ставит заглушку `confluent_kafka` и возвращает журналы её клиентов. - - Продюсер подтверждает доставку сразу и по известному адресу, консьюмер - отдаёт записанное с первого опроса. Если задан `produce_error`, запись - падает — так проверяется, что продюсер закрывается и на пути отказа. - """ - producers = [] - consumers = [] - - class Producer: - def __init__(self, _config) -> None: - self.written = b"" - self.flushed = False - producers.append(self) - - def produce(self, _topic, key=None, value=None, on_delivery=None) -> None: - if produce_error is not None: - raise RuntimeError(produce_error) - self.written = value - on_delivery(None, StubMessage(value)) - - def flush(self, _timeout) -> int: - self.flushed = True - return 0 - - class Consumer: - def __init__(self, _config) -> None: - self.assigned = [] - self.closed = False - consumers.append(self) - - def assign(self, partitions) -> None: - self.assigned = partitions - - def poll(self, _timeout): - return StubMessage(producers[-1].written) - - def close(self) -> None: - self.closed = True - - kafka_module = types.ModuleType("confluent_kafka") - kafka_module.Consumer = Consumer - kafka_module.Producer = Producer - kafka_module.TopicPartition = StubTopicPartition - sys.modules["confluent_kafka"] = kafka_module - return producers, consumers - - -class KafkaProbeTests(unittest.TestCase): - def test_producer_is_closed_when_write_fails(self) -> None: - producers, _ = install_kafka_stub(produce_error="брокер недоступен") - tasks = load_kafka_dag_tasks() - - with self.assertRaisesRegex(RuntimeError, "брокер недоступен"): - tasks["write_marker"]() - - self.assertEqual(len(producers), 1) - self.assertTrue(producers[0].flushed) - - def test_read_goes_to_the_address_broker_returned(self) -> None: - _, consumers = install_kafka_stub() - tasks = load_kafka_dag_tasks() - - written = tasks["write_marker"]() - tasks["read_marker"](written) - - self.assertEqual( - (written["partition"], written["offset"]), - (DELIVERED_PARTITION, DELIVERED_OFFSET), - ) - self.assertEqual(len(consumers), 1) - self.assertEqual(len(consumers[0].assigned), 1) - assigned = consumers[0].assigned[0] - self.assertEqual(assigned.partition, DELIVERED_PARTITION) - self.assertEqual(assigned.offset, DELIVERED_OFFSET) - self.assertTrue(consumers[0].closed) - - -def run_tests() -> int: - suite = unittest.defaultTestLoader.loadTestsFromModule(sys.modules[__name__]) - result = unittest.TestResult() - suite.run(result) - - problems = result.failures + result.errors - for test, details in problems: - print(f"ОШИБКА: {test.id()}", file=sys.stderr) - print(details, file=sys.stderr) - passed = result.testsRun - len(problems) - len(result.skipped) - print(f"ИТОГ: пройдено {passed}, ошибок {len(problems)}") - return 0 if result.wasSuccessful() else 1 - - -if __name__ == "__main__": - raise SystemExit(run_tests()) diff --git a/tests/smoke-guards.sh b/tests/smoke-guards.sh deleted file mode 100755 index 599a342..0000000 --- a/tests/smoke-guards.sh +++ /dev/null @@ -1,139 +0,0 @@ -#!/usr/bin/env bash -set -uo pipefail - -readonly ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -readonly LOCAL_TABLE="smoke_replicated_local" -readonly DISTRIBUTED_TABLE="smoke_distributed" -passed=0 -failed=0 - -compose() { - docker compose --project-directory "$ROOT_DIR" "$@" -} - -query() { - local service="$1" - local sql="$2" - compose exec -T "$service" clickhouse-client --query "$sql" -} - -pass() { - passed=$((passed + 1)) - printf 'ЗЕЛЁНО: %s.\n' "$1" -} - -fail() { - failed=$((failed + 1)) - printf 'ОШИБКА: %s.\n' "$1" >&2 -} - -wait_for_local_table() { - local attempt - for attempt in $(seq 1 100); do - if [[ "$(query clickhouse-01 "SELECT count() FROM system.tables WHERE database = 'default' AND name = '${LOCAL_TABLE}' FORMAT TSVRaw" 2>/dev/null || true)" == '1' ]]; then - return 0 - fi - sleep 0.05 - done - return 1 -} - -restore_stand() { - make --no-print-directory -C "$ROOT_DIR" up >/dev/null - query clickhouse-01 "DROP TABLE IF EXISTS default.${DISTRIBUTED_TABLE} ON CLUSTER clickstream_cluster SYNC" >/dev/null 2>&1 || true - query clickhouse-01 "DROP TABLE IF EXISTS default.${LOCAL_TABLE} ON CLUSTER clickstream_cluster SYNC" >/dev/null 2>&1 || true -} - -check_failure_guards() { - local failure_log - local failure_pid - local failure_status - local failure_elapsed - local interrupt_log - local interrupt_pid - local interrupt_status - local remaining_tables - local started_at - local bounded_failure_ok=0 - local interrupt_ok=0 - interrupt_status=999 - remaining_tables='не проверено' - - failure_log="$(mktemp)" - started_at="$(date +%s)" - setsid make --no-print-directory -C "$ROOT_DIR" smoke-cluster >"$failure_log" 2>&1 & - failure_pid=$! - if wait_for_local_table; then - compose kill clickhouse-02 >/dev/null - wait "$failure_pid" - failure_status=$? - failure_elapsed=$(( $(date +%s) - started_at )) - if [[ "$failure_status" -eq 2 ]] && [[ "$failure_elapsed" -lt 20 ]]; then - bounded_failure_ok=1 - fi - else - kill -TERM -- "-$failure_pid" >/dev/null 2>&1 || true - wait "$failure_pid" >/dev/null 2>&1 || true - fi - rm -f "$failure_log" - - restore_stand - - interrupt_log="$(mktemp)" - setsid env --default-signal=INT,TERM "$ROOT_DIR/scripts/clickhouse-smoke.sh" >"$interrupt_log" 2>&1 & - interrupt_pid=$! - if wait_for_local_table; then - kill -INT -- "-$interrupt_pid" - wait "$interrupt_pid" - interrupt_status=$? - remaining_tables="$(query clickhouse-01 "SELECT count() FROM clusterAllReplicas('clickstream_cluster', system.tables) WHERE database = 'default' AND name IN ('${LOCAL_TABLE}', '${DISTRIBUTED_TABLE}') FORMAT TSVRaw")" - if [[ "$interrupt_status" -eq 130 ]] && [[ "$remaining_tables" == '0' ]]; then - interrupt_ok=1 - fi - else - kill -TERM -- "-$interrupt_pid" >/dev/null 2>&1 || true - wait "$interrupt_pid" >/dev/null 2>&1 || true - fi - rm -f "$interrupt_log" - - if [[ "$bounded_failure_ok" -eq 1 ]] && [[ "$interrupt_ok" -eq 1 ]]; then - pass 'аварийная очистка ограничена по времени, SIGINT возвращает 130 и удаляет временные таблицы' - else - fail "нарушена аварийная семантика smoke: bounded=${bounded_failure_ok}, interrupt=${interrupt_ok}, status=${interrupt_status}, tables=${remaining_tables}" - fi -} - -check_keeper_runtime() { - local keeper_user - local keeper_nofile - local keeper_owner - keeper_user="$(compose exec -T clickhouse-keeper id -un)" - keeper_nofile="$(compose exec -T clickhouse-keeper awk '$1 == "Max" && $2 == "open" && $3 == "files" {print $4}' /proc/1/limits)" - keeper_owner="$(compose exec -T clickhouse-keeper stat -c '%U:%G' /var/lib/clickhouse/coordination)" - if [[ "$keeper_user" == 'clickhouse' ]] && [[ "$keeper_nofile" -ge 262144 ]] && [[ "$keeper_owner" == 'clickhouse:clickhouse' ]]; then - pass 'keeper работает от clickhouse с nofile 262144 и своим каталогом данных' - else - fail "неверное окружение keeper: user=${keeper_user}, nofile=${keeper_nofile}, owner=${keeper_owner}" - fi -} - -check_preflight_hint() { - local output - local status - compose kill clickhouse-02 >/dev/null - output="$("$ROOT_DIR/scripts/clickhouse-smoke.sh" 2>&1)" - status=$? - if [[ "$status" -ne 0 ]] && grep -q 'выполните make up' <<<"$output"; then - pass 'неполный стенд получает русскую подсказку выполнить make up' - else - fail 'нет русской подсказки для незапущенного стенда' - fi - make --no-print-directory -C "$ROOT_DIR" up >/dev/null -} - -check_failure_guards -check_keeper_runtime -check_preflight_hint - -printf 'ИТОГ: пройдено %d, ошибок %d\n' "$passed" "$failed" -[[ "$failed" -eq 0 ]] diff --git a/tests/stand-smoke-guards.sh b/tests/stand-smoke-guards.sh deleted file mode 100755 index a33b52f..0000000 --- a/tests/stand-smoke-guards.sh +++ /dev/null @@ -1,193 +0,0 @@ -#!/usr/bin/env bash -set -uo pipefail - -readonly ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -read -r -a COMPOSE_CMD <<<"${COMPOSE_BIN:-docker compose}" -readonly LOG_FILE="$(mktemp)" -readonly CLICKHOUSE_LOCAL_TABLE="airflow_probe_local" -readonly CLICKHOUSE_DISTRIBUTED_TABLE="airflow_probe_distributed" -restored=0 -passed=0 -red_smoke_pid='' - -compose() { - "${COMPOSE_CMD[@]}" --project-directory "$ROOT_DIR" "$@" -} - -published_port() { - local binding - local service="$1" - local container_port="$2" - binding="$(compose port "$service" "$container_port" 2>/dev/null || true)" - printf '%s\n' "${binding##*:}" -} - -query_node_2() { - local sql="$1" - local port - port="$(published_port clickhouse-02 8123)" - curl -sf --max-time 2 --data-binary "$sql" "http://127.0.0.1:${port}/" -} - -break_clickhouse_probe() { - local deadline - local table_state - - deadline=$((EPOCHSECONDS + 60)) - printf 'Ожидание служебных таблиц на ноде 2, предел 60 секунд...\n' - while [[ "$EPOCHSECONDS" -lt "$deadline" ]]; do - table_state="$(query_node_2 " - SELECT - countIf(name = '${CLICKHOUSE_LOCAL_TABLE}'), - countIf(name = '${CLICKHOUSE_DISTRIBUTED_TABLE}') - FROM system.tables - WHERE database = 'default' - AND name IN ( - '${CLICKHOUSE_LOCAL_TABLE}', - '${CLICKHOUSE_DISTRIBUTED_TABLE}' - ) - FORMAT TSV - " 2>/dev/null || true)" - if [[ "$table_state" == $'1\t1' ]]; then - query_node_2 \ - "DROP TABLE IF EXISTS default.${CLICKHOUSE_LOCAL_TABLE} SYNC" \ - >/dev/null - table_state="$(query_node_2 " - SELECT - countIf(name = '${CLICKHOUSE_LOCAL_TABLE}'), - countIf(name = '${CLICKHOUSE_DISTRIBUTED_TABLE}') - FROM system.tables - WHERE database = 'default' - AND name IN ( - '${CLICKHOUSE_LOCAL_TABLE}', - '${CLICKHOUSE_DISTRIBUTED_TABLE}' - ) - FORMAT TSV - " 2>/dev/null || true)" - if [[ "$table_state" == $'0\t1' ]]; then - printf 'Служебная локальная таблица удалена только на ноде 2.\n' - return - fi - return 1 - fi - sleep 0.05 - done - return 1 -} - -# Пробник разбит на четыре задачи, и упасть может любая из них: поломка на ноде -# 2 видна и проверке набора таблиц, и чтению маркера. Поэтому берём последний -# каталог запуска целиком и ищем образец по журналам всех его задач. -clickhouse_break_is_reported() { - compose exec -T airflow-scheduler bash -ceu ' - latest="$( - find /opt/airflow/logs/dag_id=test_clickhouse \ - -mindepth 1 -maxdepth 1 -type d -name "run_id=*" -printf "%T@ %p\n" | - sort -nr | - head -n 1 - )" - latest="${latest#* }" - test -n "$latest" - grep -Eqr --include="attempt=1.log" \ - "неверный набор таблиц на ноде 2|Unknown table expression identifier '\''default.airflow_probe_local'\''" \ - "$latest" - ' -} - -probe_failure_is_reported() { - local dag_id="$1" - grep -Eq \ - "ОШИБКА: (пробник ${dag_id}|Airflow не показал пробник ${dag_id})" \ - "$LOG_FILE" -} - -restore_stand() { - make --no-print-directory -C "$ROOT_DIR" COMPOSE="${COMPOSE_BIN:-docker compose}" up >/dev/null -} - -on_exit() { - local status=$? - trap - EXIT INT TERM - if [[ -n "$red_smoke_pid" ]]; then - kill -TERM -- "-$red_smoke_pid" >/dev/null 2>&1 || true - wait "$red_smoke_pid" >/dev/null 2>&1 || true - fi - rm -f "$LOG_FILE" - if [[ "$restored" -eq 0 ]] && ! restore_stand; then - printf 'ОШИБКА: не удалось восстановить стенд после проверки.\n' >&2 - status=1 - fi - exit "$status" -} - -trap on_exit EXIT -trap 'exit 130' INT TERM - -if make --no-print-directory -C "$ROOT_DIR" COMPOSE="${COMPOSE_BIN:-docker compose}" smoke >"$LOG_FILE" 2>&1; then - passed=$((passed + 1)) - printf 'ЗЕЛЁНО: перед проверкой отказа make smoke проходит полностью.\n' -else - printf 'ОШИБКА: исходный стенд не проходит make smoke; проверка отказа недостоверна.\n' >&2 - exit 1 -fi - -compose stop prometheus >/dev/null -setsid make --no-print-directory -C "$ROOT_DIR" \ - COMPOSE="${COMPOSE_BIN:-docker compose}" smoke >"$LOG_FILE" 2>&1 & -red_smoke_pid=$! -if ! break_clickhouse_probe; then - printf 'ОШИБКА: не удалось удалить служебную таблицу только на ноде 2.\n' >&2 - exit 1 -fi -compose stop kafka >/dev/null -wait "$red_smoke_pid" -status=$? -red_smoke_pid='' - -red_path_ok=1 -if [[ "$status" -eq 0 ]]; then - printf 'ОШИБКА: make smoke остался зелёным после внесённых поломок.\n' >&2 - red_path_ok=0 -fi -if ! grep -q 'ОШИБКА: сервис prometheus' "$LOG_FILE"; then - printf 'ОШИБКА: make smoke не назвал остановленный prometheus.\n' >&2 - red_path_ok=0 -fi -if ! probe_failure_is_reported test_clickhouse; then - printf 'ОШИБКА: make smoke не назвал пробник test_clickhouse.\n' >&2 - red_path_ok=0 -fi -if ! probe_failure_is_reported test_kafka; then - printf 'ОШИБКА: make smoke не назвал пробник test_kafka.\n' >&2 - red_path_ok=0 -fi -if ! clickhouse_break_is_reported; then - printf 'ОШИБКА: журнал test_clickhouse не связал отказ с удалённой таблицей на ноде 2.\n' >&2 - red_path_ok=0 -fi - -if [[ "$red_path_ok" -eq 1 ]]; then - passed=$((passed + 1)) - printf 'ЗЕЛЁНО: удаление таблицы на ноде 2 и остановка prometheus с kafka делают make smoke красным; оба пробника названы в отчёте.\n' -else - printf 'ОШИБКА: проверка красного пути завершилась с кодом make smoke %s.\n' "$status" >&2 - exit 1 -fi - -if ! restore_stand; then - printf 'ОШИБКА: не удалось восстановить стенд после проверки.\n' >&2 - exit 1 -fi -restored=1 - -if make --no-print-directory -C "$ROOT_DIR" COMPOSE="${COMPOSE_BIN:-docker compose}" smoke >"$LOG_FILE" 2>&1; then - passed=$((passed + 1)) - printf 'ЗЕЛЁНО: после восстановления make smoke снова проходит полностью.\n' -else - printf 'ОШИБКА: после восстановления стенд не проходит make smoke.\n' >&2 - exit 1 -fi - -rm -f "$LOG_FILE" -trap - EXIT INT TERM -printf 'ИТОГ: пройдено %d, ошибок 0\n' "$passed" diff --git a/tests/stand-smoke-static.sh b/tests/stand-smoke-static.sh deleted file mode 100755 index badaf04..0000000 --- a/tests/stand-smoke-static.sh +++ /dev/null @@ -1,49 +0,0 @@ -#!/usr/bin/env bash -set -euo pipefail - -readonly ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -readonly FIXTURE_DIR="$(mktemp -d)" - -cleanup() { - rm -rf "$FIXTURE_DIR" -} - -trap cleanup EXIT -mkdir -p "$FIXTURE_DIR/bin" "$FIXTURE_DIR/scripts" -cp "$ROOT_DIR/scripts/stand-smoke.sh" "$FIXTURE_DIR/scripts/stand-smoke.sh" -cp "$(type -P false)" "$FIXTURE_DIR/bin/docker" - -set +e -output="$(PATH="$FIXTURE_DIR/bin:$PATH" COMPOSE_BIN=false \ - "$FIXTURE_DIR/scripts/stand-smoke.sh" 2>&1)" -status=$? -set -e - -error_message_status=0 -grep -q 'ОШИБКА: compose.yaml или .env.example недоступны для чтения' \ - <<<"$output" || error_message_status=$? -green_message_status=0 -grep -q 'ЗЕЛЁНО: .env.example совпадает' <<<"$output" || green_message_status=$? -if [[ "$status" -ne 0 ]] && \ - [[ "$error_message_status" -eq 0 ]] && \ - [[ "$green_message_status" -eq 1 ]]; then - printf 'ЗЕЛЁНО: недоступные compose.yaml и .env.example не проходят статическую проверку.\n' -else - printf 'ОШИБКА: статическая проверка приняла недоступные файлы.\n' >&2 - exit 1 -fi - -cp "$ROOT_DIR/compose.yaml" "$ROOT_DIR/.env.example" "$FIXTURE_DIR/" -set +e -timeout 3s env LC_ALL=ru_RU.UTF-8 PATH="$FIXTURE_DIR/bin:$PATH" COMPOSE_BIN=false \ - "$FIXTURE_DIR/scripts/stand-smoke.sh" >/dev/null 2>&1 -status=$? -set -e -if [[ "$status" -ne 124 ]]; then - printf 'ЗЕЛЁНО: статическая проверка укладывается в три секунды при русской локали.\n' -else - printf 'ОШИБКА: статическая проверка превысила три секунды при русской локали.\n' >&2 - exit 1 -fi - -printf 'ИТОГ: пройдено 2, ошибок 0\n'