fix(airflow): добавлен повтор запуска DAG после индексации
- Зачем: - при холодном старте GET уже видел DAG, пока POST запуска ещё считал его устаревшим. - Что: - точный стартовый ответ 404 повторяется до двадцати раз с паузой три секунды. - прочие ответы и сетевые ошибки завершают команду без риска двойного запуска. - граница повторов объяснена в коде и документации проверок. - Проверка: - выполнены `bash -n scripts/run-dag.sh`, `make config-test` и изолированные сценарии ответов API. - выполнены холодный `make up`, `make lint` и `make smoke` (21/21).
This commit is contained in:
1 parent
8d82679bc1
commit
20b0666d28
2 files changed
+42
-10
No files matched your search
@@ -211,9 +211,11 @@ Kafka → STG → ODS, и живёт он в `make check-clickhouse`.
|
|||||||
Особняком стоит `scripts/run-dag.sh`: это не проверка, а общий REST-переход от
|
Особняком стоит `scripts/run-dag.sh`: это не проверка, а общий REST-переход от
|
||||||
`make` к Airflow. `make up` запускает через него `world_initialize`, а
|
`make` к Airflow. `make up` запускает через него `world_initialize`, а
|
||||||
`make rebuild-storage` — `world_recreate`; сценарий ждёт появления дага в API,
|
`make rebuild-storage` — `world_recreate`; сценарий ждёт появления дага в API,
|
||||||
печатает идентификаторы дага и прогона и возвращает вызывающему его итог.
|
а затем до минуты повторяет только стартовый ответ `Dag with dag_id not found`.
|
||||||
Ожидание доставки событий и запуск `orders_ingest` теперь видны задачами
|
Прочие ответы и сетевые ошибки он не повторяет: при потерянном ответе запуск мог
|
||||||
`world_initialize`, а не спрятаны в shell.
|
уже создаться. Сценарий печатает идентификаторы дага и прогона и возвращает
|
||||||
|
вызывающему его итог. Ожидание доставки событий и запуск `orders_ingest` теперь
|
||||||
|
видны задачами `world_initialize`, а не спрятаны в shell.
|
||||||
|
|
||||||
Общее у смоука и `check-services` — счёт проверок, обращение к Compose и две
|
Общее у смоука и `check-services` — счёт проверок, обращение к Compose и две
|
||||||
проверки — вынесено в `scripts/stand-common.sh`; сам он не запускается.
|
проверки — вынесено в `scripts/stand-common.sh`; сам он не запускается.
|
||||||
|
|||||||
+37
-7
@@ -8,6 +8,7 @@ readonly DAG_ID="${1:-}"
|
|||||||
readonly ATTEMPTS=400
|
readonly ATTEMPTS=400
|
||||||
readonly PAUSE_SECONDS=3
|
readonly PAUSE_SECONDS=3
|
||||||
readonly REPORT_EVERY=5
|
readonly REPORT_EVERY=5
|
||||||
|
readonly TRIGGER_ATTEMPTS=20
|
||||||
|
|
||||||
read -r -a COMPOSE_CMD <<<"${COMPOSE_BIN:-docker compose}"
|
read -r -a COMPOSE_CMD <<<"${COMPOSE_BIN:-docker compose}"
|
||||||
|
|
||||||
@@ -68,14 +69,43 @@ jq -e --arg dag_id "$DAG_ID" '.dag_id == $dag_id' \
|
|||||||
>/dev/null 2>&1 <<<"$dag" || \
|
>/dev/null 2>&1 <<<"$dag" || \
|
||||||
fail "даг ${DAG_ID} не появился за $((ATTEMPTS * PAUSE_SECONDS)) с; $(next_step)"
|
fail "даг ${DAG_ID} не появился за $((ATTEMPTS * PAUSE_SECONDS)) с; $(next_step)"
|
||||||
|
|
||||||
if ! response="$(curl -sS --fail-with-body --max-time 10 -X POST \
|
# GET читает DagModel по идентификатору, а POST запуска принимает только запись
|
||||||
-H "Authorization: Bearer ${token}" \
|
# с `is_stale = false`. На холодном старте GET уже может видеть даг, пока POST
|
||||||
-H 'Content-Type: application/json' \
|
# ещё отвечает этим 404. Повторяем только его: после сетевой ошибки запрос мог
|
||||||
-d '{"logical_date":null}' \
|
# успеть создать запуск, и повтор с автоматическим run_id создал бы второй.
|
||||||
"http://127.0.0.1:${port}/api/v2/dags/${DAG_ID}/dagRuns" 2>/dev/null)"; then
|
expected_missing_detail="Dag with dag_id: '${DAG_ID}' not found"
|
||||||
|
triggered=0
|
||||||
|
for ((attempt = 1; attempt <= TRIGGER_ATTEMPTS; attempt++)); do
|
||||||
|
if ! response_with_status="$(curl -sS --max-time 10 -w $'\n%{http_code}' -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
|
||||||
|
fail "Airflow не ответил при запуске дага ${DAG_ID}; проверьте список запусков перед повтором команды"
|
||||||
|
fi
|
||||||
|
http_status="${response_with_status##*$'\n'}"
|
||||||
|
response="${response_with_status%$'\n'*}"
|
||||||
detail="$(jq -r '.detail // .message // empty' <<<"$response" 2>/dev/null || true)"
|
detail="$(jq -r '.detail // .message // empty' <<<"$response" 2>/dev/null || true)"
|
||||||
fail "Airflow не запустил даг ${DAG_ID}${detail:+: ${detail}}; $(next_step)"
|
case "$http_status" in
|
||||||
fi
|
2??)
|
||||||
|
triggered=1
|
||||||
|
break
|
||||||
|
;;
|
||||||
|
404)
|
||||||
|
if [[ "$detail" == "$expected_missing_detail" ]]; then
|
||||||
|
if ((attempt == 1 || attempt % REPORT_EVERY == 0)); then
|
||||||
|
printf 'Airflow: даг %s ещё не готов к запуску; ждём (%d с).\n' \
|
||||||
|
"$DAG_ID" "$((attempt * PAUSE_SECONDS))"
|
||||||
|
fi
|
||||||
|
((attempt < TRIGGER_ATTEMPTS)) && sleep "$PAUSE_SECONDS"
|
||||||
|
continue
|
||||||
|
fi
|
||||||
|
;;
|
||||||
|
esac
|
||||||
|
fail "Airflow не запустил даг ${DAG_ID}: HTTP ${http_status}${detail:+: ${detail}}; $(next_step)"
|
||||||
|
done
|
||||||
|
((triggered == 1)) || \
|
||||||
|
fail "даг ${DAG_ID} не стал доступен для запуска за $((TRIGGER_ATTEMPTS * PAUSE_SECONDS)) с; $(next_step)"
|
||||||
run_id="$(jq -r '.dag_run_id // empty' <<<"$response")"
|
run_id="$(jq -r '.dag_run_id // empty' <<<"$response")"
|
||||||
state="$(jq -r '.state // "queued"' <<<"$response")"
|
state="$(jq -r '.state // "queued"' <<<"$response")"
|
||||||
[[ -n "$run_id" ]] || fail "Airflow не вернул идентификатор запуска ${DAG_ID}"
|
[[ -n "$run_id" ]] || fail "Airflow не вернул идентификатор запуска ${DAG_ID}"
|
||||||
|
|||||||
Reference in new issue
Block a user