Guide · Airflow

Surveiller Apache Airflow avec un dead man's switch

L'interface d'Airflow signale un DAG en échec, à condition que le scheduler soit encore là pour le constater. Elle ne dit rien quand le scheduler lui-même se bloque, qu'une file de workers sature, ou qu'un DAG se retrouve mis en pause silencieusement. Un check-in envoyé depuis le DAG comble ce trou : pas d'exécution, pas de ping, vous êtes alerté.

1. Créer un check

Créez un check dans Mortemain avec la même planification que votre DAG (plus une marge pour la durée normale d'exécution), et copiez son URL de ping.

2. Ajouter des callbacks de succès et d'échec au DAG

Airflow appelle on_success_callback une fois quand toutes les tâches d'une exécution réussissent, et on_failure_callback dès qu'une tâche échoue. Définissez les deux au niveau du DAG pour qu'un seul ping couvre l'exécution entière, pas une simple tâche :

dags/nightly_etl.py

import requests
from airflow import DAG
from airflow.operators.python import PythonOperator

PING_URL = "https://ping.mortemain.com/your-check-uuid"

def notify_success(context):
    requests.get(PING_URL, timeout=10)

def notify_failure(context):
    requests.get(PING_URL + "/fail", timeout=10)

with DAG(
    dag_id="nightly_etl",
    schedule="0 3 * * *",
    on_success_callback=notify_success,
    on_failure_callback=notify_failure,
) as dag:
    # ... vos tâches ...
    pass

Les deux callbacks reçoivent le context de l'exécution : vous pouvez donc aussi envoyer un court corps en POST (l'ID du run, l'URL des logs d'une tâche) comme payload du check-in, utile à retrouver plus tard.

3. Pas de callbacks ? Utilisez un BashOperator final

Si vous préférez garder le ping hors du code Python, ajoutez-le comme des tâches à part entière. L'une fait le check-in en cas de succès, en aval de tout le reste ; l'autre fait le check-in sur /fail et ne s'exécute que si quelque chose en amont a échoué :

dags/nightly_etl.py

from airflow.operators.bash import BashOperator

PING_URL = "https://ping.mortemain.com/your-check-uuid"

ping_success = BashOperator(
    task_id="ping_mortemain_success",
    bash_command=f"curl -fsS -m 10 {PING_URL}",
    trigger_rule="all_success",
)

ping_failure = BashOperator(
    task_id="ping_mortemain_failure",
    bash_command=f"curl -fsS -m 10 {PING_URL}/fail",
    trigger_rule="one_failed",
)

extract >> transform >> load >> [ping_success, ping_failure]

Le trigger_rule fait tout le travail : all_success ne se déclenche que si tout ce qui précède a réussi, one_failed uniquement si quelque chose a échoué.

4. Suivre la durée d'exécution (optionnel)

Faites le check-in sur /start depuis la première tâche pour que Mortemain puisse mesurer la durée réelle de l'exécution du DAG, utile pour repérer un job qui « réussit » toujours mais rallonge un peu chaque nuit :

dags/nightly_etl.py

ping_start = BashOperator(
    task_id="ping_mortemain_start",
    bash_command=f"curl -fsS -m 10 {PING_URL}/start",
)

ping_start >> extract >> transform >> load >> [ping_success, ping_failure]

Tester

Déclenchez une exécution manuelle du DAG et vérifiez que le check passe à up. Faites ensuite échouer une tâche exprès (un import cassé, une exception forcée) et vérifiez que l'alerte down arrive immédiatement : c'est exactement le type de panne qu'une vue purement basée sur la planification d'Airflow peut manquer.