Apache Airflow com .NET 10: dispare e monitore DAGs · lzocateli

Introdução#

Integrar Apache Airflow com .NET 10 não significa portar o orquestrador, reescrever DAGs em C# ou executar o runtime Python dentro da aplicação. A solução correta é manter o Airflow responsável por criar, agendar e monitorar workflows e fazer o serviço .NET consumir sua API REST pública. O serviço autentica, dispara um DAG Run com parâmetros, guarda o identificador retornado e consulta o estado até receber success, failed ou canceled. Essa separação preserva o papel de cada tecnologia e cria um contrato claro entre a aplicação transacional e a plataforma de dados.

Neste guia, eu vou implementar esse fluxo de ponta a ponta usando .NET 10, HttpClient, autenticação JWT e a API /api/v2 do Apache Airflow 3.3. O exemplo não se limita a um POST: ele gera um dag_run_id rastreável, serializa conf corretamente, reutiliza o token até perto da expiração, renova a credencial após uma resposta 401, aplica timeout ao monitoramento e propaga cancelamento. Também vou expor a integração por uma Minimal API, para que outro sistema possa iniciar o processo sem conhecer os detalhes do Airflow.

O nome atual da plataforma da Microsoft é .NET 10, e não “.NET Core 10”. A marca “.NET Core” foi usada até a versão 3.1; desde o .NET 5, o produto unificado passou a se chamar apenas .NET. Essa diferença não altera o código, mas evita confusão ao procurar documentação, imagens de container e pacotes compatíveis.

O cenário prático será um serviço de pedidos que solicita a execução do DAG etl_vendas. A configuração enviada contém a data de referência e um identificador de correlação. O Airflow continua executando tarefas Python, SQL, containers ou jobs distribuídos; o C# apenas controla o ciclo de vida da execução pela fronteira HTTP.

ℹ️ Informação: no Airflow 3, os endpoints públicos estáveis ficam sob /api/v2. Rotas internas de UI não são um contrato de integração e podem mudar conforme o frontend.

Pré-requisitos#

Para acompanhar o exemplo, você precisa do .NET 10 SDK, do Docker Compose 2.14 ou superior e de pelo menos 4 GB de memória disponíveis para o Docker. O repositório inclui uma instância do Apache Airflow 3.3.0, PostgreSQL 16 e o DAG etl_vendas, portanto não é necessário instalar Airflow ou Python diretamente na máquina. A autenticação usa JWT obtido por POST /auth/token.

O exemplo foi escrito para a API pública v2. Se sua empresa ainda usa Airflow 2, a API estável normalmente aparece sob /api/v1, com contratos e mecanismos de autenticação que podem variar. Não troque apenas o número na URL: valide o OpenAPI da versão instalada, porque nomes de campos, estados e recursos disponíveis não são necessariamente idênticos.

Você também deve confirmar estes pontos antes de codificar:

O serviço .NET consegue resolver o DNS e abrir uma conexão TLS com o endpoint do Airflow.
O usuário técnico possui somente as permissões necessárias para disparar e ler o DAG escolhido.
O proxy reverso aceita os verbos POST e GET, o header Authorization e o tamanho máximo previsto para conf.
O DAG é idempotente ou recebe uma chave de negócio que impeça processamento duplicado.
O timeout do serviço é compatível com a duração esperada do workflow.
Em produção, armazene a senha em um cofre, secret de container ou variável de ambiente. A chave hierárquica do .NET será Airflow__Senha; os dois sublinhados representam o separador de configuração. Nunca publique a credencial em appsettings.json, imagem Docker, log ou telemetria.

Subindo o Airflow 3 com dados persistentes no Docker#

O exemplo usa uma stack menor que o quick-start oficial porque o objetivo é testar a integração REST, não demonstrar o CeleryExecutor. O LocalExecutor executa tarefas no mesmo ambiente do scheduler e elimina Redis e workers Celery. Permanecem os componentes necessários do Airflow 3: API server, scheduler, DAG processor, triggerer, inicializador e PostgreSQL.

Todos os dados relevantes usam bind mounts, isto é, diretórios explícitos do repositório montados nos containers. DAGs ficam em dags/, logs em logs/, configuração em config/, plugins em plugins/ e o cluster PostgreSQL em data/postgres/. Os diretórios gerados estão no .gitignore; apenas o DAG, o Compose e arquivos .gitkeep são versionados. Executar docker compose down pode destruir containers e rede sem apagar banco, usuários, histórico ou logs.

No Airflow 3, o executor também precisa conversar com a Execution API. Por isso, todos os serviços recebem AIRFLOW__CORE__EXECUTION_API_SERVER_URL=http://airflow-api-server:8080/execution/. Usar localhost nessa variável faria o scheduler procurar a API dentro do próprio container e as tasks falhariam com Connection refused, mesmo com o healthcheck externo saudável.

