Surveiller et alerter Apache Airflow : guide pratique avec exemples

Intro

Cette version française explique Apache Airflow monitoring and alerts with practical examples avec le même objectif pratique que l article source : aider le lecteur à comprendre le contexte, les décisions à prendre et les points à vérifier avant de passer à l action.

Apache Airflow orchestre des workflows de données et d'automatisation critiques. Quand un planning dérive, que les retries explosent ou que le scheduler se fige, vous devez être averti rapidement, avec le contexte utile pour agir. Ce guide propose une approche pratique et sûre de la surveillance et des alertes pour Airflow. Vous allez :

• Identifier les premières métriques et signaux de logs qui comptent.
• Configurer des alertes de base (SLA manquées, échecs, santé du scheduler).
• Ajouter un tableau de bord compact pour la conscience de la situation.
• Vérifier la mise en place via des tests contrôlés.
• Préparer des playbooks par mode de panne et des rollbacks.

Le résultat : une fondation de monitoring que vous pourrez étendre en confiance.

Inventaire version et environnement

Avant tout changement, consignez un inventaire concis pour savoir exactement où configurer et comment revenir en arrière. Modèle d'inventaire (exemples à adapter) :

• Version Airflow : 2.6.x ou plus récent recommandé
• Version Python : 3.8+ recommandé
• Executor : LocalExecutor ou CeleryExecutor
• Hôte webserver : hôte ou VIP, port 8080 (à adapter)
• Hôte(s) scheduler : noms d'hôtes
• Workers : nombre et hôtes (si CeleryExecutor)
• Base de métadonnées : moteur (ex. Postgres), hôte, pooling
• Broker de messages : Redis ou RabbitMQ (si CeleryExecutor)
• Stockage des logs : système de fichiers local ou distant (ex. S3 ou GCS)
• SMTP Email : nom d'hôte, adresse d'expéditeur
• Canaux d'alerte : liste email, URL de webhook Slack (si utilisé)
• Fichiers de config : airflow.cfg et variables d'environnement
• Accès : compte shell du service Airflow et emplacement des sauvegardes

Pré-requis :

• Accès administratif pour modifier airflow.cfg ou les variables d'environnement.
• Accès à l'UI et à l'API Airflow (pour les checks /health).
• Un canal de notification testable (SMTP email ou webhook).
• Optionnel : une destination de métriques (StatsD ou collecteur Prometheus) et un outil de tableau de bord.

Chemin de configuration sûr

Limitez le déploiement initial à un pilote vérifiable. Utilisez un DAG représentatif pour prouver la boucle de bout en bout avant de généraliser.

Périmètre pilote :

• Un DAG de 3-5 tâches s'exécutant au moins chaque heure pour itérer vite.
• Signaux de base : santé du scheduler, durée de run du DAG, échecs de tâches, file d'attente/backlog.
• Deux règles d'alerte pour commencer : scheduler unhealthy, et pic d'échecs de tâches sur le DAG pilote.

1. Activer email et métriques de base dans Airflow

Sauvegardez airflow.cfg, puis ajoutez ou confirmez ces entrées :

# airflow.cfg (extraits)

[email]
email_backend = airflow.utils.email.send_email_smtp
smtp_host = smtp.example.com
smtp_starttls = True
smtp_ssl = False
smtp_user = [email protected]
smtp_password = YOUR_APP_PASSWORD
smtp_mail_from = [email protected]

[metrics]
# Activez StatsD si vous avez une cible compatible.
statsd_on = True
statsd_host = 127.0.0.1
statsd_port = 8125
statsd_prefix = airflow
# Émettez le hostname en tags si votre collecteur le supporte.

[logging]
# Assurez-vous que les logs (locaux ou distants) sont lisibles et conservés.
base_log_folder = /var/log/airflow

[core]
# Limitez la parallélisation pour éviter des pics bruyants en pilote.
parallelism = 16

Équivalents en variables d'environnement :

AIRFLOW__EMAIL__EMAIL_BACKEND=airflow.utils.email.send_email_smtp
AIRFLOW__EMAIL__SMTP_HOST=smtp.example.com
AIRFLOW__EMAIL__SMTP_STARTTLS=True
AIRFLOW__EMAIL__SMTP_SSL=False
[email protected]
AIRFLOW__EMAIL__SMTP_PASSWORD=YOUR_APP_PASSWORD
[email protected]

AIRFLOW__METRICS__STATSD_ON=True
AIRFLOW__METRICS__STATSD_HOST=127.0.0.1
AIRFLOW__METRICS__STATSD_PORT=8125
AIRFLOW__METRICS__STATSD_PREFIX=airflow

Redémarrez les services Airflow dans une fenêtre de maintenance adaptée.

1. Ajouter un callback d'échec et une SLA au DAG pilote

Exemple construit : un on_failure_callback de niveau DAG qui envoie un message à un webhook (ex. Slack Incoming Webhook). Adaptez l'URL et le texte.

