Testare, sincronizzare ed eseguire il deployment dei DAG da GitHub

Managed Airflow (Gen 3) | Managed Airflow (Gen 2) | Managed Airflow (Legacy Gen 1)

Questa guida spiega come creare una pipeline CI/CD per testare, sincronizzare ed eseguire il deployment dei DAG nel tuo ambiente Managed Airflow dal tuo repository GitHub.

Se vuoi sincronizzare solo i dati di altri servizi, consulta Trasferire dati da altri servizi.

Panoramica della pipeline CI/CD

Figura 1. Diagramma dell'architettura che mostra i passaggi del flusso (fai clic per ingrandire)

La pipeline CI/CD per testare, sincronizzare ed eseguire il deployment dei DAG prevede i seguenti passaggi:

1. Apporti una modifica a un DAG e la esegui il push in un ramo di sviluppo nel repository.

2. Apri una richiesta di pull nel ramo principale del repository.

3. Cloud Build esegue test delle unità per verificare che il DAG sia valido.

4. La richiesta di pull viene approvata e unita al ramo principale del tuo repository.

5. Cloud Build sincronizza l'ambiente Managed Airflow di sviluppo con queste nuove modifiche.

6. Verifichi che il DAG funzioni come previsto nell'ambiente di sviluppo ambiente.

7. Se il DAG funziona come previsto, lo carichi nell'ambiente Managed Airflow di produzione.

Obiettivi

• Eseguire un controllo di pre-invio automatico utilizzando Cloud Build. Questo controllo esegue test delle unità per un DAG.
• Sincronizzare i DAG nell'ambiente Managed Service per Apache Airflow di sviluppo con i DAG nel repository GitHub.

Prima di iniziare

• Questa guida presuppone che tu stia lavorando con due ambienti Managed Airflow identici: un ambiente di sviluppo e un ambiente di produzione.

  Ai fini di questa guida, configurerai una pipeline CI/CD solo per l'ambiente di sviluppo. Assicurati che l'ambiente che utilizzi non sia un ambiente di produzione.

• Questa guida presuppone che tu abbia archiviato i DAG e i relativi test in un repository GitHub.

  La pipeline CI/CD di esempio mostra i contenuti di un repository di esempio. I DAG e i test sono archiviati nella directory dags/, con i file dei requisiti, il file dei vincoli e i file di configurazione di Cloud Build archiviati al livello superiore. L'utilità di sincronizzazione dei DAG e i relativi requisiti si trovano nella directory utils.

Creare un job di controllo di pre-invio e test delle unità

Il primo job di Cloud Build esegue un controllo di pre-invio, che esegue test delle unità per i DAG.

Aggiungere test delle unità

Se non l'hai ancora fatto, crea test delle unità per i DAG. Salva questi test insieme ai DAG nel repository, ognuno con il suffisso _test. Ad esempio, il file di test per il DAG in example_dag.py è example_dag_test.py. Questi sono i test eseguiti come controllo di pre-invio nel repository.

Creare la configurazione YAML di Cloud Build per il controllo di pre-invio

Nel repository, crea un file YAML denominato test-dags.cloudbuild.yaml che configura il job di Cloud Build per i controlli di pre-invio. Contiene tre passaggi:

1. Installa le dipendenze richieste dai DAG.
2. Installa le dipendenze richieste dai test delle unità.
3. Esegui i test dei 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/"]

Nota: Cloud Build estrae l'immagine Python da Docker Hub e puoi specificare il tag nel campo name di ogni passaggio. Se utilizzi python come entrypoint, assicurati che entrypoint corrisponda alla versione specificata nel tag, altrimenti entrypoint assume il valore predefinito latest.

Creare il trigger di Cloud Build per il controllo di pre-invio

Segui la guida Creare repository da GitHub per creare un trigger basato su app GitHub con le seguenti configurazioni:

• Nome: test-dags

• Evento: richiesta di pull

• Origine - Repository: scegli il repository

