Apache Airflow com .NET 10: dispare e monitore DAGs

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.

Exemplo Prático

O núcleo do exemplo é um cliente tipado registrado com IHttpClientFactory. O BaseAddress vem de configuração, e cada chamada cria seu próprio HttpRequestMessage. O header Bearer fica na mensagem, não em DefaultRequestHeaders, o que evita disputa quando requisições concorrentes usam tokens diferentes durante uma renovação.

O método abaixo cria um identificador, monta conf e dispara a execução:

public async Task<DagRunResponse> DispararAsync(
    string dagId,
    IReadOnlyDictionary<string, object?>? configuracao = null,
    CancellationToken cancellationToken = default)
{
    // O ID único permite correlacionar logs do .NET e do Airflow.
    var dagRunId = $"dotnet__{DateTimeOffset.UtcNow:yyyyMMddTHHmmssfff}__{Guid.NewGuid():N}";
    var payload = new DispararDagRequest(
        dagRunId,
        configuracao ?? new Dictionary<string, object?>());

    var caminho = $"api/v2/dags/{Uri.EscapeDataString(dagId)}/dagRuns";
    return await EnviarAutenticadoAsync<DagRunResponse>(
        () => new HttpRequestMessage(HttpMethod.Post, caminho)
        {
            Content = JsonContent.Create(payload)
        },
        cancellationToken);
}

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

O monitor usa a resposta do disparo e só consulta novamente enquanto o estado não for terminal:

public async Task<DagRunResponse> DispararEMonitorarAsync(
    string dagId,
    IReadOnlyDictionary<string, object?>? configuracao,
    CancellationToken cancellationToken)
{
    var execucao = await DispararAsync(dagId, configuracao, cancellationToken);
    using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
    timeout.CancelAfter(TimeSpan.FromMinutes(10));

    while (!EstadosFinais.Contains(execucao.Estado))
    {
        // Polling moderado evita sobrecarregar o webserver do Airflow.
        await Task.Delay(TimeSpan.FromSeconds(5), timeout.Token);
        execucao = await ObterExecucaoAsync(dagId, execucao.DagRunId, timeout.Token);
    }

    return execucao;
}

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

A aplicação registra o cliente e chama uma Minimal API com o seguinte corpo:

{
  // Parâmetros pequenos e não confidenciais entregues ao DAG.
  "configuracao": {
    "data_referencia": "2026-07-21",
    "correlation_id": "pedido-8f328ba9"
  },
  // O endpoint permanece aberto até success, failed ou canceled.
  "aguardarConclusao": true
}

# Dispara o DAG e aguarda um estado terminal.
curl -X POST "https://localhost:7063/api/airflow/dags/etl_vendas/runs" \
  -H "Content-Type: application/json" \
  -d '{"configuracao":{"data_referencia":"2026-07-21","correlation_id":"pedido-8f328ba9"},"aguardarConclusao":true}'

Uma resposta concluída preserva os nomes JSON do contrato do Airflow, facilitando a correlação com sua API e interface web:

{
  // Use este ID para correlacionar a execução nas duas plataformas.
  "dag_run_id": "dotnet__20260721T130000000__8f328ba9...",
  "dag_id": "etl_vendas",
  "state": "success",
  "start_date": "2026-07-21T13:00:01Z",
  "end_date": "2026-07-21T13:02:44Z"
}

Falhas, observabilidade e segurança em produção

Integração robusta exige separar falhas de autenticação, autorização, validação, disponibilidade e execução. Um 401 normalmente pede renovação do token; um 403 indica que a identidade não possui permissão; um 404 pode significar DAG ou DAG Run incorreto; um 409 pode indicar conflito de identidade; 429 e respostas 5xx sugerem pressão ou indisponibilidade temporária. O cliente deve registrar status, endpoint lógico e correlation ID, mas nunca token, senha ou conteúdo sensível de conf.

O exemplo inclui uma repetição única após 401. Para GET, retries exponenciais com jitter são seguros porque a operação não altera estado. Para POST, a política depende de idempotência. Uma biblioteca de resiliência pode automatizar timeout, circuit breaker e retries, mas a regra de negócio ainda precisa decidir se repetir o disparo é válido. Nenhuma biblioteca consegue inferir isso apenas pelo verbo HTTP.

Crie métricas para tempo de autenticação, latência de disparo, duração total, quantidade de DAG Runs por estado e timeouts. Registre dag_id, dag_run_id e correlation ID como propriedades estruturadas. Trace a chamada HTTP com OpenTelemetry e propague um identificador no conf para que tarefas do DAG também o incluam nos logs. Dessa forma, uma investigação percorre API, Airflow e sistemas de dados sem depender de busca textual frágil.

Na segurança, use HTTPS com validação normal de certificado, conta técnica de menor privilégio e rotação de credenciais. Não desabilite validação TLS em produção. Restrinja a API do Airflow por rede, firewall ou gateway; a existência de JWT não torna desejável expor o webserver publicamente. Se o serviço roda em Kubernetes, injete a senha por Secret ou integre um cofre. Em ambientes com auth manager corporativo, prefira identidades de workload e tokens de curta duração quando houver suporte.

Quando usar API, mensageria ou execução em containers

