Vailati
Voltar

Trace distribuído que atravessa o outbox e a fila em memória

O trace de uma requisição termina no commit. O evento sai minutos depois, por outro processo, e aparece no APM como se não tivesse pai. Três peças pequenas resolvem isso.

Contexto

Eu precisava que um trace iniciado numa requisição HTTP continuasse no evento publicado no Kafka minutos depois, por um job, e também no handler que roda numa fila em memória da própria aplicação. A decisão foi persistir o traceparent na linha do outbox no momento do commit e reinjetá-lo no cabeçalho da mensagem na hora de publicar; para a fila em memória, um “sidecar” guarda o contexto ao lado do objeto do evento. O custo foi uma coluna de texto nas tabelas de outbox, um interceptor do EF Core e meia dúzia de linhas em cada job.

O cenário é um ledger de pontos com padrão outbox: cada operação grava a linha de negócio e uma linha em outbox na mesma transação de banco; um job a cada minuto lê o outbox, publica no Kafka e apaga. Há ainda uma fila em memória, um Channel consumido por um BackgroundService, para eventos de domínio que não precisam sair do processo. A instrumentação é o agente Elastic APM para .NET, mas o problema e a solução valem para qualquer tracer que siga o W3C Trace Context.

Problema

Tracing distribuído funciona por propagação de contexto: cada chamada carrega um identificador de trace, e quem recebe cria um span filho. Em HTTP o agente faz isso sozinho, injetando e lendo o cabeçalho traceparent. O outbox quebra a corrente em dois pontos.

O primeiro é o tempo. A requisição que criou o lançamento terminou, o agente fechou a transação de APM e o contexto ambiente, que vive num AsyncLocal, sumiu. Quando o job acorda, um minuto depois, não existe mais nada no processo que lembre de qual requisição aquela linha veio. O ProduceAsync para o Kafka acontece dentro de uma transação de APM do job, com um trace id novo. No Kibana, a requisição e a publicação são dois traces sem relação.

O segundo é a fila em memória. O evento é um objeto .NET colocado num Channel; o consumidor o retira em outra task, em outro contexto assíncrono. O AsyncLocal da requisição não acompanha o objeto. O handler roda sem trace, ou pior, dentro do trace de outra coisa que estava ativa no consumidor.

A consequência prática aparece no incidente. Um consumidor reclama de um evento com valor errado; o evento tem um id de transação; alguém abre o Kibana, acha o trace do consumidor, e ele começa na publicação. Para chegar na requisição de origem, é preciso sair do APM e cruzar ids no banco. E o mapa de serviços do APM mostra a API e o consumidor como duas ilhas, porque nunca há um trace ligando os dois.

O que medi

Este artigo tem mais observação do que medição, e vale dizer isso.

Antes da mudança, o trace de um POST terminava no span do commit. No worker, cada execução do job gerava uma transação de APM própria, com um span por mensagem, e nenhum deles tinha o id do trace da requisição. Depois da mudança, abrindo o trace da requisição no Kibana, aparece abaixo dela a transação Kafka.Publish <tópico> do serviço worker, com o intervalo de até um minuto visível entre o fim da requisição e a publicação. Se o consumidor também estiver instrumentado, a transação dele entra no mesmo trace. O mapa de serviços passou a desenhar API → worker → consumidor.

O custo de armazenamento é uma coluna text anulável por linha de outbox. O valor serializado é o traceparent do W3C, 55 caracteres. Como a linha é apagada logo depois da publicação, o custo em regime é desprezível; ele só cresce se o outbox acumular, e nesse caso o problema é outro.

Uma ressalva que descobri depois: o agente APM descarta eventos sob carga quando a fila interna dele enche, e isso faz traces parecerem incompletos mesmo com a propagação correta. Quando um trace “some”, eu verifico primeiro se o cabeçalho traceparent está na mensagem no Kafka. Se está, a propagação funcionou e o buraco é do agente.

