在Application Insights中追踪Azure Service Bus消息（Python实现）

核心思路 #

基于OpenCensus的Trace上下文（Trace ID、Span ID），将追踪信息嵌入Service Bus消息的自定义属性中，接收端读取这些属性后重建Trace上下文，启动关联的Span，从而实现跨服务的端到端追踪链路。

发送端实现步骤 #

1. 获取当前Trace上下文
   在发送Service Bus消息前，从OpenCensus当前活跃Span中提取Trace ID和Span ID，构造符合W3C标准的追踪标识：

from opencensus.trace import execution_context

current_span = execution_context.get_current_span()
trace_parent = None
if current_span:
    trace_id = current_span.context.trace_id
    span_id = current_span.context.span_id
    # 按W3C Trace Context格式构造
    trace_parent = f"00-{trace_id}-{span_id}-01"

2. 将追踪属性附加到消息
   使用Azure Service Bus Python SDK的消息对象，添加自定义属性传递追踪上下文：

from azure.servicebus import ServiceBusMessage

message = ServiceBusMessage("your_message_content")
if trace_parent:
    message.application_properties["traceparent"] = trace_parent
    message.application_properties["operation_name"] = "service_bus_send"

# 发送消息
with servicebus_client.get_queue_sender(queue_name=queue_name) as sender:
    sender.send_messages(message)

3. 创建发送Span（推荐）
   为Service Bus发送操作单独创建Span，让追踪链路更清晰：

from opencensus.trace.tracer import Tracer
from opencensus.trace.samplers import AlwaysOnSampler

tracer = Tracer(sampler=AlwaysOnSampler())
with tracer.span(name="ServiceBus.Send") as send_span:
    send_span.add_attribute("service_bus.queue", queue_name)
    send_span.add_attribute("message.id", message.message_id)
    # 执行上述消息构造与发送逻辑

接收端实现步骤 #

1. 从消息中提取追踪属性
   接收消息时读取自定义属性中的traceparent字段，解析出Trace ID和父Span ID：

from azure.servicebus import ServiceBusReceiver

with servicebus_client.get_queue_receiver(queue_name=queue_name) as receiver:
    for msg in receiver:
        trace_parent = msg.application_properties.get("traceparent")
        trace_id = None
        parent_span_id = None
        if trace_parent:
            parts = trace_parent.split("-")
            if len(parts) == 4:
                trace_id = parts[1]
                parent_span_id = parts[2]

2. 重建Trace上下文并启动接收Span
   用解析出的信息创建关联的Span，延续原有追踪链路：

from opencensus.trace.tracer import Tracer
from opencensus.trace.samplers import AlwaysOnSampler
from opencensus.trace.status import Status, StatusCode

tracer = Tracer(sampler=AlwaysOnSampler())
with tracer.span(name="ServiceBus.Receive", parent_span_id=parent_span_id) as receive_span:
    # 绑定Trace ID，确保链路关联
    receive_span.context.trace_id = trace_id
    receive_span.add_attribute("service_bus.queue", queue_name)
    receive_span.add_attribute("message.id", msg.message_id)
    
    # 处理消息业务逻辑
    try:
        process_message(msg)
        receiver.complete_message(msg)
    except Exception as e:
        # 记录异常到Span
        receive_span.status = Status(StatusCode.ERROR, str(e))
        receive_span.add_attribute("error_detail", str(e))
        receiver.abandon_message(msg)
        raise

关键注意事项 #

• 遵循W3C标准：使用traceparent字段传递追踪信息，保证和.NET等其他语言实现的兼容性。
• 采样策略一致：发送端和接收端使用相同的采样策略（如AlwaysOnSampler），避免部分Span被过滤导致链路断裂。
• 全链路属性补充：在Span中添加队列名称、消息ID等业务属性，便于在Application Insights中快速定位问题。

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