O recorte abaixo mostra os mounts. O arquivo completo também define healthchecks e dependências de inicialização:

services:
  postgres:
    image: postgres:16-alpine
    volumes:
      # Metadados e histórico sobrevivem à recriação do container.
      - ./data/postgres:/var/lib/postgresql/data

  airflow-api-server:
    image: apache/airflow:3.3.0
    command: api-server
    ports:
      # A porta 8081 evita conflito com outros serviços locais.
      - "8081:8080"
    volumes:
      - ./dags:/opt/airflow/dags
      - ./logs:/opt/airflow/logs
      - ./config:/opt/airflow/config
      - ./plugins:/opt/airflow/plugins

📂 Código Fonte: A stack, o DAG e as instruções completas estão disponíveis no repositório de exemplos:
BlogSamples/Orchestration/Airflow/

Inicialize o banco e crie o usuário na primeira execução. O comando up -d --wait só retorna quando os healthchecks terminarem:

# Execute a partir da pasta Orchestration/Airflow do exemplo.
Copy-Item .env.example .env
docker compose up airflow-init
docker compose up -d --wait

A interface web e a API pública estarão em http://localhost:8081. As credenciais padrão do exemplo são usuário dotnet-service e senha airflow; altere-as no .env em qualquer ambiente compartilhado. Para configurar o cliente .NET sem gravar senha em arquivo, use uma variável de ambiente:

# Os dois sublinhados representam Airflow:Senha na configuração .NET.
$env:Airflow__Senha = "airflow"
dotnet run --project src/BlogSamples

⚠️ Atenção: esta stack é destinada a desenvolvimento e demonstração. Credenciais padrão, LocalExecutor e API HTTP sem TLS não atendem requisitos de produção.

Como o Apache Airflow divide as responsabilidades#

O Apache Airflow representa workflows por DAGs, grafos acíclicos direcionados que descrevem tarefas e dependências. Um DAG é a definição; um DAG Run é uma execução concreta dessa definição em determinado contexto. Vários DAG Runs do mesmo DAG podem coexistir, e cada um possui seu próprio dag_run_id, estado, datas e configuração. Essa distinção é essencial para o cliente .NET não confundir o pipeline com uma de suas execuções.

O scheduler avalia agendas e dependências. O executor encaminha tarefas para o ambiente de execução, que pode usar processos locais, Celery ou Kubernetes, conforme a implantação. Workers executam as tarefas. A interface web apresenta grids, grafos, logs e histórico. A API pública oferece uma fronteira programática sobre esses recursos. Portanto, o serviço .NET não precisa conhecer filas internas, banco de metadados nem detalhes dos workers.

Essa arquitetura funciona bem em pipelines de ETL, treinamento de modelos, geração de relatórios e automações com muitas etapas. Um checkout em .NET pode publicar um evento e retornar rapidamente, enquanto o Airflow coordena um processamento posterior. Em outro cenário, uma API administrativa pode disparar o DAG e aguardar o resultado porque o operador precisa de confirmação síncrona. A escolha entre esperar e desacoplar depende da duração e do impacto no usuário, não da capacidade técnica da API.

O estado do DAG Run é derivado das tarefas folha. Segundo a documentação oficial, os estados finais normais de um DAG Run são success e failed; o endpoint experimental de espera também considera canceled. Estados como queued e running indicam que a execução ainda pode mudar. Um cliente não deve interpretar HTTP 200 no disparo como conclusão bem-sucedida: essa resposta confirma a criação do DAG Run, não o resultado do workflow.

API REST v2, JWT e contrato de autenticação#

O fluxo de autenticação documentado para o Airflow 3 começa em POST /auth/token. O corpo contém username e password, e a resposta contém access_token. As chamadas seguintes enviam Authorization: Bearer <token>. Como o endpoint de token é fornecido pelo auth manager, uma implantação corporativa pode usar outro mecanismo ou acrescentar requisitos. Consulte a configuração do ambiente antes de assumir que credenciais locais estarão habilitadas.

Depois da autenticação, o disparo usa POST /api/v2/dags/{dag_id}/dagRuns. No Airflow 3.3, o corpo exige dag_run_id, conf e logical_date; envie logical_date: null quando quiser que o servidor determine a data lógica. A configuração é um objeto JSON entregue ao contexto do DAG. A consulta usa GET /api/v2/dags/{dag_id}/dagRuns/{dag_run_id}. O identificador precisa ser escapado na URL, mesmo quando os nomes atuais parecem conter apenas letras e sublinhados.

No C#, eu mapeio os nomes em snake case explicitamente. Isso é importante porque JsonSerializerDefaults.Web produz camel case, não dag_run_id. O trecho abaixo mostra os contratos principais:

public sealed record DispararDagRequest(
    [property: JsonPropertyName("dag_run_id")] string DagRunId,
    [property: JsonPropertyName("conf")]
  IReadOnlyDictionary<string, object?> Configuracao,
  // O campo é obrigatório na API 3.3, mas pode ser nulo.
  [property: JsonPropertyName("logical_date")]
  DateTimeOffset? DataLogica = null);

