Observability in Messaging: OpenTelemetry Trace Context, Queue Depth and Consumer Lag

Version note: The code in this post was built and run with .NET 10, OpenTelemetry .NET 1.20.1, RabbitMQ 4.4, RabbitMQ.Client 7.2.2 and Confluent.Kafka 2.16 (Kafka 4.2).

Guess First

An order takes this path: POST /orders → RabbitMQ → Billing (issues the invoice) → the bookstore.invoices topic in Kafka → Notification (sends the e-mail). You add OpenTelemetry to the services, but Confluent.Kafka has no stable instrumentation package, and you do not add a separate package for RabbitMQ.Client 7 either. How many traces does a single order produce: one, two or three? Does the trace id of the HTTP (Hypertext Transfer Protocol) request that hit the API (Application Programming Interface) show up in Notification's log line?

The hardest part of messaging is that a request no longer lives in a single call stack but is spread over time and services. The last post of the Messaging and Event-Driven Architecture series talked about that cost; in this post we pay it in code: carrying the W3C (World Wide Web Consortium) traceparent in message headers, producer and consumer spans, a correlation id in the log scope, a processing duration metric, queue depth and consumer lag. Infrastructure such as the Collector and Grafana belongs to the upcoming Observability series; here we stay on the application side.

Setting Up OpenTelemetry

A single demo application plays all three roles: the Order API, the Billing worker and the Notification worker. The setup:

builder.Services.AddOpenTelemetry()
    .ConfigureResource(r => r.AddService("bookstore-lab12"))
    .WithTracing(tracing =>
    {
        tracing
            .AddAspNetCoreInstrumentation()
            .AddSource("RabbitMQ.Client.*")
            .AddSource(MessagingTelemetry.SourceName)
            .AddProcessor(new SpanLineProcessor());
        if (builder.Configuration["OTEL_EXPORTER_OTLP_ENDPOINT"] is not null)
            tracing.AddOtlpExporter();
    })
    .WithMetrics(metrics =>
    {
        metrics.AddMeter(MessagingTelemetry.MeterName);
        if (builder.Configuration["OTEL_EXPORTER_OTLP_ENDPOINT"] is not null)
            metrics.AddOtlpExporter();
    });

In production, spans go to a Collector through the OTLP (OpenTelemetry Protocol) exporter. The demo has no Collector; instead, a small processor writes one line for every finished span. To keep the output readable it shortens the trace and span ids:

public sealed class SpanLineProcessor : BaseProcessor<Activity>
{
    public override void OnEnd(Activity a)
    {
        string parent = a.ParentSpanId == default ? "-     " : a.ParentSpanId.ToString()[..6];
        Console.WriteLine($"span trace={a.TraceId.ToString()[..8]} id={a.SpanId.ToString()[..6]} parent={parent} " +
                          $"{a.Kind,-8} {a.DisplayName}");
    }
}

RabbitMQ.Client 7: Built-In Instrumentation

RabbitMQ.Client 7 uses .NET's ActivitySource API directly, with two sources named RabbitMQ.Client.Publisher and RabbitMQ.Client.Subscriber. AddSource("RabbitMQ.Client.*") is enough to listen to them; no extra package. The publishing code has not a single line about tracing:

public sealed class OrderPublisher(IConnection connection) : IAsyncDisposable
{
    private readonly SemaphoreSlim _gate = new(1, 1);
    private IChannel? _channel;

    public async Task PublishAsync(OrderPlaced order, CancellationToken ct = default)
    {
        var props = new BasicProperties
        {
            MessageId = Guid.NewGuid().ToString(),
            CorrelationId = order.OrderId.ToString(),
            Type = "bookstore.order-placed.v1",
            ContentType = "application/json",
            Persistent = true
        };
        byte[] body = JsonSerializer.SerializeToUtf8Bytes(order, JsonSerializerOptions.Web);
        // IChannel is not thread-safe: guard the single channel with a semaphore
        await _gate.WaitAsync(ct);
        try
        {
            _channel ??= await connection.CreateChannelAsync(new CreateChannelOptions(
                publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true), ct);
            // RabbitMQ.Client 7 writes the traceparent header itself
            await _channel.BasicPublishAsync("bookstore.ordering", "order.placed", mandatory: true, props, body, ct);
        }
        finally
        {
            _gate.Release();
        }
    }

