Tester, synchroniser et déployer vos DAG depuis GitHub

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

Ce guide explique comment créer un pipeline CI/CD pour tester, synchroniser et déployer des DAG dans votre environnement Managed Airflow à partir de votre dépôt GitHub.

Si vous ne souhaitez synchroniser que les données d'autres services, consultez Transférer des données depuis d'autres services.

Présentation du pipeline CI/CD

Figure 1. Diagramme de l'architecture illustrant les étapes du flux (cliquez pour agrandir)

Le pipeline CI/CD permettant de tester, de synchroniser et de déployer des DAG comporte les étapes suivantes :

1. Vous apportez une modification à un DAG et la transférez vers une branche de développement de votre dépôt.

2. Vous ouvrez une demande d'extraction par rapport à la branche principale de votre dépôt.

3. Cloud Build exécute des tests unitaires pour vérifier que votre DAG est valide.

4. Votre demande d'extraction est approuvée et fusionnée dans la branche principale de votre dépôt.

5. Cloud Build synchronise votre environnement de développement Managed Airflow avec ces nouvelles modifications.

6. Vous vérifiez que le DAG fonctionne comme prévu dans votre environnement de développement.

7. Si votre DAG fonctionne comme prévu, vous l'importez dans votre environnement de production Managed Airflow.

Objectifs

• Exécuter une vérification automatisée avant l'envoi à l'aide de Cloud Build. Cette vérification exécute des tests unitaires pour un DAG.
• Synchroniser les DAG de votre environnement de développement Managed Service pour Apache Airflow avec les DAG de votre dépôt GitHub.

Avant de commencer

• Ce guide part du principe que vous travaillez avec deux environnements Managed Airflow identiques : un environnement de développement et un environnement de production.

  Pour les besoins de ce guide, vous ne configurez un pipeline CI/CD que pour l'environnement de développement. Assurez-vous que l'environnement que vous utilisez n'est pas un environnement de production.

• Ce guide part du principe que vos DAG et leurs tests sont stockés dans un dépôt GitHub.

  L'exemple de pipeline CI/CD illustre le contenu d'un exemple de dépôt. Les DAG et les tests sont stockés dans le répertoire dags/, avec les fichiers d'exigences, le fichier de contraintes et les fichiers de configuration de compilation Cloud Build stockés au niveau supérieur. L'utilitaire de synchronisation des DAG et ses exigences se trouvent dans le répertoire utils.

Créer un job de vérification avant l'envoi et des tests unitaires

Le premier job Cloud Build exécute une vérification avant l'envoi, qui exécute des tests unitaires pour vos DAG.

Ajouter des tests unitaires

Si ce n'est pas déjà fait, créez des tests unitaires pour vos DAG. Enregistrez ces tests avec les DAG dans votre dépôt, chacun avec le suffixe _test. Par exemple, le fichier de test du DAG dans example_dag.py est example_dag_test.py. Il s'agit des tests qui s'exécutent en tant que vérification avant l'envoi dans votre dépôt.

Créer une configuration YAML Cloud Build pour la vérification avant l'envoi

Dans votre dépôt, créez un fichier YAML nommé test-dags.cloudbuild.yaml qui configure votre job Cloud Build pour les vérifications avant l'envoi. Il comporte trois étapes :

1. Installer les dépendances requises par vos DAG.
2. Installer les dépendances requises par vos tests unitaires.
3. Exécuter les tests DAG.

steps:
  # install dependencies
  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements.txt", "-c", "constraints.txt", "--user"]

  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements-test.txt", "--user"]

  # run in python 3.8 which is latest version in Cloud Composer
  - name: python:3.8-slim
    entrypoint: python3.8
    args: ["-m", "pytest", "-s", "dags/"]

Remarque Cloud Build extrait l'image Python de Docker Hub. Vous pouvez spécifier le tag dans le champ name de chaque étape. Si vous utilisez python comme entrypoint, assurez-vous que votre entrypoint correspond à la version spécifiée dans votre tag. Sinon, entrypoint est défini par défaut sur latest.