• Origine - Ramo di base: ^main$ (se necessario, sostituisci main con il nome del ramo di base del tuo repository)

• Origine - Controllo dei commenti: non richiesto

  **Attenzione:** se imposti il controllo dei commenti su Non richiesto, qualsiasi utente può inviare una richiesta di pull che attiva la build. Per i repository più grandi o con contributori non attendibili, è consigliabile richiedere il controllo dei commenti.

• Configurazione di compilazione - File di configurazione di Cloud Build: /test-dags.cloudbuild.yaml (il percorso del file di build)

Creare un job di sincronizzazione dei DAG e aggiungere lo script dell'utilità dei DAG

Poi, configura un job di Cloud Build che esegue uno script dell'utilità dei DAG. Lo script dell'utilità in questo job sincronizza i DAG con l'ambiente Managed Airflow dopo che sono stati uniti al ramo principale del repository.

Nota: Managed Airflow analizza tutti i DAG a intervalli ripetuti. Questo accade anche se i DAG non sono stati modificati dall'ultima analisi. A causa di questo comportamento, questa utilità carica tutti i DAG in Managed Airflow, non solo quelli modificati.

Aggiungere lo script dell'utilità dei DAG

Aggiungi lo script dell'utilità dei DAG al repository. Questo script dell'utilità copia tutti i file DAG nella directory dags/ del repository in una directory temporanea, ignorando tutti i file Python non DAG. Lo script utilizza quindi la libreria client di Cloud Storage per caricare tutti i file dalla directory temporanea alla directory dags/ nel bucket dell'ambiente 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)

Creare la configurazione YAML di Cloud Build per la sincronizzazione dei DAG

Nel repository, crea un file YAML denominato add-dags-to-composer.cloudbuild.yaml che configura il job di Cloud Build per la sincronizzazione dei DAG. Contiene due passaggi:

1. Installa le dipendenze richieste dallo script dell'utilità dei DAG.

2. Esegui lo script dell'utilità per sincronizzare i DAG nel repository con l'ambiente 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}"]

Nota: Cloud Build estrae l'immagine Python da Docker Hub e puoi specificare il tag nel campo name di ogni passaggio. Se utilizzi python come entrypoint, assicurati che entrypoint corrisponda alla versione specificata nel tag, altrimenti entrypoint assume il valore predefinito latest.

Creare il trigger di Cloud Build

Segui la guida Creare repository da GitHub per creare un trigger basato su app GitHub con le seguenti configurazioni:

• Nome: add-dags-to-composer

• Evento: esegui il push sul ramo

• Origine - Repository: scegli il repository

• Origine - Ramo di base: ^main$ (se necessario, sostituisci main con il nome del ramo di base del tuo repository)