A API REST é adequada quando o serviço precisa iniciar um workflow específico, fornecer parâmetros e acompanhar uma execução identificável. Ela oferece feedback imediato, contrato HTTP claro e boa rastreabilidade. O custo é o acoplamento temporal: Airflow precisa estar acessível no disparo, e esperar a conclusão prende recursos do chamador. Para operações administrativas e baixo volume, esse compromisso costuma ser aceitável.

Mensageria é melhor quando o produtor deve continuar funcionando durante indisponibilidades do orquestrador ou quando o volume é alto. O .NET publica em Azure Service Bus, RabbitMQ ou Kafka, e um componente consumidor transforma mensagens em DAG Runs ou ativa pipelines orientados a eventos. A fila oferece buffering e entrega controlada, mas introduz semântica de duplicidade, dead-letter, ordenação e observabilidade adicional. Airflow não deve substituir um broker em um fluxo transacional de baixa latência.

Containers e Kubernetes resolvem outro problema: onde executar a carga. Um DAG pode iniciar um job que usa a mesma imagem de uma ferramenta .NET, ou chamar um serviço C# por HTTP. Isso não exige que o Airflow entenda a implementação interna. A imagem deve expor uma operação determinística, retornar exit code correto e gravar saída em armazenamento durável. Logs locais do pod não são uma integração de dados.

Em muitos sistemas, as três estratégias coexistem. A API .NET grava uma transação e uma outbox; um worker publica um evento; o Airflow inicia um DAG; uma tarefa Kubernetes executa um utilitário .NET. O desenho deve seguir requisitos de consistência, latência e recuperação. Escolher API, fila ou container apenas porque a equipe já conhece a tecnologia tende a esconder o verdadeiro contrato operacional.

Dicas e Boas Práticas

• Use uma identidade técnica de menor privilégio. Autorize somente os DAGs e operações exigidos pelo serviço. Separe a conta da integração das contas humanas para facilitar auditoria e rotação.

• Torne o disparo idempotente. Derive dag_run_id de uma chave de negócio ou persista a associação antes da chamada. Trate respostas ambíguas consultando a execução existente, em vez de criar outra silenciosamente.

• Mantenha conf pequeno e sem segredos. Envie IDs, datas e URIs para dados duráveis. Payloads volumosos pressionam o webserver e informações sensíveis podem aparecer nos metadados da execução.

• Diferencie timeout de chamada e timeout de workflow. Use um limite curto para cada requisição HTTP e outro para o ciclo completo de polling. Propague CancellationToken em autenticação, disparo, espera e consulta.

• Aplique retry conforme a semântica. Repetir GET após falha transitória é previsível; repetir POST pode duplicar processamento. Combine backoff exponencial, jitter e circuit breaker somente depois de definir idempotência.

• Registre identificadores, não credenciais. Inclua dag_id, dag_run_id, correlation ID, estado e duração em logs estruturados. Nunca registre JWT, senha ou configuração sensível.

• Teste contratos sem depender do Airflow real. Um HttpMessageHandler falso valida URLs, headers, JSON, renovação após 401 e transições de estado. Mantenha também um teste de integração contra uma versão suportada antes do deploy.

Resumo Objetivo

• Apache Airflow 3 — expõe uma API pública estável sob /api/v2; endpoints internos da interface web não devem ser usados como contrato de integração.
• .NET 10 — integra-se ao Airflow por HTTP e não executa DAGs em C#; o Airflow continua responsável por agendamento, dependências, workers e estado.
• Autenticação JWT — a documentação padrão usa POST /auth/token e envia o token em Authorization: Bearer; o cliente renova a credencial uma vez após 401 ou 403 causado por token inválido.
• DAG Run — é uma execução concreta de um DAG e deve ser rastreada por dag_id e dag_run_id; HTTP 200 no disparo não significa que o workflow terminou com sucesso.
• Monitoramento — consulta GET /api/v2/dags/{dag_id}/dagRuns/{dag_run_id} até success, failed ou canceled, com intervalo moderado, timeout total e cancelamento propagado.
• Idempotência — retries automáticos de POST podem duplicar workflows; uma chave de negócio no dag_run_id permite reconhecer uma execução já criada.
• Configuração do DAG — no Airflow 3.3, o payload inclui dag_run_id, conf e logical_date; o último pode ser null, enquanto conf deve carregar referências pequenas e não confidenciais.

Leia Também

• .NET Worker e Background Service: Alto Volume
• Configuração .NET 8+: IOptions, Secrets e Docker
• Log Sem Contexto é Ruído: Logging Estruturado no .NET 8
• Terminal vs Pipeline: diferenças ao executar scripts

Referências

• Airflow Public REST API Reference — contrato OpenAPI dos endpoints públicos estáveis sob /api/v2.
• Airflow Public API Authentication — autenticação JWT, endpoint de token e recomendações para payloads e CORS.
• Airflow DAG Runs — estados, data intervals, disparos externos e endpoint experimental de espera.
• Airflow Best Practices — práticas oficiais para escrever e operar DAGs confiáveis.
• Use the IHttpClientFactory — documentação do .NET para clientes HTTP tipados, handlers e gerenciamento de conexões.
• Options pattern in .NET — configuração tipada, validação e separação de credenciais por ambiente.
• Código completo: BlogSamples/Orchestration/Airflow — implementação .NET 10 e contratos usados neste artigo. 📬

---

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