Surveillance et alertes Apache Airflow : guide d'implémentation pratique

Apache Airflow orchestre des workflows de données et d'automatisation critiques. Lorsque les planifications dérivent, les réessais s'intensifient ou le scheduler se bloque, vous devez le savoir rapidement avec suffisamment de contexte 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és, échecs, santé du scheduler).
• Ajouter un tableau de bord compact pour la conscience situationnelle.
• Vérifier la configuration avec des tests contrôlés.
• Préparer des playbooks de modes de défaillance et des procédures de restauration.

Le résultat est une fondation de surveillance que vous pouvez étendre en toute confiance.

Inventaire des versions et de l'environnement

Avant d'apporter des modifications, notez un inventaire concis pour savoir exactement où configurer et comment restaurer. Utilisez ce modèle et remplacez les valeurs pour votre environnement.

Modèle d'inventaire (remplissez avec vos valeurs) :

• Version Airflow : 2.6.x ou ultérieure recommandée
• Version Python : 3.8+ recommandée
• Executor : LocalExecutor ou CeleryExecutor
• Hôte webserver : hôte ou VIP, port 8080 (ajustez selon l'usage)
• Hôte(s) scheduler : noms d'hôtes
• Workers : nombre et noms d'hôtes (si CeleryExecutor)
• Base de données de métadonnées : moteur (ex. Postgres), hôte, pool de connexions
• Courtier de messages : ex. 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 webhook Slack (si utilisé)
• Chemin(s) du fichier de configuration : airflow.cfg et toute variable d'environnement de surcharge
• Accès : compte shell pour l'utilisateur de service Airflow et emplacement de sauvegarde des fichiers

Prérequis :

• Accès administrateur pour mettre à jour airflow.cfg ou les variables d'environnement.
• Accès à l'interface web Airflow et à l'API (pour les vérifications /health).
• Un canal de notification (SMTP email ou webhook) que vous pouvez tester en toute sécurité.
• Optionnel : un collecteur de métriques (ex. endpoint compatible StatsD ou collecteur Prometheus) et un outil de tableau de bord.

Chemin de configuration sûr

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

Périmètre du pilote

• Un DAG avec 3 à 5 tâches qui s'exécute au moins toutes les heures pour itérer rapidement.
• Signaux de base : santé du scheduler, durée d'exécution du DAG, échecs de tâches, file d'attente/backlog.
• Deux règles d'alerte pour commencer : scheduler en mauvais état, et pic d'échecs de tâches pour le DAG pilote.

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

Ajoutez ou confirmez ces entrées dans airflow.cfg (créez une sauvegarde datée au préalable) :

# Extraits de airflow.cfg

[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 = VOTRE_MOT_DE_PASSE_APP
smtp_mail_from = [email protected]

[metrics]
# Activez StatsD si vous avez un collecteur compatible StatsD.
statsd_on = True
statsd_host = 127.0.0.1
statsd_port = 8125
statsd_prefix = airflow
# Émettez le nom d'hôte dans les tags si supporté par votre collecteur de métriques.

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

[core]
# Maintenez un parallélisme modeste pour éviter les pics bruyants pendant le pilote.
parallelism = 16

Équivalents en variables d'environnement (si vous préférez la configuration par 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=VOTRE_MOT_DE_PASSE_APP
[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 appropriée pour votre environnement.

2. Ajouter un callback d'échec et un SLA à un DAG pilote

Cet exemple montre un on_failure_callback au niveau du DAG qui poste sur un webhook (ex. Slack Incoming Webhook). Adaptez l'URL et les champs du message à votre canal. Le code est volontairement simple pour le pilote.

# 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"Alerte Airflow : {dag_id}.{task_id} a échoué sur l'exécution {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 faire échouer la tâche à cause de problèmes de notification

with DAG(
    dag_id="pilot_monitoring_example",
    start_date=days_ago(1),
    schedule_interval="@hourly",
    catchup=False,
    sla_miss_callback=notify,  # alerte sur les SLA manqués
    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("échec construit pour le test pilote🔗)

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

Notes :

• Le webhook et l'email sont tous deux utilisés, afin de comparer la rapidité et la fiabilité des alertes.
• Le SLA dans default_args signifie que chaque tâche a un SLA de 20 minutes ; ajustez selon les caractéristiques de votre DAG.

3. Acheminer les métriques vers un collecteur

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

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}🔗

4. Créer un tableau de bord compact

Commencez avec 4 à 6 panneaux que vous pouvez inspecter d'un coup d'œil :

• Nombre et tendance d'âge des heartbeats du scheduler (fenêtre 5 min).
• Durée d'exécution du DAG p50/p95 pour le DAG pilote.
• Nombre de succès vs échecs de tâches (taux) pour le DAG pilote.
• Tâches en file d'attente et tâches en cours (plateforme entière).
• Nombre d'erreurs d'import (problèmes d'analyse).
• Optionnel : saturation du pool de connexions à la base de données si exposé par votre stack.

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

Créez des alertes faciles à raisonner. Remplacez les noms de métriques et labels pour correspondre à votre collecteur.

6. Documenter les propriétaires et la sévérité

• Propriétaire du DAG pilote : équipe ou personne capable de corriger les échecs.
• Rotation d'astreinte : qui page, quand, et escalade.
• Mapping de sévérité : par exemple, scheduler down = SEV-1 ; échecs d'un seul DAG = SEV-3.

Quoi surveiller en priorité

Ce tableau priorise les métriques et signaux de logs précoces à haut signal. Ajustez les seuils après une semaine d'observation.

Vérification et diagnostics

Prouvez la configuration 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é de base dans les versions récentes :

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

Réponse attendue :

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

Si vous ne pouvez pas accéder à l'endpoint, utilisez les logs serveur ou les outils de supervision de service pour confirmer que les processus tournent.

2. Provoquer un échec contrôlé

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

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

Si vous utilisez statsd_exporter sur localhost port 9102 :

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

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

4. Inspecter les logs pour la qualité du signal

Ouvrez le log de la tâche flaky dans l'UI. Confirmez que l'exception est visible et que les timestamps s'alignent avec ceux de vos alertes. Assurez-vous que les logs sont conservés où vos opérateurs peuvent les lire.

5. Valider les panneaux du tableau de bord

• Panneau heartbeat scheduler : doit montrer des incréments réguliers.
• Durée d'exécution DAG p95 : doit refléter les exécutions récentes ; l'exécution échouée peut être plus courte mais doit apparaître sur la timeline.
• Succès/échec des tâches : pic visible au moment du test.
• Tâches en file : restent proches de la normale si la capacité est adéquate.

6. Diagnostiquer si quelque chose n'a pas fonctionné

• Pas d'email : vérifiez l'hôte SMTP, les identifiants, et les règles de pare-feu. Testez un email via un script Python simple avec les mêmes paramètres SMTP.
• Pas d'alerte webhook : testez l'URL webhook avec curl ; vérifiez les règles de proxy.
• Pas de métriques : confirmez que statsd_on est true, l'exporter tourne, et les ports sont ouverts. Cherchez les erreurs dans les logs du scheduler concernant les clients métriques.
• Health check échoue : redémarrez les services Airflow concernés selon votre procédure de maintenance et vérifiez la connectivité à la base de métadonnées.

Modes de défaillance et récupération

Problèmes courants que vous pouvez rencontrer et comment récupérer en toute sécurité.

Playbook de restauration

• Sauvegardez airflow.cfg avant les modifications ; pour restaurer, remettez la copie précédente et redémarrez les services.
• Si un callback cause des échecs, retirez les lignes on_failure_callback et sla_miss_callback du DAG pilote, redéployez, et nettoyez seulement les exécutions du DAG affecté.
• Désactivez les règles d'alerte bruyantes en les basculant off dans votre outil d'alerte, puis revisitez les seuils.
• Si le chemin des métriques cause de l'instabilité, mettez statsd_on = False et redémarrez pour confirmer la stabilité.
• Relancez toujours la vérification /health et une exécution manuelle du DAG pilote pour valider la récupération.

Flux de réponse aux incidents

• Triage sous 5 minutes : classer la sévérité ; si scheduler en mauvais état, déclarer SEV-1 et pager l'astreinte.
• Contenir : mettre en pause les DAGs à fort churn si nécessaire pour réduire la charge pendant la récupération.
• Corriger : redémarrer les composants au besoin ; appliquer des correctifs ciblés (ex. revenir sur un fichier DAG défectueux).
• Vérifier : exécuter /health, confirmer la normalisation du tableau de bord, et vérifier les nouvelles exécutions des DAGs affectés.
• Documenter : noter la cause, la correction, et les ajustements de seuils pour la prochaine revue de bruit.

Liste de contrôle opérationnelle

Quotidien

• Vérifier l'endpoint /health Airflow pour un statut vert sur metadatabase et scheduler.
• Revoir les tâches échouées des dernières 24 heures ; s'assurer que les propriétaires sont notifiés et les problèmes suivis.
• Vérifier tâches en file vs en cours ; confirmer que les pools ne sont pas saturés.
• Scanner les erreurs d'import ; corriger ou revenir sur les DAGs cassés promptement.
• Confirmer que les alertes ont tiré et ont été acquittées pour tout incident.

Hebdomadaire

• Revue de bruit : lister les 3 alertes les plus fréquentes ; ajuster seuils ou ajouter conditions.
• Revue de capacité : comparer tâches en file et durées d'exécution semaine sur semaine.
• Hygiène du tableau de bord : retirer les panneaux inutilisés ; ajouter annotations pour les incidents.
• Audit des dépendances : noter les mises à jour à venir d'Airflow, drivers DB, ou Python qui pourraient affecter la surveillance.

Mensuel

• Exercice de reprise après sinistre : simuler une panne du scheduler et pratiquer la restauration.
• Mettre à jour les runbooks : garder les procédures de récupération pas à pas à jour avec la topologie actuelle.
• Étendre la couverture : ajouter 1 à 2 DAGs supplémentaires au périmètre d'alerte selon l'impact métier.

Exemples pratiques et résultats attendus

1. Alerte email à l'échec

• Provoquez un échec dans pilot_monitoring_example.flaky.
• Attendu : un email arrive dans la minute. Le message contient l'ID du DAG, l'ID de la tâche, et l'ID de l'exécution.

2. Notification webhook

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

3. Changement de santé du scheduler

• Arrêtez le processus scheduler pendant 3 minutes dans une fenêtre de test contrôlée.
• Attendu : le health check rapporte unhealthy ; votre règle d'alerte se déclenche ; le tableau de bord montre des incréments de heartbeat à plat.
• Restauration : redémarrez le scheduler ; confirmez que l'alerte se résout et que le heartbeat reprend.

4. SLA manqué

• Ajoutez temporairement time.sleep(1800) à ok_task pour dépasser le SLA de 20 minutes (pour test uniquement).
• Attendu : notification de SLA manqué via le même chemin de callback ; envisagez de garder les alertes SLA en sévérité notification seule au début.

Conclusion

Commencez petit et rendez-le observable. Un pilote ciblé avec un DAG, une poignée de métriques à haut signal, et deux ou trois alertes claires valide votre approche de surveillance rapidement. Prouvez que les alertes atteignent les bonnes personnes, que les tableaux de bord répondent aux questions de premier ordre, et que les logs fournissent le détail nécessaire. Une fois vérifié, étendez la couverture à vos DAGs principaux, ajustez les seuils pour réduire le bruit, et formalisez la réponse aux incidents.

Cette approche incrémentale maintient le risque bas, permet un retour rapide, et construit une couche de surveillance fiable pour Apache Airflow au fur et à mesure que vos besoins d'orchestration grandissent.