如何在Python版Azure Functions与APIM间传播OpenTelemetry追踪上下文？

工作流概述 #

• Function_A 向API Management发送HTTP请求
• API Management验证请求后，将消息推送至Azure Service Bus
• Function_B由Service Bus消息触发

需求：全程传播W3C traceparent 和 tracestate 头部，使Application Insights中所有遥测数据关联到同一Operation_Id。

当前状态 #

• Python版Azure Functions已启用OpenTelemetry
• Function_A到APIM的traceparent传播正常
• APIM通过Headers将traceparent/tracestate传递至Service Bus，Service Bus到Function_B通过context.trace_context传播

问题现象 #

• Function_B触发时会生成新的Operation_Id，导致追踪关联中断
• 原trace-id被设置为Operation_ParentId，可被收集为ai_legacyRootId
• 临时方案：使用OpenTelemetry在Function_B中生成的traceparent追踪业务代码，但系统级遥测（如Microsoft.Azure.WebJob.Script.Workers.Rpc.RpcException）无法关联到正确的Operation_Id，仅span范围内的追踪有效

现有配置与代码 #

API Management出站配置 #

<outbound>
    <base />
    <set-header name="traceparent" exists-action="override">
        <value>@(context.Request.Headers.GetValueOrDefault("traceparent", ""))</value>
    </set-header>
    <set-header name="tracestate" exists-action="override">
        <value>@(context.Request.Headers.GetValueOrDefault("tracestate", ""))</value>
    </set-header>
</outbound>

Function_B代码 #

async def main(msg: func.ServiceBusMessage, context) -> None:
    """Enrichment of system data"""
    carrier = {
        'traceparent': msg.user_properties.get('traceparent'),
        'tracestate': msg.user_properties.get('tracestate', ''),
    }
    tracer = trace.get_tracer(__name__)

    with tracer.start_as_current_span(
        'function_b/do',
        context=extract(carrier),
    ):
        assert 1 == 0

期望目标 #

1. 原trace-id作为整个链路的Operation_Id
2. Function_B的Parent_Id为Function_A的span-id

解决方案 #

1. 调整APIM到Service Bus的属性传递 #

Service Bus的Headers不会自动映射到Function触发时的追踪上下文，需将traceparent/tracestate放入消息的User Properties中，确保Function_B能读取到完整的追踪上下文：

<outbound>
    <base />
    <!-- 保留原Header设置，同时添加到消息属性 -->
    <set-header name="traceparent" exists-action="override">
        <value>@(context.Request.Headers.GetValueOrDefault("traceparent", ""))</value>
    </set-header>
    <set-header name="tracestate" exists-action="override">
        <value>@(context.Request.Headers.GetValueOrDefault("tracestate", ""))</value>
    </set-header>
    <!-- 将追踪上下文添加到Service Bus消息的User Properties -->
    <set-property name="traceparent" value="@(context.Request.Headers.GetValueOrDefault("traceparent", ""))" />
    <set-property name="tracestate" value="@(context.Request.Headers.GetValueOrDefault("tracestate", ""))" />
</outbound>

注：若APIM通过"Send to Service Bus"策略发送消息，需确保在策略中明确将这些属性作为消息属性传递。

2. 修改Function_B的追踪上下文初始化 #

Azure Functions宿主默认会生成新的追踪上下文，需手动提取并覆盖全局上下文，同时修改Application Insights遥测处理器，让系统级遥测也关联到正确的Operation_Id：

from opentelemetry import trace
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter
from opentelemetry.instrumentation.azure_functions import AzureFunctionsInstrumentor
import azure.functions as func
from opentelemetry.context import attach, detach
import logging
from opentelemetry.sdk._logs import LoggingHandler

# 初始化OpenTelemetry追踪器
def init_trace_provider():
    exporter = AzureMonitorTraceExporter.from_connection_string("<你的Application Insights连接字符串>")
    provider = TracerProvider()
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)

init_trace_provider()
AzureFunctionsInstrumentor().instrument()

async def main(msg: func.ServiceBusMessage, context) -> None:
    traceparent = msg.user_properties.get('traceparent')
    tracestate = msg.user_properties.get('tracestate', '')
    token = None

    if traceparent:
        # 解析并附加传入的追踪上下文
        carrier = {'traceparent': traceparent, 'tracestate': tracestate}
        trace_context = TraceContextTextMapPropagator().extract(carrier)
        token = attach(trace_context)

        # 修改Application Insights遥测处理器的Operation属性
        current_span = trace.get_current_span()
        span_context = current_span.get_span_context()
        trace_id_hex = format(span_context.trace_id, '032x')
        parent_span_id_hex = format(span_context.span_id, '016x')

        for handler in logging.getLogger().handlers:
            if isinstance(handler, LoggingHandler):
                # 更新遥测的Operation ID和Parent ID
                for processor in handler._exporter._telemetry_processors:
                    if hasattr(processor, '_operation_id'):
                        processor._operation_id = trace_id_hex
                    if hasattr(processor, '_operation_parent_id'):
                        processor._operation_parent_id = parent_span_id_hex

    try:
        tracer = trace.get_tracer(__name__)
        with tracer.start_as_current_span('function_b/do'):
            # 业务逻辑代码
            assert 1 == 0
    finally:
        if token:
            detach(token)

3. 依赖包安装 #

确保安装所需的Python包：

pip install azure-functions opentelemetry-api opentelemetry-sdk opentelemetry-instrumentation-azure-functions azure-monitor-opentelemetry-exporter

原理说明 #

• Azure Functions Service Bus触发器默认不读取消息中的追踪上下文，会生成独立的Operation_Id。通过手动附加传入的traceparent上下文，让OpenTelemetry全局上下文使用链路的trace-id。
• 修改Application Insights的遥测处理器，确保系统级日志、异常等遥测数据的Operation_Id和Parent_Id与业务代码追踪保持一致，实现全链路关联。

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