Зачем. Демонстрационный 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>
191 lines
6.6 KiB
Bash
Executable File
191 lines
6.6 KiB
Bash
Executable File
#!/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
|
||
}
|
||
|
||
clickhouse_break_is_reported() {
|
||
compose exec -T airflow-scheduler bash -ceu '
|
||
latest="$(
|
||
find /opt/airflow/logs/dag_id=test_clickhouse \
|
||
-type f -name "attempt=1.log" -printf "%T@ %p\n" |
|
||
sort -nr |
|
||
head -n 1
|
||
)"
|
||
latest="${latest#* }"
|
||
test -n "$latest"
|
||
grep -Eq \
|
||
"неверный набор таблиц на ноде 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"
|