Créer le déclencheur de compilation Cloud Build pour la vérification avant l'envoi

Suivez le guide Créer des dépôts à partir de GitHub pour créer un déclencheur basé sur une application GitHub avec les configurations suivantes :

• Nom : test-dags

• Événement : demande d'extraction

• Source - Dépôt : sélectionnez votre dépôt

• Source - Branche de base : ^main$ (remplacez main par le nom de la branche de base de votre dépôt, si nécessaire)

• Source - Contrôle des commentaires : non requis

  Attention : Si vous définissez le contrôle des commentaires sur "Non requis", tout utilisateur peut envoyer une demande d'extraction qui déclenche la compilation. Pour les dépôts plus volumineux ou ceux avec des contributeurs non approuvés, il est recommandé d'exiger le contrôle des commentaires.

• Configuration de compilation - Fichier de configuration Cloud Build : /test-dags.cloudbuild.yaml (chemin d'accès à votre fichier de compilation)

Créer un job de synchronisation des DAG et ajouter un script d'utilitaire de DAG

Ensuite, configurez un job Cloud Build qui exécute un script d'utilitaire de DAG. Le script d'utilitaire de ce job synchronise vos DAG avec votre environnement Managed Airflow une fois qu'ils sont fusionnés dans la branche principale de votre dépôt.

Remarque : Managed Airflow analyse tous les DAG à intervalles réguliers. Cela se produit même si les DAG n'ont pas changé depuis la dernière analyse. En raison de ce comportement, cet utilitaire importe tous les DAG dans Managed Airflow, et pas seulement ceux qui ont été modifiés.

Ajouter le script d'utilitaire de DAG

Ajoutez le script d'utilitaire de DAG à votre dépôt. Ce script d'utilitaire copie tous les fichiers DAG du répertoire dags/ de votre dépôt dans un répertoire temporaire, en ignorant tous les fichiers Python non DAG. Le script utilise ensuite la bibliothèque cliente Cloud Storage pour importer tous les fichiers de ce répertoire temporaire dans le répertoire dags/ du bucket de votre environnement Managed Airflow.

from __future__ import annotations

import argparse
import glob
import os
from shutil import copytree, ignore_patterns
import tempfile

# Imports the Google Cloud client library
from google.cloud import storage


def _create_dags_list(dags_directory: str) -> tuple[str, list[str]]:
    temp_dir = tempfile.mkdtemp()

    # ignore non-DAG Python files
    files_to_ignore = ignore_patterns("__init__.py", "*_test.py")

    # Copy everything but the ignored files to a temp directory
    copytree(dags_directory, f"{temp_dir}/", ignore=files_to_ignore, dirs_exist_ok=True)

    # The only Python files left in our temp directory are DAG files
    # so we can exclude all non Python files
    dags = glob.glob(f"{temp_dir}/*.py")
    return (temp_dir, dags)


def upload_dags_to_composer(
    dags_directory: str, bucket_name: str, name_replacement: str = "dags/"
) -> None:
    """
    Given a directory, this function moves all DAG files from that directory
    to a temporary directory, then uploads all contents of the temporary directory
    to a given cloud storage bucket
    Args:
        dags_directory (str): a fully qualified path to a directory that contains a "dags/" subdirectory
        bucket_name (str): the GCS bucket of the Cloud Composer environment to upload DAGs to
        name_replacement (str, optional): the name of the "dags/" subdirectory that will be used when constructing the temporary directory path name Defaults to "dags/".
    """
    temp_dir, dags = _create_dags_list(dags_directory)

    if len(dags) > 0:
        # Note - the GCS client library does not currently support batch requests on uploads
        # if you have a large number of files, consider using
        # the Python subprocess module to run gcloud storage cp --recursive on your dags
        # See https://cloud.google.com/storage/docs/gsutil/commands/cp for more info
        storage_client = storage.Client()
        bucket = storage_client.bucket(bucket_name)

        for dag in dags:
            # Remove path to temp dir
            dag = dag.replace(f"{temp_dir}/", name_replacement)

            try:
                # Upload to your bucket
                blob = bucket.blob(dag)
                blob.upload_from_filename(dag)
                print(f"File {dag} uploaded to {bucket_name}/{dag}.")
            except FileNotFoundError:
                current_directory = os.listdir()
                print(
                    f"{name_replacement} directory not found in {current_directory}, you may need to override the default value of name_replacement to point to a relative directory"
                )
                raise

    else:
        print("No DAGs to upload.")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--dags_directory",
        help="Relative path to the source directory containing your DAGs",
    )
    parser.add_argument(
        "--dags_bucket",
        help="Name of the DAGs bucket of your Composer environment without the gs:// prefix",
    )

    args = parser.parse_args()

    upload_dags_to_composer(args.dags_directory, args.dags_bucket)

