Dépanner les DAG

Managed Airflow (3e génération) | Managed Airflow (2e génération) | Managed Airflow (ancienne 1re génération)

Cette page décrit les étapes de dépannage, ainsi que des informations sur les problèmes courants liés aux workflows.

De nombreux problèmes d'exécution de DAG sont dus à des performances non optimales de l'environnement. Vous pouvez optimiser votre environnement en suivant le guide Optimiser les performances et les coûts de l'environnement.

Certains problèmes d'exécution de DAG peuvent être dus au fait que le planificateur Airflow ne fonctionne pas correctement ou de manière optimale. Suivez les instructions de dépannage du planificateur pour résoudre ces problèmes.

Résoudre les problèmes avec l'agent Managed Airflow

L'agent Managed Airflow peut vous aider à comprendre, diagnostiquer et résoudre les problèmes liés aux échecs d'exécution de tâches et de DAG Airflow en analysant les journaux d'exécution des tâches, le code source des DAG et les métriques d'environnement. Pour en savoir plus, consultez Résoudre les problèmes liés aux tâches Airflow et aux exécutions de DAG avec l'agent Managed Airflow.

Résoudre un problème lié aux workflows

Pour commencer à résoudre les problèmes, procédez comme suit :

  1. Vérifiez les journaux Airflow.

    Vous pouvez augmenter le niveau de journalisation d'Airflow en remplaçant l'option de configuration Airflow suivante.

    Section Clé Valeur
    logging logging_level La valeur par défaut est INFO. Définissez la valeur sur DEBUG pour obtenir plus de verbosité dans les messages du journal.

  2. Consultez le tableau de bord de surveillance.

  3. Consultez Cloud Monitoring.

  4. Dans la console Google Cloud , recherchez les erreurs sur les pages des composants de votre environnement.

  5. Dans l'interface Web Airflow, recherchez les instances de tâche ayant échoué dans la Vue Graphe du DAG.

    Section Clé Valeur
    webserver dag_orientation LR, TB, RL ou BT

Déboguer des échecs de l'opérateur

Pour déboguer un échec de l'opérateur, procédez comme suit :

  1. Recherchez les erreurs spécifiques à la tâche.
  2. Vérifiez les journaux Airflow.
  3. Consultez Cloud Monitoring.
  4. Vérifiez les journaux spécifiques à l'opérateur.
  5. Corrigez les erreurs.
  6. Importez le DAG dans le dossier /dags.
  7. Dans l'interface Web Airflow, effacez les états antérieurs du DAG.
  8. Relancez ou exécutez le DAG.

Résoudre les problèmes d'exécution des tâches

Airflow est un système distribué avec de nombreuses entités telles que le programmateur, l'exécuteur et les nœuds de calcul qui communiquent entre eux via une file d'attente de tâches et la base de données Airflow, et qui envoient des signaux (comme SIGTERM). Le schéma suivant offre une vue d'ensemble des interconnexions entre les composants Airflow.

Interaction entre les composants Airflow
Figure 1. Interaction entre les composants Airflow (cliquez pour agrandir)

Dans un système distribué comme Airflow, il peut y avoir des problèmes de connectivité réseau ou l'infrastructure sous-jacente peut rencontrer des problèmes intermittents. Cela peut entraîner l'échec et la reprogrammation de l'exécution des tâches, ou l'échec de l'exécution des tâches (par exemple, les tâches zombies ou les tâches bloquées en cours d'exécution). Airflow dispose de mécanismes pour faire face à de telles situations et reprendre automatiquement le fonctionnement normal. Les sections suivantes expliquent les problèmes courants qui se produisent lors de l'exécution des tâches par Airflow.

Résoudre les problèmes liés aux tâches KubernetesExecutor

CeleryKubernetesExecutor est un type d'exécuteur dans Managed Airflow (3e génération) qui peut utiliser CeleryExecutor et KubernetesExecutor en même temps.

Pour en savoir plus sur la résolution des problèmes liés aux tâches exécutées avec KubernetesExecutor, consultez la page Utiliser CeleryKubernetesExecutor.

Les tâches échouent sans émettre de journaux

Il peut arriver qu'une instance de tâche échoue sans émettre de journaux pour plusieurs raisons. Par exemple, cela peut se produire en raison d'erreurs d'analyse de DAG, de retards de synchronisation de DAG ou si un pod de nœud de calcul Airflow est évincé lors de l'exécution d'une tâche (voir Échec de la tâche en raison de l'éviction du pod).

Dépassement du délai d'analyse du DAG ou erreurs d'analyse

Si un fichier DAG comporte des erreurs de programmation ou si l'analyse d'un fichier DAG prend trop de temps, le programmeur Airflow peut planifier des tâches, mais les nœuds de calcul Airflow ne peuvent pas les exécuter. Dans ce cas, une tâche peut être marquée comme Failed sans aucun journal d'exécution.