# dags/pilot_monitoring_example.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from datetime import timedelta
import json
import urllib.request

WEBHOOK_URL = "https://hooks.example.com/services/T000/B000/XXXXX"  # à remplacer

def notify(context):
    dag_id = context.get("dag_run").dag_id if context.get("dag_run") else context.get("dag").dag_id
    task_id = context.get("task_instance").task_id
    run_id = context.get("dag_run").run_id if context.get("dag_run") else "manual__test"
    text = f"Airflow alert: {dag_id}.{task_id} failed on run {run_id}"
    data = json.dumps({"text": text}).encode("utf-8")
    req = urllib.request.Request(WEBHOOK_URL, data=data, headers={"Content-Type": "application/json"})
    try:
        urllib.request.urlopen(req, timeout=5)
    except Exception:
        pass  # Ne pas échouer la tâche à cause de la notification

with DAG(
    dag_id="pilot_monitoring_example",
    start_date=days_ago(1),
    schedule_interval="@hourly",
    catchup=False,
    sla_miss_callback=notify,  # alerte en cas de SLA manquée
    default_args={
        "email_on_failure": True,
        "email_on_retry": False,
        "email": ["[email protected]"],
        "on_failure_callback": notify,
        "retries": 1,
        "retry_delay": timedelta(minutes=5),
        "sla": timedelta(minutes=20),
    },
    tags=["monitoring-pilot"],
) as dag:

    def ok_task():
        return "ok"

    def flaky_task():
        raise RuntimeError("constructed failure for pilot test")

    t1 = PythonOperator(task_id="ok", python_callable=ok_task)
    t2 = PythonOperator(task_id="flaky", python_callable=flaky_task)
    t1 >> t2

Notes :

• Webhook et email sont utilisés pour comparer réactivité et fiabilité.
• La SLA dans default_args applique 20 minutes à chaque tâche ; adaptez à votre DAG.

1. Acheminer les métriques vers un collecteur

Si votre collecteur supporte StatsD, pointez-le vers l'hôte: port ci-dessus. Pour Prometheus via statsd_exporter, un fichier de mapping d'exemple :

mappings:
  - match: "airflow.scheduler.heartbeat"
    name: "airflow_scheduler_heartbeat_count"
    labels: {}
  - match: "airflow.dag_processing.import_errors"
    name: "airflow_dag_import_errors_total"
  - match: "airflow.operator_successes"
    name: "airflow_task_success_total"
    labels:
      operator: "${1}"
  - match: "airflow.operator_failures"
    name: "airflow_task_fail_total"
    labels:
      operator: "${1}"

1. Créer un tableau de bord compact

Visez 4-6 panneaux lisibles d'un coup d'œil :

• Heartbeat du scheduler (compte et âge, fenêtre 5 min).
• p50/p95 de durée des runs pour le DAG pilote.
• Taux de succès vs échecs de tâches pour le DAG pilote.
• Tâches en file d'attente et en cours (plateforme).
• Compte d'erreurs d'import (problèmes de parsing).
• Optionnel : saturation du pool de connexions DB si exposée.

1. Définir deux ou trois règles d'alerte ciblées

Créez des alertes faciles à raisonner. Exemples (à adapter selon vos métriques) :

1. Documenter responsables et sévérité

• Propriétaire du DAG pilote : équipe/personne capable de corriger.
• Astreinte : qui est pagé, quand, et escalade.
• Cartographie de sévérité : scheduler indisponible = SEV-1 ; échecs d'un seul DAG = SEV-3.

Que surveiller en premier