Créer une configuration YAML Cloud Build pour la synchronisation des DAG

Dans votre dépôt, créez un fichier YAML nommé add-dags-to-composer.cloudbuild.yaml qui configure votre job Cloud Build pour la synchronisation des DAG. Il comporte deux étapes :

1. Installer les dépendances requises par le script d'utilitaire de DAG.

2. Exécuter le script d'utilitaire pour synchroniser les DAG de votre dépôt avec votre environnement Managed Airflow.

steps:
  # install dependencies
  - name: python
    entrypoint: pip
    args: ["install", "-r", "utils/requirements.txt", "--user"]

  # run
  - name: python
    entrypoint: python
    args: ["utils/add_dags_to_composer.py", "--dags_directory=${_DAGS_DIRECTORY}", "--dags_bucket=${_DAGS_BUCKET}"]

Remarque Cloud Build extrait l'image Python de Docker Hub. Vous pouvez spécifier le tag dans le champ name de chaque étape. Si vous utilisez python comme entrypoint, assurez-vous que votre entrypoint correspond à la version spécifiée dans votre tag. Sinon, entrypoint est défini par défaut sur latest.

Créer le déclencheur de compilation Cloud Build

Suivez le guide Créer des dépôts à partir de GitHub pour créer un déclencheur basé sur une application GitHub avec les configurations suivantes :

• Nom : add-dags-to-composer

• Événement : déployer sur une branche

• Source - Dépôt : sélectionnez votre dépôt

• Source - Branche de base : ^main$ (remplacez main par le nom de la branche de base de votre dépôt, si nécessaire)

