Testar, sincronizar e implantar os DAGs usando o controle de versões

Neste guia, explicamos como criar um fluxo de CI/CD para sincronizar os DAGs no ambiente do Cloud Composer com o sistema de controle de versões. Esse fluxo tem as seguintes etapas:

  1. Você faz uma alteração em um DAG e envia-a para uma ramificação de desenvolvimento no seu repositório
  2. Você abre uma solicitação de envio na ramificação principal do repositório
  3. O Cloud Build executa testes de unidade para verificar se o DAG é válido
  4. Sua solicitação de envio será aprovada e mesclada com a ramificação principal
  5. O Cloud Build sincroniza o ambiente de desenvolvimento do Cloud Composer com essas novas alterações.
  6. Verifique se o DAG se comporta conforme o esperado no seu ambiente de desenvolvimento
  7. Se o DAG funcionar como esperado, sincronize-o manualmente com o ambiente de produção do Cloud Composer
Diagrama da arquitetura mostrando as etapas do fluxo. A revisão de pré-envio e PR está em uma seção do GitHub, e a sincronização e a verificação manual de DAGs estão em uma seção do GCP.
Figura 1. Diagrama de arquitetura mostrando as etapas do fluxo (clique para ampliar)

Objetivos

  • Execute uma verificação automatizada de pré-envio para testes de unidade DAG usando o Cloud Build. Essa verificação executa testes de unidade para um DAG.
  • Sincronizar DAGs no ambiente de desenvolvimento do Cloud Composer com DAGs no sistema de controle de versões

Custos

Neste tutorial, usamos os seguintes componentes faturáveis do Google Cloud:

Para gerar uma estimativa de custo baseada na projeção de uso deste tutorial, use a calculadora de preços. Novos usuários do Google Cloud podem estar qualificados para uma avaliação gratuita.

Ao concluir este tutorial, exclua os recursos criados para evitar o faturamento contínuo. Para mais informações, consulte Limpeza.

Novos usuários do Google Cloud podem estar qualificados para um teste gratuito.

Antes de começar

Neste tutorial, presumimos que os DAGs e os testes estejam armazenados em um repositório do GitHub. No diretório composer/cicd_sample, que inclui o conteúdo de um repositório de exemplo, os DAGs e testes são armazenados em uma pasta dags, com arquivos de requisitos, de restrições e de configuração do Cloud Build armazenados no nível superior. O utilitário de sincronização do DAG e os requisitos dele estão na pasta utils. Essa estrutura pode ser usada para os ambientes Airflow 1, Airflow 2, Cloud Composer 1 e Cloud Composer 2.

Além disso, neste tutorial, presumimos que você esteja trabalhando com dois ambientes idênticos do Cloud Composer: um ambiente de desenvolvimento e um de produção.

  1. No console do Google Cloud, na página do seletor de projetos, selecione ou crie um projeto do Google Cloud.

    Acessar o seletor de projetos

  2. Verifique se o faturamento está ativado para seu projeto na nuvem. Saiba como verificar se o faturamento está ativado em um projeto.

  3. Se você não tiver criado os ambientes de desenvolvimento e produção do Cloud Composer, faça isso agora. É possível criar ambientes do Cloud Composer 2 ou do Cloud Composer 1 neste tutorial.

Prepare o ambiente

Nesta seção, você vai configurar dois jobs do Cloud Build. O primeiro job executa uma verificação de pré-envio que executa testes de unidade nos DAGs. O segundo job sincroniza os DAGs com o ambiente do Cloud Composer após eles serem mesclados com a ramificação principal no repositório.

Adicionar uma verificação de pré-envio

Adicionar testes de unidade

Crie testes de unidade para seus DAGs, caso ainda não tenha feito isso. Salve esses testes com os DAGs no repositório, cada um com o sufixo _test. Por exemplo, o arquivo de teste do DAG em example_dag.py é example_dag_test.py. Esses testes serão executados como uma verificação de pré-envio no seu repositório.

Adicionar uma verificação de pré-envio do Cloud Build

Nesta seção, você criará uma configuração de YAML do Cloud Build e um gatilho do Cloud Build para a verificação de pré-envio.

Criar a configuração YAML do Cloud Build