Solução

São três peças, cada uma num ponto diferente da cadeia.

1. Carimbar o trace na linha do outbox, no SaveChanges

Um SaveChangesInterceptor do EF Core percorre as entidades de outbox em estado Added e preenche a coluna com o contexto serializado da transação de APM corrente. O interceptor é registrado como singleton e injetado no DbContext; ninguém precisa lembrar de preencher o campo.

public sealed class TraceContextSaveChangesInterceptor : SaveChangesInterceptor
{
    private static void Stamp(DbContext? ctx)
    {
        if (ctx is null) return;

        foreach (var entry in ctx.ChangeTracker.Entries<OutboxMessage>())
        {
            if (entry.State == EntityState.Added && string.IsNullOrEmpty(entry.Entity.TraceParent))
                entry.Entity.TraceParent = Agent.Tracer.CurrentTransaction?
                    .OutgoingDistributedTracingData?.SerializeToString();
        }
    }

    public override InterceptionResult<int> SavingChanges(
        DbContextEventData e, InterceptionResult<int> result)
    {
        Stamp(e.Context);
        return base.SavingChanges(e, result);
    }

    public override ValueTask<InterceptionResult<int>> SavingChangesAsync(
        DbContextEventData e, InterceptionResult<int> result, CancellationToken ct = default)
    {
        Stamp(e.Context);
        return base.SavingChangesAsync(e, result, ct);
    }
}

Dois detalhes. Só entidades Added e só se o campo estiver vazio, para não sobrescrever um contexto que alguém definiu de propósito. E o OutgoingDistributedTracingData é o que o agente injetaria num cabeçalho HTTP de saída: é o span corrente como pai, não o id da requisição inteira. É exatamente o que se quer.

2. Reinjetar o cabeçalho na publicação

O job lê o lote do outbox, monta uma mensagem por linha e, se a linha tem traceparent, coloca no cabeçalho do Kafka. Depois, para que o próprio ato de publicar apareça no trace, abre uma transação de APM filha daquele contexto:

var message = new Message<string, byte[]>
{
    Key = row.AccountId,
    Value = Encoding.UTF8.GetBytes(row.Payload),
    Headers = [new Header("correlationId", Encoding.UTF8.GetBytes(row.EntryId.ToString()))]
};

if (!string.IsNullOrWhiteSpace(row.TraceParent))
    message.Headers.Add("traceparent", Encoding.UTF8.GetBytes(row.TraceParent));

var parent = message.Headers.TryGetLastBytes("traceparent", out var bytes)
    ? DistributedTracingData.TryDeserializeFromString(Encoding.UTF8.GetString(bytes))
    : null;

var publish = Agent.Tracer.StartTransaction(
    $"Kafka.Publish {topic}", ApiConstants.TypeMessaging, parent);
try
{
    if (!message.Headers.Any(h => h.Key == "traceparent"))
    {
        // linha antiga, sem contexto: o consumidor ainda recebe um traceparent válido
        var outgoing = publish.OutgoingDistributedTracingData?.SerializeToString();
        if (outgoing is not null)
            message.Headers.Add("traceparent", Encoding.UTF8.GetBytes(outgoing));
    }
    await producer.ProduceAsync(topic, message, ct);
}
catch (Exception ex)
{
    publish.CaptureException(ex);
    publish.Outcome = Outcome.Failure;
    throw;
}
finally
{
    publish.End();
}

É uma transação, não um span, porque o laço do job não tem transação ambiente por mensagem. O fallback do if cobre linhas gravadas antes da mudança e qualquer caminho que grave outbox fora de uma requisição: o consumidor sempre recebe um traceparent, mesmo que o trace comece no worker.

3. Sidecar para a fila em memória

Na fila em memória não há cabeçalho onde pendurar nada, e eu não queria um campo de tracing dentro do evento de domínio, porque o domínio não conhece APM. A saída é um mapa fraco do evento para o contexto:

public sealed class EventTraceSidecar<TEvent> where TEvent : class
{
    private readonly ConditionalWeakTable<TEvent, DistributedTracingData?> _map = new();

    public void Attach(TEvent evt, DistributedTracingData? ctx) { _map.Remove(evt); _map.Add(evt, ctx); }
    public DistributedTracingData? Get(TEvent evt) => _map.TryGetValue(evt, out var ctx) ? ctx : null;
    public void Detach(TEvent evt) => _map.Remove(evt);
}

public static TEvent WithTrace<TEvent>(this TEvent evt, EventTraceSidecar<TEvent> sidecar)
    where TEvent : class
{
    sidecar.Attach(evt, Agent.Tracer.CurrentTransaction?.OutgoingDistributedTracingData);
    return evt;
}

Quem enfileira escreve queue.Enqueue(new EntryProcessed(...).WithTrace(sidecar)). O consumidor faz o inverso:

var evt = await queue.DequeueAsync(ct);
var parent = sidecar.Get(evt);

await Agent.Tracer.CaptureTransaction(
    $"Dispatch {typeof(TEvent).Name}", ApiConstants.TypeMessaging,
    async _ => await dispatcher.HandleAsync(evt, ct),
    distributedTracingData: parent);

sidecar.Detach(evt);

ConditionalWeakTable é a escolha certa aqui por dois motivos: a chave é a identidade do objeto, então não precisa de id no evento; e a referência é fraca, então um evento descartado sem ser consumido não deixa entrada presa no mapa. O sidecar é um singleton por tipo de evento, registrado como genérico aberto no DI.

sequenceDiagram
    participant C as Cliente
    participant A as API
    participant DB as PostgreSQL
    participant J as Job (worker)
    participant K as Kafka
    participant X as Consumidor
    C->>A: POST (trace T)
    A->>DB: INSERT entrada + outbox(traceparent=T)
    A-->>C: 201
    Note over A,J: até 1 min depois
    J->>DB: SELECT outbox
    J->>K: publish, header traceparent=T (transação filha de T)
    J->>DB: DELETE outbox
    K->>X: consume (continua T)

O que não mudou

Nada no domínio. A entidade de outbox ganhou uma propriedade string?, e só. Os jobs não sabem de onde o contexto veio. O interceptor é a única peça que conhece o agente do lado da escrita.

O que faria diferente

Eu mediria a propagação pelo cabeçalho, não pela interface do APM. Durante meses eu avaliei se “o trace estava chegando” olhando o Kibana. Quando descobri que o agente descartava eventos sob carga, percebi que parte dos traces incompletos nunca foi problema de propagação. Um teste de integração que consome a mensagem do Kafka e afirma a presença e o formato do traceparent teria separado as duas coisas desde o início.

Eu usaria System.Diagnostics.Activity como formato interno. O sidecar e a coluna guardam o tipo do agente Elastic. Como o valor serializado é o traceparent do W3C, trocar de fornecedor não quebra os dados, mas quebra o código do interceptor, do sidecar e dos jobs. Com Activity e ActivityContext, essas três peças ficariam neutras e o agente, seja qual for, leria o contexto sozinho.

Eu olharia span links em vez de pai e filho. Modelar a publicação como filha da requisição estica o trace pelo intervalo do job e sugere que a requisição “demorou” um minuto. O W3C e o OpenTelemetry têm a noção de link entre spans para exatamente esse caso assíncrono. Na época, o agente que eu usava não expunha isso de forma simples; hoje eu verificaria de novo.

Referências

Compartilhe este artigo:
Compartilhar no LinkedinCompartilhar no WhatsappCompartilhar no XCompartilhar por e-mail
Artigo anterior
Particionar um ledger em dois níveis: grupo de conta e ano
Próximo artigo
CDC ou outbox: por que escolhi o outbox para publicar eventos