• Source - Filtre par fichiers inclus (glob) : dags/**

• Configuration de compilation - Fichier de configuration Cloud Build : /add-dags-to-composer.cloudbuild.yaml (chemin d'accès à votre fichier de compilation)

Dans la configuration avancée, ajoutez deux variables de substitution :

• _DAGS_DIRECTORY : répertoire dans lequel se trouvent les DAG dans votre dépôt. Si vous utilisez l'exemple de dépôt de ce guide, il s'agit de dags/.

• _DAGS_BUCKET : bucket Cloud Storage contenant le dags/ répertoire dans votre environnement de développement Managed Airflow environnement. Omettez le préfixe gs://. Par exemple : us-central1-example-env-1234ab56-bucket.

Attention : Si vous ne supprimez pas le suffixe dags/ du nom du bucket Cloud Storage dans la variable de substitution _DAGS_BUCKET, le job de synchronisation des DAG échoue avec une erreur NotFound. En effet, dans Cloud Storage, dags/ est un dossier considéré comme faisant partie du nom de l'objet, et non du nom du bucket.

Tester votre pipeline CI/CD

Dans cette section, suivez un flux de développement de DAG qui utilise les déclencheurs Cloud Build que vous venez de créer.

Exécuter un job avant l'envoi

Créez une demande d'extraction dans votre branche principale pour tester votre compilation. Recherchez votre vérification avant l'envoi sur la page. Cliquez sur Détails , puis sélectionnez Afficher plus de détails sur Google Cloud Build pour afficher les journaux de compilation dans la Google Cloud console.

Important : Cloud Build exécute des tests unitaires avec les packages spécifiés dans requirements.txt et requirements-test.txt, qui incluent une version publique d'Airflow. Votre environnement Managed Airflow peut contenir un ensemble différent de packages PyPI et de packages préinstallés. Nous vous recommandons d' exécuter un environnement Airflow local lors du développement de vos DAG.

Figure 2. Capture d'écran de l'état de la vérification avant l'envoi Cloud Build sur GitHub (cliquez pour agrandir)

Si votre vérification avant l'envoi a échoué, consultez Résoudre les échecs de compilation.

Vérifier que votre DAG fonctionne dans votre environnement de développement

Une fois votre demande d'extraction approuvée, fusionnez-la dans votre branche principale. Affichez vos résultats de compilation à l'aide de la Google Cloud console. Si vous disposez de nombreux déclencheurs Cloud Build, vous pouvez filtrer vos compilations sur le nom du déclencheur add-dags-to-composer.

Une fois le job de synchronisation Cloud Build réussi, le DAG synchronisé s'affiche dans votre environnement de développement Managed Airflow. Vous pouvez alors vérifier que le DAG fonctionne comme prévu.

Ajouter le DAG à votre environnement de production

Une fois que le DAG fonctionne comme prévu, ajoutez-le manuellement à votre environnement de production. Pour ce faire, importez le fichier DAG dans le répertoire dags/ du bucket de votre environnement de production Managed Airflow.

Si votre job de synchronisation des DAG a échoué ou si votre DAG ne se comporte pas comme prévu dans votre environnement de développement Managed Airflow, consultez Résoudre les échecs de compilation.

Résoudre les échecs de compilation

Cette section explique comment résoudre les scénarios d'échec de compilation courants.

Que faire si ma vérification avant l'envoi a échoué ?

Dans votre demande d'extraction, cliquez sur Détails , puis sélectionnez Afficher plus de détails sur Google Cloud Build pour afficher les journaux de compilation dans la Google Cloud console. Utilisez ces journaux pour vous aider à déboguer le problème lié à votre DAG. Une fois les problèmes résolus, validez le correctif et transférez-le vers votre branche. La vérification avant l'envoi s'exécute à nouveau, et vous pouvez continuer à itérer à l'aide des journaux comme outil de débogage.

Que faire si mon job de synchronisation des DAG a échoué ?

Affichez vos résultats de compilation à l'aide de la Google Cloud console . Si vous disposez de nombreux déclencheurs Cloud Build, vous pouvez filtrer vos compilations sur le nom du déclencheur add-dags-to-composer. Examinez les journaux du job de compilation et résolvez les erreurs. Si vous avez besoin d'aide supplémentaire pour résoudre les erreurs, utilisez les canaux d'assistance.

Que faire si mon DAG ne fonctionne pas correctement dans mon environnement Managed Airflow ?

Si votre DAG ne fonctionne pas comme prévu dans votre environnement de développement Managed Airflow, ne le promouvez pas manuellement dans votre environnement de production Managed Airflow. Effectuez plutôt l'une des opérations suivantes :

• Restaurez la demande d'extraction avec les modifications qui ont interrompu votre DAG pour le rétablir à l'état immédiatement antérieur à vos modifications (cela restaure également tous les autres fichiers de cette demande d'extraction).
• Créez une demande d'extraction pour rétablir manuellement les modifications apportées au DAG interrompu.
• Créez une demande d'extraction pour corriger les erreurs dans votre DAG.

L'une de ces étapes déclenche une nouvelle vérification avant l'envoi et, lors de la fusion, le job de synchronisation des DAG.

Étape suivante

• Exécuter des environnements Airflow locaux
• Écrire des DAG
• Planifier et déclencher des DAG
• Tester des DAG