Crie um arquivo YAML para configurar o job do Cloud Build chamado test-dags.cloudbuild.yaml. Nele, há três etapas:

  1. Instale as dependências necessárias para os DAGs
  2. Instale as dependências necessárias para os testes de unidade
  3. Executar os testes do 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/"]
    
Crie o gatilho do Cloud Build

Siga o guia sobre como criar repositórios do GitHub para criar um gatilho baseado em aplicativo do GitHub com as seguintes configurações:

  • Nome: test-dags
  • Evento: solicitação de envio
  • Source - Repository: escolha seu repositório
  • Origem: ramificação base: ^main$ (alterar main para o nome da ramificação base do seu repositório, se necessário)
  • Origem - Controle dos comentários: não obrigatório
  • Configuração da compilação: arquivo de configuração do build do Cloud: /test-dags.cloudbuild.yaml (o caminho para seu arquivo de build)

Adicionar um job de sincronização do DAG

Em seguida, você vai configurar um job do Cloud Build que executa um script utilitário. O script de utilitário neste job sincroniza os DAGs com o ambiente do Cloud Composer após eles serem mesclados à ramificação principal no repositório.

Adicionar script de utilitário de DAGs

Esse script de utilitário copia todos os arquivos DAG no diretório dags/ do repositório para um diretório temporário, ignorando todos os arquivos não DAG do Python. Em seguida, o script usa a biblioteca de cliente do Cloud Storage para fazer upload de todos os arquivos desse diretório temporário para a pasta dags/ no ambiente do Cloud Composer.

import argparse
import glob
import os
from shutil import copytree, ignore_patterns
import tempfile
from typing import List, Tuple

# 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) -> None:
    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
            # if/else is a relative directory workaround for our tests
            current_directory = os.listdir()
            if "dags" in current_directory:
                dag = dag.replace(f"{temp_dir}/", "dags/")
            else:
                dag = dag.replace(f"{temp_dir}/", "../dags/")

            # Upload to your bucket
            blob = bucket.blob(dag)
            blob.upload_from_filename(dag)
            print(f"File {dag} uploaded to {bucket_name}/{dag}.")

    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)

Adicionar um job do Cloud Build para sincronizar os DAGs

Nesta seção, você criará uma configuração YAML do Cloud Build e um gatilho do Cloud Build para o job de sincronização do DAG.

Criar a configuração YAML do Cloud Build

Crie um arquivo YAML para configurar o job do Cloud Build chamado add-dags-to-composer.cloudbuild.yaml. Nele, há duas etapas:

  1. Instalar as dependências necessárias ao script de utilitário dos DAGs
  2. Executar o script de utilitário para sincronizar os DAGs no repositório com o ambiente do Cloud Composer

    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}"]
    
Crie o gatilho do Cloud Build