Tableau de priorités (seuils à affiner après une semaine d'observation) :

Vérifications et diagnostics

Prouvez la mise en place avec des tests contrôlés et des résultats observables.

1. Health check du webserver et du scheduler

Airflow expose un endpoint de santé dans les versions récentes :

curl -s http://<webserver_host>:8080/health | jq .

Attendu (exemple) :

{
  "metadatabase": {"status": "healthy"},
  "scheduler": {"status": "healthy"}
}

Si l'endpoint n'est pas accessible, utilisez les logs ou votre superviseur de services pour confirmer l'état des processus.

1. Provoquer un échec contrôlé

• Débloquez le DAG pilote dans l'UI.
• Déclenchez un run manuellement (Run -> Trigger DAG) ou attendez la planification.
• La tâche "flaky" échoue par conception. Sous une minute, vous devriez voir :
• Un email à [email protected].
• Une notification webhook dans votre canal.
• Une hausse des métriques d'échec de tâches sur le tableau de bord.

1. Vérifier l'émission des métriques

Si vous utilisez statsd_exporter sur localhost port 9102 (exemple) :

curl -s http://127.0.0.1:9102/metrics | grep -E "airflow_"

Attendu : plusieurs séries airflow_* incluant scheduler, succès/échecs des tâches, erreurs d'import.

1. Inspecter les logs et la qualité du signal

Ouvrez le log de la tâche "flaky" dans l'UI. Confirmez l'exception et l'alignement temporel avec les alertes. Vérifiez la rétention et l'accessibilité des logs.

1. Valider les panneaux du tableau de bord

• Heartbeat scheduler : incréments réguliers.
• p95 de durée du DAG : reflète les runs récents.
• Succès/échecs : pic visible au moment du test.
• Tâches en file : proche de la normale si la capacité suffit.

1. Diagnostiquer si quelque chose manque

• Pas d'email : vérifiez SMTP, identifiants, pare-feu. Testez via un script Python simple.
• Pas de webhook : testez l'URL avec curl ; contrôlez proxy et ACL.
• Pas de métriques : confirmez statsd_on, exporter actif et ports ouverts. Consultez les logs du scheduler.
• Health check en échec : redémarrez les composants dans la fenêtre prévue et vérifiez la connectivité DB de métadonnées.

Modes de panne et reprise

Problèmes fréquents et remédiations sûres.

Playbook de rollback

• Sauvegardez airflow.cfg avant les changements ; pour revenir en arrière, restaurez et redémarrez les services.
• Si le callback provoque des échecs, retirez on_failure_callback et sla_miss_callback du DAG pilote, redéployez et nettoyez les runs affectés.
• Désactivez les règles d'alerte bruyantes dans l'outil d'alerte, puis ajustez les seuils.
• Si l'émission de métriques est instable, mettez statsd_on = False puis redémarrez pour valider la stabilité.
• Rejouez /health et un run manuel du DAG pilote pour confirmer la reprise.

Flux de réponse à incident (exemple)

• Triage sous 5 minutes : classifier la sévérité ; si scheduler unhealthy, déclarer SEV-1 et pager l'astreinte.
• Contenir : mettre en pause les DAGs très bavards pour réduire la charge.
• Remédier : redémarrer les composants ; appliquer des correctifs ciblés (ex. revert d'un fichier DAG fautif).
• Vérifier : exécuter /health, normalisation du tableau de bord, nouveaux runs OK.
• Documenter : cause, correctif et ajustements de seuils pour la revue de bruit.

Liste d'opérations

Quotidien

• Vérifier l'endpoint /health d'Airflow pour un statut vert sur metadatabase et scheduler.
• Passer en revue les tâches échouées des dernières 24 h ; garantir la notification des owners et le suivi.
• Contrôler tâches en file vs en cours ; confirmer l'absence de saturation de pools.
• Scanner les erreurs d'import ; corriger ou revert rapidement.
• Confirmer que les alertes ont bien été acquittées.

Hebdomadaire

• Revue de bruit : top 3 des alertes par volume ; ajuster seuils/conditions.
• Revue de capacité : comparer files d'attente et durées semaine sur semaine.
• Hygiène du dashboard : retirer les panneaux inutilisés ; annoter les incidents.
• Audit de dépendances : noter les mises à jour Airflow, drivers DB ou Python pouvant impacter le monitoring.

Mensuel

• Exercice de reprise : simuler une panne du scheduler et pratiquer la restauration.
• Mise à jour des runbooks : refléter la topologie actuelle.
• Élargir la couverture : ajouter 1-2 DAGs à l'échelle d'alerte selon l'impact métier.

Exemples pratiques et résultats attendus

1. Alerte email sur échec

• Déclenchez un échec sur pilot_monitoring_example.flaky.
• Attendu : email en moins d'une minute, avec DAG ID, task ID et run ID.

1. Notification webhook

• Pour le même échec, un message est posté dans votre canal.
• Attendu : message court avec identifiants du DAG et de la tâche.

1. Changement de santé du scheduler

• Arrêtez le processus du scheduler 3 minutes dans une fenêtre contrôlée.
• Attendu : health check en "unhealthy" ; alerte déclenchée ; flatline du heartbeat au dashboard.
• Retour : redémarrez le scheduler ; l'alerte se résorbe et le heartbeat reprend.

1. SLA manquée

• Ajoutez temporairement time.sleep(1800) à ok_task pour dépasser la SLA de 20 minutes (test contrôlé).
• Attendu : notification de SLA manquée via le même callback ; gardez ces alertes en sévérité d'information au départ.

Conclusion

Commencez petit et rendez le tout observable. Un pilote focalisé avec un DAG, quelques métriques à fort signal et deux ou trois alertes claires valide rapidement votre approche de monitoring. Prouvez que les alertes atteignent les bonnes personnes, que le tableau de bord répond aux premières questions, et que les logs apportent le détail nécessaire. Une fois vérifié, étendez la couverture à vos DAGs prioritaires, affinez les seuils pour réduire le bruit, et formalisez la réponse à incident. Cette démarche incrémentale minimise le risque, accélère le feedback et bâtit une couche de surveillance fiable pour Apache Airflow à mesure que vos besoins d'orchestration grandissent.