• Origine - Filtro dei file inclusi (glob): dags/**

• Configurazione di compilazione - File di configurazione di Cloud Build: /add-dags-to-composer.cloudbuild.yaml (il percorso del file di build)

Nella configurazione avanzata, aggiungi due variabili di sostituzione:

• _DAGS_DIRECTORY : la directory in cui si trovano i DAG nel repository. Se utilizzi il repository di esempio di questa guida, è dags/.

• _DAGS_BUCKET : il bucket Cloud Storage che contiene la dags/ directory nell'ambiente Managed Airflow di sviluppo. Ometti il prefisso gs://. Ad esempio: us-central1-example-env-1234ab56-bucket.

Attenzione: se non rimuovi il suffisso dags/ dal nome del bucket Cloud Storage nella variabile di sostituzione _DAGS_BUCKET, il job di sincronizzazione dei DAG non riesce e viene visualizzato un errore NotFound. Questo accade perché in Cloud Storage, dags/ è una cartella, considerata parte del nome dell'oggetto, non del nome del bucket.

Testare la pipeline CI/CD

In questa sezione, segui un flusso di sviluppo dei DAG che utilizza i trigger di Cloud Build appena creati.

Eseguire un job di pre-invio

Crea una richiesta di pull nel ramo principale per testare la build. Individua il controllo di pre-invio nella pagina. Fai clic su Dettagli e scegli Visualizza ulteriori dettagli su Google Cloud Build per visualizzare i log di build nella Google Cloud console.

Importante: Cloud Build esegue test delle unità con i pacchetti specificati in requirements.txt e requirements-test.txt, che includono una versione pubblica di Airflow. L'ambiente Managed Airflow potrebbe avere un diverso insieme di pacchetti PyPI e pacchetti preinstallati. Ti consigliamo di eseguire un ambiente Airflow locale durante lo sviluppo dei DAG.

Figura 2. Screenshot dello stato del controllo di pre-invio di Cloud Build su GitHub (fai clic per ingrandire)

Se il controllo di pre-invio non è riuscito, consulta Risolvere gli errori di build.

Verificare che il DAG funzioni nell'ambiente di sviluppo

Una volta approvata la richiesta di pull, uniscila al ramo principale. Utilizza la Google Cloud console per visualizzare i risultati della build. Se hai molti trigger di Cloud Build, puoi filtrare le build in base al nome del trigger add-dags-to-composer.

Una volta completato il job di sincronizzazione di Cloud Build, il DAG sincronizzato viene visualizzato nell'ambiente Managed Airflow di sviluppo. Lì, puoi verificare che il DAG funzioni come previsto.

Aggiungere il DAG all'ambiente di produzione

Una volta verificato che il DAG funzioni come previsto, aggiungilo manualmente all'ambiente di produzione. Per farlo, carica il file DAG nella directory dags/ nel bucket dell'ambiente Managed Airflow di produzione.

Se il job di sincronizzazione dei DAG non è riuscito o se il DAG non funziona come previsto nell'ambiente Managed Airflow di sviluppo, consulta Risolvere gli errori di build.

Risolvere gli errori di build

Questa sezione spiega come risolvere gli scenari di errore di build comuni.

Cosa succede se il controllo di pre-invio non è riuscito?

Dalla richiesta di pull, fai clic Dettagli e scegli Visualizza ulteriori dettagli su Google Cloud Build per visualizzare i log di build nella Google Cloud console. Utilizza questi log per eseguire il debug del problema con il tuo DAG. Una volta risolti i problemi, esegui il commit della correzione e il push nel ramo. Il controllo di pre-invio viene eseguito di nuovo e puoi continuare a eseguire l'iterazione utilizzando i log come strumento di debug.

Cosa succede se il job di sincronizzazione dei DAG non è riuscito?

Utilizza la Google Cloud console per visualizzare i risultati della build. Se hai molti trigger di Cloud Build, puoi filtrare le build in base al nome del trigger add-dags-to-composer. Esamina i log del job di build e risolvi gli errori. Se hai bisogno di ulteriore assistenza per risolvere gli errori, utilizza i canali di assistenza.

Cosa succede se il DAG non funziona correttamente nell'ambiente Managed Airflow?

Se il DAG non funziona come previsto nell'ambiente Managed Airflow di sviluppo, non promuoverlo manualmente all'ambiente Managed Airflow di produzione. Procedi invece in uno dei seguenti modi:

• Ripristina la richiesta di pull con le modifiche che hanno interrotto il tuo DAG allo stato immediatamente precedente alle modifiche (in questo modo vengono ripristinati anche tutti gli altri file nella richiesta di pull).
• Crea una nuova richiesta di pull per ripristinare manualmente le modifiche al DAG interrotto.
• Crea una nuova richiesta di pull per correggere gli errori nel DAG.

Se segui uno di questi passaggi, viene attivato un nuovo controllo di pre-invio e, dopo l'unione, il job di sincronizzazione dei DAG.

Passaggi successivi

• Eseguire ambienti Airflow locali
• Scrivere i DAG
• Pianificare e attivare i DAG
• Testare i DAG