Testa, sincronizza ed esegui il deployment dei DAG da GitHub

Cloud Composer 1 | Cloud Composer 2

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

Panoramica della pipeline CI/CD

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

La pipeline CI/CD con cui eseguire il test, la sincronizzazione e il deployment dei DAG prevede i seguenti passaggi:

1. Puoi apportare una modifica a un DAG ed eseguirne il push a un ramo di sviluppo nel tuo repository.

2. Apri una richiesta di pull sul ramo principale del tuo repository.

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

4. La richiesta di pull è stata approvata e unita al ramo principale del repository.

5. Cloud Build sincronizza il tuo ambiente di sviluppo Cloud Composer con queste nuove modifiche.

6. Verifica che il DAG funzioni come previsto nel tuo ambiente di sviluppo.

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

Obiettivi

Obiettivo:

• Eseguire un controllo automatico prima dell'invio utilizzando Cloud Build. Questo controllo esegue i test delle unità per un DAG.
• Sincronizza i DAG nel tuo ambiente di sviluppo Cloud Composer con i DAG nel tuo repository GitHub.

Prima di iniziare

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

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

• Questa guida presuppone che i tuoi DAG e i relativi test siano archiviati 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 più alto. L'utilità di sincronizzazione DAG e i relativi requisiti si trovano nella directory utils.

  Questa struttura può essere utilizzata per gli ambienti Airflow 1, Airflow 2, Cloud Composer 1 e Cloud Composer 2.

Creare un job di controllo prima dell'invio e test delle unità

Il primo job Cloud Build esegue un controllo preinvio, che esegue i test delle unità per i DAG.

Aggiungi test delle unità

Se non lo hai già fatto, crea i test delle unità per i DAG. Salva questi test insieme ai DAG nel repository, ciascuno 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 che vengono eseguiti come controllo pre-invio nel tuo repository.

Crea la configurazione YAML Cloud Build per il controllo preinvio

Nel tuo repository, crea un file YAML denominato test-dags.cloudbuild.yaml che configuri il job Cloud Build per i controlli preinvio. Sono previsti tre passaggi:

1. Installa le dipendenze necessarie per i DAG.
2. Installa le dipendenze necessarie per i test delle unità.
3. Esegui i test DAG.

Composer/cicd_sample/test-dags.cloudbuild.yaml

Visualizza su GitHub Feedback

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 sarà latest per impostazione predefinita.

Crea il trigger di Cloud Build per il controllo preinvio

Segui la guida Creazione di 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$ (cambia main con il nome del ramo di base del tuo repository, se necessario)

• Origine - Controllo commenti: non obbligatorio.

  Attenzione: l'impostazione del controllo dei commenti come non obbligatorio significa che qualsiasi utente può inviare una richiesta di pull che attiva la build. Per repository più grandi o repository con collaboratori non attendibili, è consigliabile richiedere il controllo dei commenti.

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

Crea un job di sincronizzazione DAG e aggiungi lo script di utilità DAG

Quindi, configura un job Cloud Build che esegue uno script di utilità dei DAG. Lo script di utilità in questo job sincronizza i DAG con il tuo ambiente Cloud Composer dopo che sono stati uniti al ramo principale nel repository.

Nota: Cloud Composer analizza tutti i DAG in base a un intervallo ripetuto. Questo si verifica anche se i DAG non sono cambiati dall'ultima analisi. A causa di questo comportamento, questa utilità carica tutti i DAG in Cloud Composer, non solo quelli che sono cambiati.

Aggiungere lo script dell'utilità DAG

Aggiungi lo script dell'utilità DAG al repository. Questo script di utilità copia tutti i file DAG della directory dags/ del tuo 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 da quella directory temporanea alla directory dags/ nel bucket del tuo ambiente Cloud Composer.

composer/cicd_sample/utils/add_dags_to_composer.py

Visualizza su GitHub Feedback

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 gsutil -m cp -r 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)

Crea una configurazione YAML Cloud Build per la sincronizzazione dei DAG

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

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

2. Esegui lo script di utilità per sincronizzare i DAG nel tuo repository con il tuo ambiente Cloud Composer.

composer/cicd_sample/add-dags-to-composer.cloudbuild.yaml

Visualizza su GitHub Feedback

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 sarà latest per impostazione predefinita.

Crea il trigger di Cloud Build

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

• Nome: add-dags-to-composer

• Evento: push a un ramo

• Origine - Repository: scegli il repository

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

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

• Configurazione build - 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 DAG nel repository. Se stai utilizzando il repository di esempio di questa guida, è dags/.

• _DAGS_BUCKET: il bucket Cloud Storage che contiene la directory dags/ nel tuo ambiente di sviluppo Cloud Composer. 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 DAG non riesce e restituisce un errore NotFound. Questo accade perché in Cloud Storage dags/ è una cartella, che è considerata parte del nome dell'oggetto, non del nome del bucket.

Testa la tua pipeline CI/CD

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

Esecuzione di un job di preinvio

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

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. Il tuo ambiente Cloud Composer potrebbe avere un set diverso 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 preinvio di Cloud Build su GitHub (fai clic per ingrandire)

Se il controllo prima dell'invio non è riuscito, consulta Risolvere gli errori della build.

Verifica che il tuo DAG funzioni nel tuo ambiente di sviluppo Cloud Composer

Dopo l'approvazione della richiesta di pull, uniscila al ramo principale. Utilizza la console Google Cloud per visualizzare i risultati della build. Se disponi di 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 nel tuo ambiente di sviluppo Cloud Composer. Qui puoi convalidare il funzionamento del DAG come previsto.

Aggiungere il DAG all'ambiente di produzione

Dopo che il DAG funziona come previsto, aggiungilo manualmente all'ambiente di produzione. Per farlo, carica il file DAG nella directory dags/ del bucket dell'ambiente Cloud Composer di produzione.

Se il job di sincronizzazione DAG non è riuscito o se il DAG non si comporta come previsto nell'ambiente di sviluppo Cloud Composer, consulta la sezione Risolvere gli errori delle build.

Gestione degli errori di build

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

Che cosa succede se l'assegno prima dell'invio non è andato a buon fine?

Nella richiesta di pull, fai clic su Dettagli e scegli Visualizza altri dettagli su Google Cloud Build per visualizzare i log di build nella console Google Cloud. Utilizza questi log per eseguire il debug del problema con il DAG. Dopo aver risolto i problemi, esegui il commit della correzione ed esegui il push al tuo ramo. Il controllo preinvio 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 DAG non è riuscito?

Utilizza la console Google Cloud per visualizzare i risultati della build. Se disponi di 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 mio DAG non funziona correttamente nel mio ambiente Cloud Composer?

Se il DAG non funziona come previsto nel tuo ambiente di sviluppo Cloud Composer, non promuoverlo manualmente nell'ambiente Cloud Composer di produzione. Procedi invece in uno dei seguenti modi:

• Ripristina la richiesta di pull con le modifiche che hanno danneggiato il DAG per ripristinarlo allo stato immediatamente prima delle modifiche (vengono ripristinati anche tutti gli altri file nella richiesta di pull).
• Crea una nuova richiesta di pull per ripristinare manualmente le modifiche al DAG non funzionante.
• Crea una nuova richiesta di pull per correggere gli errori nel DAG.

Seguendo uno di questi passaggi viene attivato un nuovo controllo prima dell'invio e, in caso di unione, il job di sincronizzazione DAG.

Passaggi successivi

• Esegui ambienti Airflow locali
• Scrittura di DAG
• Attiva i DAG
• Testare i DAG