    public async ValueTask DisposeAsync()
    {
        if (_channel is not null)
            await _channel.DisposeAsync();
    }
}

When there is a listener, the client opens a Producer span for the publish, writes the W3C traceparent into the message headers, and on the consumer side starts a Consumer span that continues from that header when the message arrives. The spans get OpenTelemetry messaging semantic convention attributes such as messaging.system, messaging.destination.name, messaging.rabbitmq.destination.routing_key, messaging.message.id, messaging.message.conversation_id (our CorrelationId) and messaging.message.body.size. An important detail: without a listener, no span is opened and no header is written. Because the outbox relay from the previous post ran in a process without OpenTelemetry, the messages it published showed up in this demo as brand-new traces without a parent.

Carrying Trace Context by Hand in Kafka

Confluent.Kafka itself does not publish an ActivitySource, and the Kafka instrumentation package in OpenTelemetry contrib is still alpha (0.5.0-alpha.1) on the day of writing. The stable route is to call the propagator by hand:

public static class KafkaTracing
{
    private static readonly TextMapPropagator Propagator = Propagators.DefaultTextMapPropagator;

    public static async Task ProduceAsync<T>(IProducer<string, byte[]> producer, string topic, string key, T message)
    {
        // Confluent.Kafka has no stable instrumentation: we open the producer span ourselves
        using Activity? activity = MessagingTelemetry.Source.StartActivity($"send {topic}", ActivityKind.Producer);
        activity?.SetTag("messaging.system", "kafka");
        activity?.SetTag("messaging.operation.type", "send");
        activity?.SetTag("messaging.destination.name", topic);

        var headers = new Confluent.Kafka.Headers();
        Propagator.Inject(new PropagationContext(activity?.Context ?? default, Baggage.Current), headers,
            static (h, name, value) => h.Add(name, Encoding.UTF8.GetBytes(value)));
        await producer.ProduceAsync(topic, new Message<string, byte[]>
        {
            Key = key,
            Value = JsonSerializer.SerializeToUtf8Bytes(message, JsonSerializerOptions.Web),
            Headers = headers
        });
    }

    public static Activity? StartProcess(ConsumeResult<string, byte[]> result)
    {
        PropagationContext parent = Propagator.Extract(default, result.Message.Headers,
            static (h, name) => h.TryGetLastBytes(name, out byte[] value) ? [Encoding.UTF8.GetString(value)] : []);
        Baggage.Current = parent.Baggage;
        Activity? activity = MessagingTelemetry.Source.StartActivity($"process {result.Topic}", ActivityKind.Consumer,
            parent.ActivityContext);
        activity?.SetTag("messaging.system", "kafka");
        activity?.SetTag("messaging.operation.type", "process");
        activity?.SetTag("messaging.destination.name", result.Topic);
        activity?.SetTag("messaging.kafka.offset", result.Offset.Value);
        return activity;
    }
}

Once the OpenTelemetry SDK (Software Development Kit) is set up, Propagators.DefaultTextMapPropagator is the composite propagator that carries W3C trace context and baggage together. Inject writes the traceparent (and, if present, tracestate and baggage) headers into the Kafka headers as UTF-8 (Unicode Transformation Format); Extract does the reverse. The Notification worker opens the span for every message with using Activity? activity = KafkaTracing.StartProcess(result);. In Kafka's synchronous Consume loop, Activity.Current stays valid on that thread for the lifetime of the span, so log lines automatically get the right trace id as well.

An Order's Journey in One Trace

The output of a single POST /orders request (the API response and the span lines, in the order they finished):

{"orderId":"4a9febc7-b2b1-46cd-8e03-9e87a6e97f13","traceId":"3738626c672a77c93609c10d195ae5be"}
span trace=3738626c id=946803 parent=4f2c3e Producer publish order.placed
span trace=3738626c id=4f2c3e parent=-      Server   POST /orders
span trace=3738626c id=62b35a parent=1429e7 Producer send bookstore.invoices
span trace=3738626c id=1429e7 parent=946803 Consumer deliver order.placed
span trace=3738626c id=5eac68 parent=62b35a Consumer process bookstore.invoices

