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

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:

• Cloud Build
• Cloud Composer

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.

   Observação: se você não pretende manter os recursos criados neste procedimento, crie um projeto novo em vez de selecionar um que já existe. Depois de concluir essas etapas, é possível excluir o projeto. Para fazer isso, basta remover todos os recursos associados a ele.

   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

   composer/cicd_sample/test-dags.cloudbuild.yaml

   Ver no GitHub

   
   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/"]
   

Observação: o Cloud Build extrai a imagem em Python do Docker Hub, e você pode especificar a tag no campo name de cada etapa. Se você estiver usando python como entrypoint, verifique se a entrypoint corresponde à versão especificada na tag. Caso contrário, entrypoint assumirá latest por padrão.

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)

Cuidado: definir o controle de comentários não é obrigatório, o que significa que qualquer usuário pode enviar uma solicitação de envio que aciona a criação. No caso de repositórios maiores ou com colaboradores não confiáveis, recomendamos o uso do controle de comentários. Use a discrição com essa configuração.

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.

Observação: o Cloud Composer analisa todos os DAGs em um intervalo repetido. Isso acontece mesmo que os DAGs não mudem desde a última análise. Devido a esse comportamento, esse utilitário faz upload de todos os DAGs para o Cloud Composer, não apenas os que foram alterados.

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.

composer/cicd_sample/utils/add_dags_to_composer.py

Ver no GitHub

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

   composer/cicd_sample/add-dags-to-composer.cloudbuild.yaml (em inglês)

   Ver no GitHub

   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}"]
   

Observação: o Cloud Build extrai a imagem em Python do Docker Hub, e você pode especificar a tag no campo name de cada etapa. Se você estiver usando python como entrypoint, verifique se a entrypoint corresponde à versão especificada na tag. Caso contrário, entrypoint assumirá latest por padrão.

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.

Atenção: se você não remover o sufixo dags/ do nome do bucket do Cloud Storage na variável de substituição _DAGS_BUCKET, o job de sincronização do DAG falhará com um erro NotFound. Isso acontece porque, no Cloud Storage, dags/ é uma pasta, que é considerada parte do nome do objeto, não do nome do 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.

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

Cuidado: excluir um projeto tem os seguintes efeitos:

• Tudo no projeto é excluído. Se você tiver usado um projeto atual para seguir este tutorial, a exclusão dele incluirá quaisquer outras ações feitas no projeto.
• Os IDs do projeto personalizados são perdidos. Ao criar o projeto, você pode ter criado um código do projeto personalizado para ser usado no futuro. Para preservar os URLs que usam o ID do projeto, como um URL appspot.com, exclua recursos específicos do projeto, em vez de excluir o projeto inteiro.

Se você planeja explorar vários tutoriais e guias de início rápido, a reutilização de projetos pode evitar que você exceda os limites da cota do 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

• Como escrever DAGs
• Como acionar DAGs
• Como testar DAGs