Distributed traces with messages consumers

[ttha3]

Hello, I have a questions about the concept of traces, spans in distributed system. I have a producer, a queue and a consumer. The producer put each message into the queue with the trace information. And the consumer polls a batch of messages to process. How the distributed tracing is done for this usecase. I mean, the Producer put Message_1 with {TraceId_Producer_1, SpanId_1}, Message_2 with {TraceId_Producer_2, SpandId_2} to the queue. In the consumer, only ONE instance of consumer will create 2 spans to process the Message_1 and Message_2. These two spans share the same TraceId_Consumer. How I can group all spans of Message_1 to the TraceId_Producer_1 and all spans of Message_2 to the TraceId_Producer_2? My idea is to have a map of service that shows the flow from the producer, queue and consumer for each message (so each TraceId_Producer_1 and TraceId_Producer_2). I see that I can you SpanLink but with the link, I need to click to the linkedSpan to visualize the forward/backward processing. But I prefer having all of the treatments in one map. Thanks for your idea.

[dingminghbut]

If you want Message_1 to stay in TraceId_Producer_1 and Message_2 to stay in TraceId_Producer_2, then the consumer should not create one shared consumer trace for the whole batch.

For batch polling, I would model it as two levels:

1. Receive/poll operation: one span for the batch receive, often with links to the message contexts in the batch.
2. Per-message processing: one consumer/process span per message, created with that message's extracted context as the parent.

That means the processing for each message joins the trace that came from the producer.

In JS terms, the important part is to call propagation.extract() separately for each message, then run that message's processing inside that extracted context:

import { context, propagation, trace, SpanKind } from '@opentelemetry/api';

const tracer = trace.getTracer('consumer');

for (const message of messages) {
  const parentCtx = propagation.extract(context.active(), message.headers);

  await context.with(parentCtx, async () => {
    await tracer.startActiveSpan(
      'process message',
      {
        kind: SpanKind.CONSUMER,
        attributes: {
          'messaging.operation.name': 'process',
          'messaging.message.id': message.id,
        },
      },
      async span => {
        try {
          await processMessage(message);
        } catch (err) {
          span.recordException(err as Error);
          throw err;
        } finally {
          span.end();
        }
      },
    );
  });
}

Then the traces look like:

• Producer span for Message_1 -> Consumer processing span for Message_1
• Producer span for Message_2 -> Consumer processing span for Message_2

SpanLink is still useful for the batch receive span, because that single receive operation is related to many upstream contexts. But for the actual message handling, if you want the service map/timeline to show the full path for each message, make the per-message processing span a child of the extracted message context instead of linking only.

So the rule of thumb is:

• one operation caused by one message: use extracted context as parent;
• one operation caused by many messages: use links;
• if you need both, create a batch receive span with links and per-message process spans with extracted parents.