异步队列任务追踪石沉大海？利用 OpenTelemetry 实现 Celery 跨服务分布式调用链路追踪

线上排查一个诡异的业务 Bug，链路追踪查到 HTTP 接口服务就断了。排查了半天，才发现罪魁祸首是接口里丢进 Celery 异步队列的一个后台任务。由于 Trace Context 没有自动透传给 Celery 消费者，分布式链路追踪的 Trace ID 在入队的那一刻直接“石沉大海”。每次遇到这种异步队列报错，只能靠关键词去 Kibana 里大海捞针式地人肉匹配日志，心智负担极重。

Celery 官方的 opentelemetry-instrumentation-celery 插桩库在面对跨服务的复杂链路时，经常因为上下文传播器（Carrier）配置不正确而失效。为了彻底解决这个问题，干脆手动实现 Celery 信号（signals）注入与提取 Trace Context 的逻辑。

解决这个大坑的核心在于：在 Celery 任务发布（before_task_publish）时，将当前的 Trace Context 注入到 Celery 消息的 headers 中；然后在 Celery 任务执行前（task_prerun），从 headers 中提取上下文，并将其作为当前线程的父上下文激活。

下面是经过线上验证的完整初始化与注入代码：

import logging
from celery.signals import before_task_publish, task_prerun, task_postrun
from opentelemetry import trace
from opentelemetry.context import attach, detach
from opentelemetry.propagate import inject, extract
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

logger = logging.getLogger(__name__)

def init_tracer(service_name: str, endpoint: str):
    resource = Resource.create(attributes={"service.name": service_name})
    provider = TracerProvider(resource=resource)
    processor = BatchSpanProcessor(OTLPSpanExporter(endpoint=endpoint, insecure=True))
    provider.add_span_processor(processor)
    trace.set_tracer_provider(provider)

@before_task_publish.connect
def on_task_publish(headers=None, body=None, sender=None, **kwargs):
    if headers is None:
        return
    
    # 提取当前活跃的 Trace 上下文，注入到 Celery 消息 headers 中
    inject(headers)
    logger.debug("Successfully injected trace context into Celery headers: %s", headers)

@task_prerun.connect
def on_task_prerun(task_id=None, task=None, *args, **kwargs):
    # Celery 任务执行前，尝试从 request headers 中提取 Trace 上下文
    request = task.request
    headers = getattr(request, "headers", None) or {}
    
    ctx = extract(headers)
    # 激活提取出的上下文，并将 token 保存到任务实例中，以便后续释放
    token = attach(ctx)
    setattr(task, "__trace_token", token)
    
    # 启动当前任务的 Span
    tracer = trace.get_tracer("celery-worker")
    span = tracer.start_span(
        name=task.name,
        context=ctx,
        attributes={
            "messaging.system": "celery",
            "messaging.destination": task.queue or "default",
            "messaging.celery.task_id": task_id
        }
    )
    setattr(task, "__trace_span", span)

@task_postrun.connect
def on_task_postrun(task_id=None, task=None, retval=None, state=None, **kwargs):
    # 任务执行结束，关闭 Span
    span = getattr(task, "__trace_span", None)
    if span:
        if state == "FAILURE":
            span.set_status(trace.StatusCode.ERROR)
            if isinstance(retval, Exception):
                span.record_exception(retval)
            else:
                span.set_attribute("error.message", str(retval))
        else:
            span.set_status(trace.StatusCode.OK)
        span.end()
    
    # 恢复原有的上下文
    token = getattr(task, "__trace_token", None)
    if token:
        detach(token)

折腾完这套逻辑后，我们用 Jaeger 观察调用链，发现 HTTP 接口、Celery 生产者和 Celery 消费者的 Span 终于完美连成了一条线。不仅能清晰看到任务在队列里积压了多久（通过对比入队 Span 和执行 Span 的时间戳），还能一眼看清哪个下游 gRPC 服务挂了。

如果你们也遇到类似的跨服务异步追踪丢失问题，建议直接把这套信号监听器贴进项目的 celery.py 初始化代码里。不过需要特别注意：在生产高并发场景下，默认的 OTLP gRPC Exporter 可能会引起网络阻塞，建议把队列大小和批量发送超时调大，或者干脆本地部署 Otelemetry Collector 用 UDP 转发。

大家在用 OpenTelemetry 做异步任务追踪时，还踩过哪些比较恶心的深坑？