Vailati
Go back

Distributed tracing across the outbox and the in-memory queue

A request’s trace ends at the commit. The event leaves minutes later, from another process, and shows up in the APM as if it had no parent. Three small pieces fix that.

Context

I needed a trace started in an HTTP request to continue into the event published to Kafka minutes later by a job, and also into the handler that runs on the application’s own in-memory queue. The decision was to persist the traceparent on the outbox row at commit time and re-inject it into the message header at publish time; for the in-memory queue, a “sidecar” keeps the context next to the event object. The cost was one text column on the outbox tables, one EF Core interceptor and half a dozen lines in each job.

The setting is a points ledger with the outbox pattern: every operation writes the business row and a row in outbox in the same database transaction; a job runs every minute, reads the outbox, publishes to Kafka and deletes. There is also an in-memory queue, a Channel consumed by a BackgroundService, for domain events that do not need to leave the process. The instrumentation is the Elastic APM agent for .NET, but the problem and the solution apply to any tracer that follows W3C Trace Context.

Problem

Distributed tracing works by context propagation: each call carries a trace identifier, and whoever receives it creates a child span. Over HTTP the agent does this on its own, injecting and reading the traceparent header. The outbox breaks the chain at two points.

The first is time. The request that created the entry has finished, the agent has closed the APM transaction, and the ambient context, which lives in an AsyncLocal, is gone. When the job wakes up a minute later, nothing in the process remembers which request that row came from. The ProduceAsync to Kafka happens inside the job’s own APM transaction, with a brand-new trace id. In Kibana, the request and the publish are two unrelated traces.

The second is the in-memory queue. The event is a .NET object placed in a Channel; the consumer takes it out on another task, in another async context. The request’s AsyncLocal does not travel with the object. The handler runs without a trace, or worse, inside the trace of whatever else was active on the consumer.

The practical consequence shows up during an incident. A consumer complains about an event with a wrong value; the event carries a transaction id; someone opens Kibana, finds the consumer’s trace, and it starts at the publish. To reach the originating request, you have to leave the APM and cross-reference ids in the database. And the APM service map shows the API and the consumer as two islands, because there is never a trace connecting them.

What I measured

This article has more observation than measurement, and that is worth saying.

Before the change, the trace of a POST ended at the commit span. On the worker, each run of the job produced its own APM transaction, with one span per message, and none of them carried the request’s trace id. After the change, opening the request’s trace in Kibana shows, below it, the worker service’s Kafka.Publish <topic> transaction, with the gap of up to a minute between the end of the request and the publish visible. If the consumer is instrumented too, its transaction joins the same trace. The service map started drawing API → worker → consumer.

The storage cost is one nullable text column per outbox row. The serialized value is the W3C traceparent, 55 characters. Since the row is deleted right after publishing, the steady-state cost is negligible; it only grows if the outbox backs up, and in that case the problem is elsewhere.

One caveat I discovered later: the APM agent drops events under load when its internal queue fills up, and that makes traces look incomplete even with correct propagation. When a trace “disappears”, I first check whether the traceparent header is on the message in Kafka. If it is, propagation worked and the hole belongs to the agent.

Solution

Three pieces, each at a different point of the chain.

1. Stamp the trace on the outbox row, in SaveChanges

An EF Core SaveChangesInterceptor walks the outbox entities in Added state and fills the column with the serialized context of the current APM transaction. The interceptor is registered as a singleton and injected into the DbContext; nobody has to remember to fill the field.

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);
    }
}

Two details. Only Added entities, and only if the field is empty, so as not to overwrite a context someone set on purpose. And OutgoingDistributedTracingData is what the agent would inject into an outgoing HTTP header: it is the current span as parent, not the id of the whole request. Which is exactly what you want.

2. Re-inject the header at publish time

The job reads the batch from the outbox, builds one message per row and, if the row has a traceparent, puts it in the Kafka header. Then, so that the act of publishing itself shows up in the trace, it opens an APM transaction that is a child of that context:

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"))
    {
        // old row, no context: the consumer still receives a valid traceparent
        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();
}

It is a transaction, not a span, because the job loop has no ambient transaction per message. The fallback in the if covers rows written before the change and any code path that writes to the outbox outside a request: the consumer always receives a traceparent, even if the trace starts at the worker.

3. A sidecar for the in-memory queue

On the in-memory queue there is no header to hang anything on, and I did not want a tracing field inside the domain event, because the domain knows nothing about APM. The way out is a weak map from the event to the context:

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;
}

Whoever enqueues writes queue.Enqueue(new EntryProcessed(...).WithTrace(sidecar)). The consumer does the reverse:

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 is the right choice here for two reasons: the key is the object’s identity, so the event needs no id; and the reference is weak, so an event dropped without being consumed leaves no entry stuck in the map. The sidecar is a singleton per event type, registered as an open generic in DI.

sequenceDiagram
    participant C as Client
    participant A as API
    participant DB as PostgreSQL
    participant J as Job (worker)
    participant K as Kafka
    participant X as Consumer
    C->>A: POST (trace T)
    A->>DB: INSERT entry + outbox(traceparent=T)
    A-->>C: 201
    Note over A,J: up to 1 min later
    J->>DB: SELECT outbox
    J->>K: publish, header traceparent=T (child transaction of T)
    J->>DB: DELETE outbox
    K->>X: consume (continues T)

What did not change

Nothing in the domain. The outbox entity gained one string? property, and that is all. The jobs do not know where the context came from. The interceptor is the only piece that knows about the agent on the write side.

What I would do differently

I would measure propagation by the header, not by the APM UI. For months I judged whether “the trace was arriving” by looking at Kibana. When I found out the agent was dropping events under load, I realized that part of the incomplete traces had never been a propagation problem. An integration test that consumes the message from Kafka and asserts the presence and format of the traceparent would have separated the two from the start.

I would use System.Diagnostics.Activity as the internal format. The sidecar and the column hold the Elastic agent’s type. Since the serialized value is the W3C traceparent, switching vendors does not break the data, but it does break the code of the interceptor, the sidecar and the jobs. With Activity and ActivityContext, those three pieces would be vendor-neutral and whichever agent is in use would read the context on its own.

I would look at span links instead of parent and child. Modelling the publish as a child of the request stretches the trace across the job’s interval and suggests the request “took” a minute. W3C and OpenTelemetry have the notion of a link between spans for exactly this asynchronous case. At the time, the agent I was using did not expose that in a simple way; today I would check again.

References

Share this post:
Share this post on LinkedinShare this post on WhatsappShare this post on XShare this post via email
Previous Post
Partitioning a ledger at two levels: account group and year
Next Post
CDC or outbox: why I chose the outbox to publish events