Symptômes

  • Les journaux des nœuds de calcul Airflow dans Cloud Logging contiennent des messages tels que :

    • airflow.exceptions.AirflowException: Dag "example-dag" could not be found; either it does not exist or it failed to parse.
    • airflow.exceptions.AirflowTaskTimeout: Timeout, PID: 12345 (ce message peut ne pas contenir le nom ni le chemin d'accès du DAG).
    • ERROR - Failed to import: /home/airflow/gcs/dags/example-dag.py (sans tracebacks détaillés).

  • Si des erreurs d'importation de DAG se produisent, elles peuvent s'afficher dans l'interface utilisateur d'Airflow ou dans les messages Broken DAG sur la page Détails de l'environnement de la consoleGoogle Cloud .

Solution

  • Consultez les journaux des nœuds de calcul Airflow pour détecter les erreurs liées à l'analyse DAG. Si des erreurs AirflowTaskTimeout s'affichent, cela peut signifier que l'analyse du DAG est en train d'expirer. Le délai avant expiration de l'analyse sur les nœuds de calcul Airflow est contrôlé par dagbag_import_timeout.

  • Si les durées d'analyse des DAG sont longues, vérifiez si le processeur est en conflit sur le cluster de votre environnement. Si les nœuds de calcul ne disposent pas de suffisamment de processeurs : Augmentez le nombre de processeurs pour les nœuds de calcul Airflow. Vous pouvez également réduire worker_concurrency, comme décrit dans Optimiser votre environnement.

  • Si l'utilisation du processeur est faible, optimisez la définition de votre DAG pour réduire le temps d'analyse, par exemple en évitant le code de premier niveau.

  • Si vous voyez des erreurs Failed to import ou des erreurs dans l'interface utilisateur Airflow ou la consoleGoogle Cloud , consultez les journaux du processeur de DAG pour obtenir des informations détaillées sur les traces, comme décrit dans Résoudre les problèmes liés au processeur de DAG. Vous pouvez également afficher les erreurs d'importation de DAG en exécutant la commande gcloud CLI suivante :

    gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags list-import-errors

    Si votre DAG appelle des services externes lors de l'analyse, envisagez d'ajouter des blocs try...except autour de ces appels pour gérer les erreurs temporaires.

  • Si l'optimisation de l'analyse DAG n'est pas possible, augmentez dagbag_import_timeout. Remplacez cette option de configuration Airflow par une valeur supérieure à la valeur par défaut de 30 secondes (par exemple, 120 secondes).

Retards de synchronisation des fichiers DAG

Lorsque vous importez ou mettez à jour des fichiers DAG dans le bucket de votre environnement, il faut du temps pour que ces fichiers soient synchronisés avec les nœuds de calcul et les programmeurs Airflow. Cette synchronisation s'effectue indépendamment sur tous les planificateurs et nœuds de calcul. Si vous déclenchez une exécution de DAG peu après avoir importé ou mis à jour un fichier DAG, et que le fichier DAG n'est pas encore synchronisé avec un nœud de calcul qui récupère une tâche, la tâche échoue sans journaux et vous pouvez voir airflow.exceptions.AirflowException: Dag "example-dag" could not be found... dans les journaux des nœuds de calcul.

Cette synchronisation prend généralement une à deux minutes, mais peut durer plus longtemps si vous avez de nombreux fichiers ou des fichiers volumineux dans les dossiers dags/ ou plugins/ du bucket.

Solution

Attendez au moins deux minutes après avoir importé ou mis à jour des DAG ou des plug-ins avant de déclencher ou d'activer des DAG.

Tâches bloquées à l'état "En file d'attente"

Dans les versions d'Airflow antérieures à la version 2.6.3, les tâches peuvent parfois rester bloquées de manière permanente dans l'état queued. Cela peut se produire si une tâche est marquée comme mise en file d'attente dans la base de données Airflow, mais n'existe pas réellement dans Celery. Dans ce cas, les nœuds de calcul Airflow peuvent échouer aux vérifications de vivacité et redémarrer, ce qui peut entraîner l'échec des tâches avec des erreurs "Fichier journal introuvable".

Ce problème est résolu dans Airflow 2.6.3 et versions ultérieures. Si vous utilisez une version antérieure d'Airflow, vous pouvez mettre à niveau votre environnement vers une version d'image qui utilise Airflow 2.6.3 ou une version ultérieure.

Pour contourner ce problème, vous pouvez supprimer manuellement les tâches bloquées dans l'état "En file d'attente". Dans l'UI Airflow, accédez à Browse > Task Instances (Parcourir > Instances de tâche), recherchez les instances de tâche bloquées à l'état queued, puis définissez leur état sur failed.

Les tâches sont interrompues brusquement

Lors de l'exécution des tâches, les nœuds de calcul Airflow peuvent s'arrêter brusquement en raison de problèmes qui ne sont pas spécifiquement liés à la tâche elle-même. Consultez la section Causes racines courantes pour obtenir la liste de ces scénarios et des solutions possibles. Les sections suivantes couvrent d'autres symptômes qui pourraient découler de ces causes profondes :

Tâches zombies

Airflow détecte deux types d'incohérences entre une tâche et un processus qui l'exécute :

  • Les tâches zombies sont des tâches qui sont censées s'exécuter, mais qui ne le font pas. Cela peut se produire si le processus de la tâche a été arrêté ou ne répond pas, si le nœud de calcul Airflow n'a pas signalé l'état d'une tâche à temps parce qu'il est surchargé, ou si la VM sur laquelle la tâche est exécutée a été éteinte. Airflow recherche régulièrement ces tâches et les fait échouer ou les relance, selon les paramètres de la tâche.

    Découvrir les tâches zombies

    resource.type="cloud_composer_environment"
resource.labels.environment_name="ENVIRONMENT_NAME"
log_id("airflow-scheduler")
textPayload:"Detected zombie job"

  • Les tâches mortes-vivantes sont des tâches qui ne sont pas censées être en cours d'exécution. Airflow recherche ces tâches périodiquement et les arrête.

Pour en savoir plus sur le dépannage des tâches zombies, consultez Causes racines courantes.

Signaux SIGTERM

Les signaux SIGTERM sont utilisés par Linux, Kubernetes, le programmateur Airflow et Celery pour arrêter les processus responsables de l'exécution des nœuds de calcul ou des tâches Airflow.

Plusieurs raisons peuvent expliquer l'envoi de signaux SIGTERM dans un environnement :

  • Une tâche est devenue une tâche zombie et doit être arrêtée.

  • Le planificateur a détecté un doublon d'une tâche et envoie les signaux Terminating instance et SIGTERM à la tâche pour l'arrêter.

  • Dans l'autoscaling horizontal des pods, le plan de contrôle GKE envoie des signaux SIGTERM pour supprimer les pods qui ne sont plus nécessaires.

  • Le planificateur peut envoyer des signaux SIGTERM au processus DagFileProcessorManager. Ces signaux SIGTERM sont utilisés par le planificateur pour gérer le cycle de vie du processus DagFileProcessorManager et peuvent être ignorés sans risque.

    Exemple :

    Launched DagFileProcessorManager with pid: 353002
Sending Signals.SIGTERM to group 353002. PIDs of all processes in the group: []
Sending the signal Signals.SIGTERM to group 353002
Sending the signal Signals.SIGTERM to process 353002 as process group is missing.

  • Condition de concurrence entre le rappel de battement de cœur et les rappels de sortie dans local_task_job, qui surveille l'exécution de la tâche. Si le signal de présence détecte qu'une tâche a été marquée comme réussie, il ne peut pas faire la différence entre une tâche qui a réellement réussi et une tâche qu'Airflow a été invité à considérer comme réussie. Néanmoins, il mettra fin à un exécuteur de tâches sans attendre qu'il se ferme.

    Vous pouvez ignorer ces signaux SIGTERM. La tâche est déjà à l'état "Réussie" et l'exécution de l'ensemble de l'exécution du DAG ne sera pas affectée.

    L'entrée de journal Received SIGTERM. est la seule différence entre la sortie normale et l'arrêt de la tâche dans l'état "Réussie".

    Condition de concurrence entre les rappels de battement de cœur et de sortie
    Figure 2. Condition de concurrence entre les rappels de signal de présence et de sortie (cliquez pour agrandir)

  • Un composant Airflow utilise plus de ressources (CPU, mémoire) que ce qui est autorisé par le nœud de cluster.

  • Le service GKE effectue des opérations de maintenance et envoie des signaux SIGTERM aux pods qui s'exécutent sur un nœud qui va être mis à niveau.

    Lorsqu'une instance de tâche est arrêtée avec SIGTERM, les entrées de journal suivantes s'affichent dans les journaux d'un nœud de calcul Airflow ayant exécuté la tâche :

    {local_task_job.py:211} WARNING - State of this instance has been externally
set to queued. Terminating instance. {taskinstance.py:1411} ERROR - Received
SIGTERM. Terminating subprocesses. {taskinstance.py:1703} ERROR - Task failed
with exception

Solutions possibles :

Ce problème se produit lorsqu'une VM exécutant la tâche manque de mémoire. Cela n'est pas lié aux configurations Airflow, mais à la quantité de mémoire disponible pour la VM.

  • Dans Managed Airflow (3e génération), vous pouvez attribuer davantage de ressources de processeur et de mémoire aux nœuds de calcul Airflow.

  • Vous pouvez réduire la valeur de l'option de configuration Airflow