如何通过Kafka传递OpenTelemetry链路追踪Span？

你的问题出在手动拆分并序列化SpanContext的方式——这种做法会遗漏OTel追踪上下文的关键元数据（比如采样标志TraceFlags），而且不符合官方传播规范，导致Agent无法正确识别并上报子Span。

OTel官方推荐使用**传播器（Propagator）**来序列化和传递追踪上下文，而非手动拆解字段。虽然HTTP传播器较为常见，但TextMapPropagator是通用型传播器，完全适配Kafka这类消息队列场景，只需将上下文序列化成字节数组即可。

正确实现步骤 #

1. 初始化传播器 #

推荐使用W3C Trace Context传播器（OTel默认标准），也可根据需求组合其他传播器：

import (
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/propagation"
)

// 服务启动时初始化一次全局传播器
func init() {
    otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
        propagation.TraceContext{}, // W3C Trace Context标准
        propagation.Baggage{},       // 可选，用于传递业务元数据
    ))
}

2. 生产者端：序列化追踪上下文并发送 #

从当前上下文提取SpanContext，用传播器注入到载体，再将载体序列化成字节数组发送到Kafka：

import (
    "encoding/json"
    "context"
    "go.opentelemetry.io/otel"
)

func produceKafkaMessage(ctx context.Context, producer KafkaProducer, topic string, payload []byte) error {
    // 创建载体存储传播的键值对
    carrier := make(map[string]string)
    // 用全局传播器将上下文注入到载体
    otel.GetTextMapPropagator().Inject(ctx, propagation.MapCarrier(carrier))
    
    // 将载体序列化成字节数组（这里用JSON，也可使用Protobuf等其他格式）
    carrierBytes, err := json.Marshal(carrier)
    if err != nil {
        return err
    }
    
    // 将追踪上下文字节数组作为消息头或消息体的一部分发送
    return producer.Send(ctx, topic, payload, map[string][]byte{"trace-context": carrierBytes})
}

3. 消费者端：解析字节数组并重建上下文 #

从Kafka消息中获取字节数组，还原成载体，再用传播器提取上下文，最后创建子Span：

import (
    "encoding/json"
    "context"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/trace"
)

func consumeKafkaMessage(ctx context.Context, msg KafkaMessage, tracer trace.Tracer) error {
    // 从消息头获取追踪上下文字节数组
    carrierBytes := msg.Headers["trace-context"]
    if carrierBytes == nil {
        // 无上下文时直接创建根Span
        ctx, span := tracer.Start(ctx, "consume-kafka-message")
        defer span.End()
        return processMessage(ctx, msg.Payload)
    }
    
    // 将字节数组还原成载体
    var carrier map[string]string
    if err := json.Unmarshal(carrierBytes, &carrier); err != nil {
        return err
    }
    
    // 用传播器从载体提取远程上下文
    remoteCtx := otel.GetTextMapPropagator().Extract(ctx, propagation.MapCarrier(carrier))
    
    // 创建子Span
    ctx, span := tracer.Start(remoteCtx, "consume-kafka-message")
    defer span.End()
    
    return processMessage(ctx, msg.Payload)
}

原代码失效原因 #

原代码仅手动传递了TraceID、SpanID和TraceState，但遗漏了**TraceFlags**——这个字段包含采样标志（比如是否要上报该Span）。如果采样标志未正确传递，OTel Agent会判定该Span无需上报，自然不会发送到otel-collector。

此外，手动处理上下文不符合OTel传播规范，后续若需传递新的元数据（如Baggage），这种方式也无法兼容。

内容的提问来源于stack exchange，提问作者ashish malgawa