public sealed record DagRunResponse(
    [property: JsonPropertyName("dag_run_id")] string DagRunId,
    [property: JsonPropertyName("dag_id")] string DagId,
    [property: JsonPropertyName("state")] string Estado,
    [property: JsonPropertyName("start_date")] DateTimeOffset? Inicio,
    [property: JsonPropertyName("end_date")] DateTimeOffset? Fim);

📂 Código Fonte: O exemplo completo está disponível no repositório de exemplos do blog:
BlogSamples/Orchestration/Airflow/

Um cliente robusto não solicita um novo token em cada polling. Ele lê o claim exp do JWT e reutiliza a credencial enquanto houver margem de validade. Um SemaphoreSlim impede que várias requisições concorrentes renovem o token ao mesmo tempo. Se o servidor responder 401 ou 403 após uma rotação da chave JWT, o cliente invalida o cache, autentica novamente e repete a chamada uma única vez. Uma permissão realmente insuficiente continuará retornando 403 na segunda tentativa. Repetições ilimitadas esconderiam credenciais incorretas e poderiam pressionar o endpoint de autenticação.

Identidade da execução, parâmetros e idempotência#

O dag_run_id é a ponte de rastreabilidade entre as duas plataformas. No exemplo, ele segue o formato dotnet__<timestamp UTC>__<guid>. O timestamp ajuda durante uma investigação manual, enquanto o GUID evita colisões entre instâncias do serviço. Eu registro esse identificador junto ao dag_id, ao estado e ao correlation ID da requisição de origem. Assim, uma busca no log do .NET leva ao DAG Run correspondente na interface do Airflow.

Para processos de negócio, aleatoriedade não basta. Imagine que o cliente envie o mesmo pedido duas vezes após um timeout de rede. O primeiro POST pode ter criado a execução, mesmo que a resposta não tenha chegado. Repetir automaticamente o disparo com outro GUID produziria dois workflows. Uma estratégia mais forte deriva o identificador de uma chave idempotente, como pedido__8f32..., e trata 409 Conflict consultando a execução já existente. Outra alternativa persiste uma outbox antes da integração e deixa um worker controlar as tentativas.

O objeto conf deve conter referências pequenas e imutáveis, não arquivos nem grandes lotes de registros. Envie pedido_id, data_referencia, URI de um objeto ou correlation ID. O DAG usa essas referências para buscar os dados na origem autorizada. A própria documentação alerta que payloads muito grandes podem afetar a estabilidade do webserver e recomenda limites no proxy.

⚠️ Atenção: não aplique uma política genérica de retry ao POST de disparo sem uma chave idempotente. Falha de transporte não prova que o Airflow deixou de criar o DAG Run.

Também evite colocar segredos em conf. A configuração aparece em metadados do DAG Run e pode ficar visível para operadores autorizados. Passe apenas identificadores; recupere credenciais em um secrets backend do Airflow, como HashiCorp Vault, AWS Secrets Manager ou outro provider compatível com sua infraestrutura.

Arquitetura do disparo e monitoramento#

O fluxo síncrono possui três fases. Primeiro, o serviço .NET obtém ou reutiliza um JWT. Depois, cria o DAG Run e recebe o estado inicial, geralmente queued. Por fim, consulta a mesma execução em intervalos controlados até um estado terminal. O cancelamento da requisição e um timeout próprio precisam interromper o laço para não manter conexões e tarefas indefinidamente.

Eu prefiro polling explícito com GET no exemplo principal porque os endpoints usados são estáveis. O Airflow 3 oferece também GET /api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/wait, que transmite atualizações em NDJSON, mas a documentação o classifica como experimental. Ele pode ser útil em automações controladas, porém não é a melhor fundação para um contrato de longa duração sem encapsulamento e testes de compatibilidade.

O intervalo de polling deve equilibrar latência e carga. Consultar a cada 100 milissegundos não acelera o DAG e multiplica requisições ao webserver. Para workflows de minutos, cinco ou dez segundos costumam ser um ponto inicial razoável. Adicione jitter quando muitas instâncias puderem iniciar juntas, evitando que todas consultem no mesmo instante. Para milhares de workflows simultâneos, prefira um worker central, callbacks ou eventos em vez de manter uma requisição HTTP por execução.

O timeout do HttpClient limita uma chamada individual; ele não limita todo o monitoramento. Por isso, o exemplo usa um CancellationTokenSource vinculado ao token recebido e chama CancelAfter com o tempo máximo do workflow. Essa distinção evita um erro comum: configurar 30 segundos no cliente HTTP e acreditar que o laço inteiro terminará nesse prazo.

📖 Artigo completo com exemplos de código: Apache Airflow com .NET 10: dispare e monitore DAGs