The answer to the opening question: one trace. All five spans are in trace 3738626c, and the parent chain is unbroken: the HTTP request (4f2c3e) → RabbitMQ publish (946803) → deliver in Billing (1429e7) → send to Kafka (62b35a) → process in Notification (5eac68). The order in which spans finish differs from the order in which they start: the publish ends before the HTTP response; Billing's deliver span does not close until the Kafka send inside it has finished. In a trace viewer this appears as a single waterfall timeline and answers the question "how many milliseconds after the order was placed did the e-mail go out?".

The consumer span's parent is the producer span; that is RabbitMQ.Client 7's default behavior. The semantic conventions also define linking the consumer span to the producer with a "link" instead of a parent, especially for batch consumption. For consumers that process one message at a time, the parent relationship makes the trace easier to read; a consumer that processes thousands of messages in one batch needs links to a separate trace per message.

Putting the Correlation Id in the Log Scope

The trace id flows into the logs by itself: when Activity.Current is set, .NET logging adds a scope with SpanId, TraceId and ParentId to every log line. We add the message-specific IDs:

consumer.ReceivedAsync += async (_, ea) =>
{
    long started = Stopwatch.GetTimestamp();
    string outcome = "success";
    // the logger already adds TraceId/SpanId; we add the message-specific IDs
    using IDisposable? scope = logger.BeginScope(new Dictionary<string, object?>
    {
        ["MessageId"] = ea.BasicProperties.MessageId,
        ["CorrelationId"] = ea.BasicProperties.CorrelationId
    });
    try
    {
        OrderPlaced order = JsonSerializer.Deserialize<OrderPlaced>(ea.Body.Span, JsonSerializerOptions.Web)!;
        var invoice = new InvoiceCreated(Guid.NewGuid(), order.OrderId, order.Total);
        logger.LogInformation("Invoice {InvoiceId} created", invoice.InvoiceId);
        await KafkaTracing.ProduceAsync(producer, "bookstore.invoices", order.OrderId.ToString(), invoice);
        await channel.BasicAckAsync(ea.DeliveryTag, false);
    }
    catch (Exception ex)
    {
        outcome = "failure";
        telemetry.Failures.Add(1, new KeyValuePair<string, object?>("messaging.destination.name", "lab12.billing"));
        logger.LogError(ex, "Billing failed");
        await channel.BasicRejectAsync(ea.DeliveryTag, requeue: false);
    }
    finally
    {
        telemetry.ProcessDuration.Record(Stopwatch.GetElapsedTime(started).TotalSeconds,
            new KeyValuePair<string, object?>("messaging.system", "rabbitmq"),
            new KeyValuePair<string, object?>("messaging.destination.name", "lab12.billing"),
            new KeyValuePair<string, object?>("outcome", outcome));
    }
};

The JSON (JavaScript Object Notation) console formatter is on with IncludeScopes = true. The log line Billing writes:

{"EventId": 0, "LogLevel": "Information", "Category": "Bookstore.Messaging.BillingWorker", "Message": "Invoice 980f6fe7-853c-40d4-af8f-dd5b1cf6f71a created", "State": {"InvoiceId": "980f6fe7-853c-40d4-af8f-dd5b1cf6f71a", "{OriginalFormat}": "Invoice {InvoiceId} created"}, "Scopes": [{"Message": "SpanId:1429e7a7d4daeda3, TraceId:3738626c672a77c93609c10d195ae5be, ParentId:94680345bbaaa52b", "SpanId": "1429e7a7d4daeda3", "TraceId": "3738626c672a77c93609c10d195ae5be", "ParentId": "94680345bbaaa52b"}, {"Message": "System.Collections.Generic.Dictionary`2[System.String,System.Object]", "MessageId": "59816837-63b6-48e0-918b-8259b1d5b2e0", "CorrelationId": "4a9febc7-b2b1-46cd-8e03-9e87a6e97f13"}]}

The line holds the API's trace id (3738626c...), Billing's deliver span (1429e7...), the message id and the order id together. In your log search tool, a query by CorrelationId brings back the lines from every service related to an order; a query by trace id brings back the chain of a single request. In my first attempt I had also added a TraceId to the scope; it showed up twice in the output, because the logger already adds it. The details of structured logging are in the upcoming Logging in .NET series.