Seguindo este guia, crie um acionador baseado em aplicativo do GitHub com as seguintes configurações:

  • Nome: add-dags-to-composer
  • Evento: push para uma ramificação
  • Source - Repository: escolha seu repositório
  • Origem: ramificação base: ^main$ (alterar main para o nome da ramificação base do seu repositório, se necessário)
  • Origem - Filtro de arquivos incluídos (glob): dags/**
  • Configuração da compilação: arquivo de configuração do build do Cloud: /add-dags-to-composer.cloudbuild.yaml (o caminho para seu arquivo de build)

Na configuração avançada, adicione duas variáveis de substituição:

  • _DAGS_DIRECTORY: o diretório onde os dags estão localizados no seu repositório. Se você estiver seguindo este tutorial exatamente, é dags/.
  • _DAGS_BUCKET: o bucket do Cloud Storage que contém a pasta dags/ no ambiente de desenvolvimento do Cloud Composer sem o prefixo gs://. Exemplo: us-east1-my-env-1234ab56-bucket.

Como tudo funciona em conjunto

Nesta seção, você seguirá um fluxo de desenvolvimento do DAG que usa os gatilhos recém-criados do Cloud Build.

Executar um job de pré-envio

Crie uma solicitação de envio para sua ramificação principal para testar o build. O cheque estará na parte inferior da página. Clique em "Detalhes" e escolha "Ver mais detalhes no Google Cloud Build" para ver os registros da versão no Console do Google Cloud.

Captura de tela de uma verificação do GitHub chamada "test-dags" com uma seta vermelha apontando para o nome do projeto entre parênteses
Figura 2. Captura de tela do status da verificação de pré-envio do Cloud Build no GitHub (clique para ampliar)

Consulte a seção Como solucionar falhas de build se a verificação de pré-envio falhar.

Validar o sucesso do DAG no ambiente de desenvolvimento do Cloud Composer

Quando a solicitação for aprovada, mescle-a à ramificação principal. Use o Console do Google Cloud para ver os resultados da versão. Se você tiver muitos gatilhos do Cloud Build, poderá filtrar suas builds pelo nome do gatilho add-dags-to-composer.

Depois que o job de sincronização do Cloud Build for concluído, o DAG será exibido no seu ambiente de desenvolvimento do Cloud Composer. Lá, é possível validar se o DAG funciona conforme o esperado. Quando o DAG tiver o desempenho esperado, promova-o no ambiente de produção adicionando manualmente o arquivo DAG à pasta dags/ no ambiente de produção do Cloud Composer.

Consulte a seção Como resolver falhas de build se o job de sincronização do DAG falhar ou se o DAG não estiver se comportando conforme o esperado no ambiente de desenvolvimento do Cloud Composer.

Como resolver falhas no build

Nesta seção, você aprenderá a lidar com cenários comuns de falha de compilação.

E se minha verificação de pré-envio falhar?

Na solicitação, clique em "Detalhes" e escolha "Ver mais detalhes no Google Cloud Build" para ver os registros do build no Console do Google Cloud. Use esses registros para depurar o problema com seus DAGs. Depois de resolver os problemas, confirme e envie a correção para sua ramificação. A verificação de pré-envio será executada novamente e será possível continuar a iteração com os registros como uma ferramenta de depuração.

E se o job de sincronização do DAG falhar?

Use o Console do Google Cloud para ver os resultados da versão. Se você tiver muitos gatilhos do Cloud Build, poderá filtrar suas builds pelo nome do gatilho add-dags-to-composer. Examinar os registros do job de compilação e resolver os erros. Se você precisar de mais ajuda para resolver os erros, use os canais de suporte.

E se o DAG não funcionar corretamente no meu ambiente do Cloud Composer?

Se o DAG não funcionar como esperado no ambiente de desenvolvimento do Cloud Composer, não o promova manualmente para o ambiente de produção. Em vez disso, siga um destes procedimentos:

  1. Reverter a solicitação de envio com as alterações que interromperam o DAG para restaurá-lo para o estado imediatamente anterior às alterações. Isso também reverterá todos os outros arquivos na solicitação.
  2. Criar uma nova solicitação de envio para reverter manualmente as alterações no DAG corrompido
  3. Crie uma nova solicitação de envio para corrigir os erros no DAG

Seguir qualquer uma dessas etapas acionará novamente a verificação de pré-envio e, após a mesclagem, o job de sincronização do DAG.

Limpar

Para evitar cobranças na sua conta do Google Cloud pelos recursos usados no tutorial, exclua o projeto que os contém ou mantenha o projeto e exclua os recursos individuais.

Excluir o projeto

  1. No console do Cloud, acesse a página Gerenciar recursos:

    Acessar "Gerenciar recursos"

  2. Na lista de projetos, selecione o projeto que você quer excluir e clique em Excluir .
  3. Na caixa de diálogo, digite o ID do projeto e clique em Encerrar para excluí-lo.

Excluir recursos individuais

Limpar recursos do Cloud Build

Use o Console do Google Cloud ou a CLI do Google Cloud para excluir seus dois gatilhos de compilação:

  • test-dags
  • add-dags-to-composer

Limpar recursos do Cloud Composer

Excluir o ambiente do Cloud Composer

Se você criou um ambiente do Cloud Composer para este tutorial e não quer continuar a usá-lo, exclua-o para evitar cobranças. Se você quiser manter o ambiente, remova os DAGs.

Remover seus DAGs do Cloud Composer

Exclua os DAGs dos ambientes de desenvolvimento e produção do Cloud Composer.

A seguir