OpenTelemetry如何跨进程开启与关闭Span（Golang+NATS场景）

OpenTelemetry 官方本身就不支持「跨进程传入Span ID获取Span实例、远程管控Span生命周期」的能力——这是刻意设计的：Span是进程内的SDK资源，所有对Span的修改、结束操作都必须在创建该Span的进程内完成，跨进程仅传递符合W3C标准的Trace上下文（TraceID、SpanID、采样标记等），不暴露跨进程直接操作Span的API。
你要实现「生产侧启动Span，等下游处理完成再结束」的异步链路场景，不需要找不存在的跨进程Span操作接口，按照异步工作流追踪的标准方案实现即可。

核心实现逻辑 #

整体逻辑拆成三个环节，所有Span的操作都留在创建它的生产侧进程内完成，跨节点只传上下文和处理结果信号：

1. 生产侧创建Span后不立即结束，和消息唯一ID绑定存在进程内缓存
2. 下游节点从消息头提取上下文创建子Span执行业务，处理完成后回传携带消息ID的处理结果
3. 生产侧监听处理结果回调，匹配到对应Span补录状态后结束

分步代码实现 #

• 生产侧初始化与消息发送逻辑
  注意不要在发送函数内用defer span.End()，同时给每个消息生成全局唯一ID，用于后续回调匹配，把Trace上下文通过OTel标准传播器注入NATS消息头：

  import (
    "context"
    "errors"
    "sync"
  
    "github.com/google/uuid"
    "github.com/nats-io/nats.go"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/codes"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/trace"
  )
  
  var tracer = otel.Tracer("nats-async-workflow")
  // 进程内缓存待结束的Span，key为消息唯一ID，生产环境需替换为带读写锁、过期淘汰逻辑的存储
  var pendingSpans sync.Map
  
  // PublishMessage 发送带追踪上下文的消息
  func PublishMessage(js nats.JetStreamContext, subject string, payload []byte) error {
    // 标记为生产者类型的异步Span
    ctx, span := tracer.Start(context.Background(), "workflow.produce",
      trace.WithSpanKind(trace.SpanKindProducer),
    )
  
    msgID := uuid.NewString()
    pendingSpans.Store(msgID, span)
  
    // 构造NATS消息，注入Trace上下文到消息头
    msg := nats.NewMsg(subject)
    msg.Data = payload
    msg.Header.Set("X-Msg-ID", msgID)
    otel.GetTextMapPropagator().Inject(ctx, propagation.HeaderCarrier(msg.Header))
  
    _, err := js.PublishMsg(msg)
    if err != nil {
      span.RecordError(err)
      span.End()
      pendingSpans.Delete(msgID)
      return err
    }
    return nil
  }
  

• 下游消费侧处理逻辑
  从消息头提取生产侧传递的Trace上下文，创建消费侧子Span执行业务逻辑，处理完成后往约定的回调主题回传处理结果，带上原消息ID：

  // StartConsumer 启动下游消费者
  func StartConsumer(js nats.JetStreamContext) error {
    _, err := js.Subscribe("biz.target.subject", func(msg *nats.Msg) {
      // 提取上游传递的Trace上下文
      ctx := otel.GetTextMapPropagator().Extract(
        context.Background(),
        propagation.HeaderCarrier(msg.Header),
      )
      // 创建消费侧子Span，随当前处理逻辑结束
      _, span := tracer.Start(ctx, "workflow.consume",
        trace.WithSpanKind(trace.SpanKindConsumer),
      )
      defer span.End()
  
      msgID := msg.Header.Get("X-Msg-ID")
      processErr := DoBizLogic(ctx, msg.Data) // 替换为你的实际业务逻辑
  
      // 构造回调消息
      ackMsg := nats.NewMsg("biz.target.subject.ack")
      ackMsg.Header.Set("X-Msg-ID", msgID)
      if processErr != nil {
        span.RecordError(processErr)
        span.SetStatus(codes.Error, processErr.Error())
        ackMsg.Header.Set("X-Process-Status", "error")
        ackMsg.Header.Set("X-Process-Err", processErr.Error())
      } else {
        span.SetStatus(codes.Ok, "process success")
        ackMsg.Header.Set("X-Process-Status", "success")
      }
  
      // 发送回调，确认原消息
      js.PublishMsg(ackMsg)
      msg.Ack()
    })
    return err
  }
  

• 生产侧回调监听逻辑
  生产侧服务启动时就订阅回调主题，收到回调后根据消息ID从缓存中取出对应Span，补录处理状态后结束Span：

  // InitAckListener 初始化生产侧回调监听
  func InitAckListener(js nats.JetStreamContext) error {
    _, err := js.Subscribe("biz.target.subject.ack", func(ackMsg *nats.Msg) {
      msgID := ackMsg.Header.Get("X-Msg-ID")
      spanVal, ok := pendingSpans.LoadAndDelete(msgID)
      if !ok {
        return
      }
      span := spanVal.(trace.Span)
  
      status := ackMsg.Header.Get("X-Process-Status")
      if status == "error" {
        errMsg := ackMsg.Header.Get("X-Process-Err")
        span.RecordError(errors.New(errMsg))
        span.SetStatus(codes.Error, errMsg)
      } else {
        span.SetStatus(codes.Ok, "downstream process success")
      }
      // 此时才结束生产侧最初创建的Span
      span.End()
    })
    return err
  }
  

生产环境注意事项 #

• 不要尝试自行实现跨进程通过SpanID操作Span的逻辑，这类实现会破坏OpenTelemetry的上下文语义，极易造成链路数据错乱、内存泄漏问题。
• 待结束Span的缓存必须配置过期淘汰机制：后台定时扫描缓存中超过业务最大处理时长的Span，标记为超时异常后主动结束，避免下游宕机未回传ACK导致Span永久驻留内存。
• 如果使用NATS JetStream的原生ACK/NAK机制，可以不用单独发回调消息，直接利用JetStream的发布确认、消费者通知事件触发Span结束，核心逻辑不变：所有Span的End调用必须在创建它的进程内执行。

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