Metrics: Processing Duration and Failures

A trace tells the story of one message; metrics tell you the overall health of the system. A Meter from IMeterFactory and two instruments:

public sealed class MessagingTelemetry
{
    public const string SourceName = "Bookstore.Messaging";
    public const string MeterName = "Bookstore.Messaging";
    public static readonly ActivitySource Source = new(SourceName);

    public Histogram<double> ProcessDuration { get; }
    public Counter<long> Failures { get; }

    public MessagingTelemetry(IMeterFactory meterFactory)
    {
        Meter meter = meterFactory.Create(MeterName);
        // metric name and unit from the OpenTelemetry messaging semantic conventions
        ProcessDuration = meter.CreateHistogram<double>("messaging.process.duration", unit: "s",
            description: "Duration of processing a message");
        Failures = meter.CreateCounter<long>("bookstore.messaging.failures", unit: "{message}");
    }
}

In the demo a MeterListener prints every recorded measurement:

metric messaging.process.duration=0.0139s messaging.system=rabbitmq messaging.destination.name=lab12.billing outcome=success

The name messaging.process.duration and the unit of seconds come from the OpenTelemetry messaging semantic conventions. Those conventions are still in "Development" status and recommend the error.type attribute for failures; I used outcome here for readability, and if you want to follow the standard, add the exception type as error.type in the failure branch. Mind cardinality when choosing attributes: low-cardinality attributes such as queue name, outcome and system go on metrics; values that change with every message, such as MessageId or OrderId, go only on spans and logs. A metric that opens a separate time series per order brings down a metrics backend in short order.

Two things can be read from this histogram: p95/p99 processing time (is the consumer slowing down?) and processing count per second (how much work is the consumer doing?). Putting the failure counter next to the retry and DLQ (Dead-Letter Queue) counters is the basis of the alerts from the eighth and ninth posts of the series.

Queue Depth and Consumer Lag

Processing duration shows the inside of the consumer; it does not show the work piling up behind it. In RabbitMQ and Service Bus that number is the queue depth, the number of waiting messages. In RabbitMQ, QueueDeclarePassiveAsync returns the number of ready messages (the DLQ health check from the earlier posts used it) and the Prometheus plugin publishes per-queue metrics; in Service Bus, ServiceBusAdministrationClient.GetQueueRuntimePropertiesAsync returns ActiveMessageCount. In Kafka messages are not deleted when processed; what you measure is consumer lag, the difference in each partition between the last offset and the offset the group has committed:

public static class ConsumerLag
{
    public static async Task<object> GetAsync(string bootstrap, string groupId, string topic)
    {
        using IAdminClient admin = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrap }).Build();
        Metadata metadata = admin.GetMetadata(topic, TimeSpan.FromSeconds(5));
        List<TopicPartition> partitions = [.. metadata.Topics[0].Partitions.Select(p => new TopicPartition(topic, p.PartitionId))];

        // offsets committed by the group
        List<ListConsumerGroupOffsetsResult> committed = await admin.ListConsumerGroupOffsetsAsync(
            [new ConsumerGroupTopicPartitions(groupId, partitions)]);

        // the last offset of each partition (high watermark)
        using IConsumer<Ignore, Ignore> probe = new ConsumerBuilder<Ignore, Ignore>(
            new ConsumerConfig { BootstrapServers = bootstrap, GroupId = $"{groupId}-lag-probe" }).Build();
        var perPartition = committed[0].Partitions.Select(tpo =>
        {
            WatermarkOffsets w = probe.QueryWatermarkOffsets(tpo.TopicPartition, TimeSpan.FromSeconds(5));
            long position = tpo.Offset == Offset.Unset ? w.Low.Value : tpo.Offset.Value;
            return new { partition = tpo.Partition.Value, committed = position, end = w.High.Value, lag = w.High.Value - position };
        }).ToList();
        return new { group = groupId, topic, totalLag = perPartition.Sum(p => p.lag), partitions = perPartition };
    }
}

I started the application with the Notification worker turned off, placed 5 orders, then restarted it with the worker on. Three measurements from GET /lag/notification:

{"group":"notification","topic":"bookstore.invoices","totalLag":0,"partitions":[{"partition":0,"committed":65,"end":65,"lag":0},{"partition":1,"committed":81,"end":81,"lag":0},{"partition":2,"committed":73,"end":73,"lag":0}]}
{"group":"notification","topic":"bookstore.invoices","totalLag":5,"partitions":[{"partition":0,"committed":65,"end":67,"lag":2},{"partition":1,"committed":81,"end":82,"lag":1},{"partition":2,"committed":73,"end":75,"lag":2}]}
{"group":"notification","topic":"bookstore.invoices","totalLag":0,"partitions":[{"partition":0,"committed":67,"end":67,"lag":0},{"partition":1,"committed":82,"end":82,"lag":0},{"partition":2,"committed":75,"end":75,"lag":0}]}

The five invoices were spread across three partitions (key = OrderId), and with no worker the lag rose to 5; once the worker came back it dropped to zero. As the messages left behind were processed, each one attached to its own old trace: Notification's process span was added to the order's trace that had started minutes earlier, and the waiting time is plainly visible in the trace viewer. The trend of the lag matters as much as its absolute value: a steady 50 may be healthy, while a lag going from 0 to 5000 in 10 minutes says the consumer has fallen behind. Turning lag into time (the age of the oldest unprocessed message) makes it easier to discuss the alert threshold in business terms: "Notification is more than 2 minutes behind".

Production Notes

Choose sampling deliberately. Keeping the trace of every message is expensive. With head-based sampling the decision is made at the start of the trace and carried to every service through the sampled flag in traceparent; if the HTTP request was not sampled, the whole message chain is not sampled either. If you want to keep failed or slow traces selectively, that decision is made in the Collector with tail-based sampling.

Other clients. Azure.Messaging.ServiceBus's ActivitySource support is experimental and is turned on with the Azure.Experimental.EnableActivitySource switch; you then listen to it with AddSource("Azure.*"). MassTransit publishes its own source, added with DiagnosticHeaders.DefaultListenerName. Check which header each client uses: if one system has two different propagation formats (for example the old Diagnostic-Id and traceparent), the chain breaks in the middle.

Messages that pass through the outbox. The relay is a separate background job and does not carry trace context; write Activity.Current?.Id into the outbox row as a column and put it into a header in the relay. That way the "order saved → relay → consumer" chain shows up in one trace, including the time the relay kept it waiting.

Take metrics from the broker, not only from the app. Queue depth measured from inside the application is not measured when the application is not running, exactly when you need it most. Collecting queue depth and lag from the broker's own exporter (the RabbitMQ Prometheus plugin, Azure Monitor, Kafka lag exporters) closes that blind spot.

Common Pitfalls

Forgetting AddSource. RabbitMQ.Client 7's instrumentation is built in, but without a listener there is neither a span nor a header; downstream services start a new trace for every message.

Writing the header with the wrong encoding. A Kafka header is a byte[]; write and read traceparent as UTF-8. If services are written in different languages, make sure they agree on the same format.

High-cardinality metric attributes. Turning MessageId or CustomerId into a metric tag drowns the metrics backend in time series.

Adding the trace id to the log scope by hand. The logger already adds it from Activity.Current; adding it by hand writes the same value twice and can carry a stale value after the span changes.

Measuring queue depth only from inside the application. When the consumer crashes, the metric goes silent too; the alert never fires.

A Small Experiment

★☆☆ Add the span duration (a.Duration) to SpanLineProcessor and verify from the durations that the deliver order.placed span contains the send bookstore.invoices span.

★★☆ Add a TraceParent column to the outbox from the previous post; write Activity.Current?.Id in the endpoint, put it into BasicProperties.Headers["traceparent"] in the relay and start the relay's own span from that context. Verify that the consumer's span now shows up in the HTTP request's trace.

★★★ Hook ConsumerLag up to an ObservableGauge<long> (one measurement per partition, with messaging.consumer.group.name and the partition as attributes) and have it measure every 15 seconds. Stop Notification, produce 100 messages, and show the gauge rising and falling again once the worker is back.

We can now see a message's journey. In the next post we secure that journey with automated tests: unit testing handler logic without a broker, integration tests against a real RabbitMQ with Testcontainers, the MassTransit test harness, and contract tests proving that a v2 consumer can read a v1 message.