Python-Flask API如何借助Traceparent在Azure Application Insights中关联日志

问题原因 #

你当前的代码中，日志记录使用全局logger，但未绑定到FlaskMiddleware为每个请求创建的请求级trace上下文。虽然FlaskMiddleware已经解析了Traceparent头并生成对应operation_Id，但日志调用未关联该上下文，导致trace表中的关联字段为空。

解决方案步骤 #

• 移除全局Tracer实例：FlaskMiddleware会自动为每个请求创建并维护Tracer，全局Tracer会干扰请求级上下文的使用。
• 关联请求级Trace上下文到日志：在日志记录时，从请求上下文获取当前Tracer的Span信息，将operation_Id（对应trace_id）和operation_ParentId（对应span_id）添加到日志的自定义维度中。
• 确保Logging集成生效：config_integration.trace_integrations(['logging'])已启用，需保证日志调用能获取到当前trace上下文。

修改后的关键代码片段 #

1. 移除全局Tracer #

删除以下行：

tracer = Tracer(exporter=azureExporter, sampler=AlwaysOnSampler())

2. 导入上下文获取工具 #

添加导入语句：

from opencensus.trace import execution_context

3. 修改日志记录函数（before_request/after_request/路由） #

以before_request为例：

@app.before_request
def before():
    properties = get_properties_for_customDimensions_from_request(request)
    # 获取当前请求的Trace上下文
    tracer = execution_context.get_opencensus_tracer()
    if tracer:
        span_context = tracer.span_context
        # 添加上下文关联字段到自定义维度
        properties['custom_dimensions'].update({
            'operation_Id': span_context.trace_id,
            'operation_ParentId': span_context.span_id
        })
    logger.warning("request {} {}".format(request.method, request.url), extra=properties)

after_request修改：

@app.after_request
def after(response):
    response_properties = get_properties_for_customDimensions_from_response(request,response)
    tracer = execution_context.get_opencensus_tracer()
    if tracer:
        span_context = tracer.span_context
        response_properties['custom_dimensions'].update({
            'operation_Id': span_context.trace_id,
            'operation_ParentId': span_context.span_id
        })
    logger.warning("response: {}".format(response.status), extra=response_properties)
    return response

路由函数health_check修改：

@app.route('/api/{}/status'.format("v1"), methods=['GET'])
def health_check():
    message = "Health ok!"
    properties = {'custom_dimensions': {}}
    tracer = execution_context.get_opencensus_tracer()
    if tracer:
        span_context = tracer.span_context
        properties['custom_dimensions'].update({
            'operation_Id': span_context.trace_id,
            'operation_ParentId': span_context.span_id
        })
    logger.info(message, extra=properties)
    return message

4. （可选）优化日志过滤器 #

如果希望所有日志自动关联trace上下文，可修改CustomDimensionsFilter，在filter方法中自动添加trace字段：

class CustomDimensionsFilter(logging.Filter):
    def __init__(self, custom_dimensions=None):
        self.custom_dimensions = custom_dimensions or {}

    def filter(self, record):
        dim = {**self.custom_dimensions, **getattr(record, "custom_dimensions", {})}
        # 自动添加trace上下文
        tracer = execution_context.get_opencensus_tracer()
        if tracer:
            span_context = tracer.span_context
            dim.update({
                'operation_Id': span_context.trace_id,
                'operation_ParentId': span_context.span_id
            })
        record.custom_dimensions = dim
        return True

这样无需在每个日志调用里手动添加trace字段，所有经过该过滤器的日志都会自动关联上下文。

验证效果 #

修改后，Flask API的trace表日志会自动带上与request表相同的operation_Id及对应的operation_ParentId，实现端到端日志关联。

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