结构化日志和追踪

Python 日志与追踪 Python 日志追踪实践结构化日志与追踪

副标题/摘要：结合 logging + OpenTelemetry 实现结构化日志并把 trace_id 注入日志，便于在生产环境串联调用链与定位问题。

TL;DR：设置 json 格式日志并通过 OpenTelemetry 在每条日志里注入 trace_id/span_id。关键步骤：安装依赖 → 配置 logging（JSON）→ 配置 TracerProvider → 用 Filter 从当前 span 提取 trace 信息并添加到日志记录中。

背景与动机（为什么需要）
关键概念与术语解释
环境与依赖（安装命令）
逐步实战示例（可直接运行）
原理与实现要点
常见问题与注意事项
最佳实践总结
结论与下一步建议
可视化建议
参考与延伸阅读
可复制示例代码

背景与动机（为什么需要）现代后端服务分布式部署后，单靠文本日志很难把一次请求链路从入口到后端串起来。结构化日志（JSON）便于聚合与查询；而分布式追踪（tracing）给出调用链与 span 信息。二者结合能快速定位延迟与错误根因：日志告诉你“发生了什么”，trace 告诉你“这个请求经过了哪些服务/操作”。

关键概念与术语解释（简明）

日志（Logging）：程序运行时的事件记录，通常按级别（INFO/ERROR）输出。
结构化日志：以 JSON 等结构化格式输出，便于机器处理与检索。
Trace/Span：一次分布式操作（trace）由若干子操作（span）组成，span 含有 trace_id 与 span_id。
Context Propagation：在不同服务/线程/协程中传递 trace context 以串联调用链。

环境与依赖（列出安装命令）推荐环境：Python 3.8+ 安装依赖： pip install python-json-logger opentelemetry-api opentelemetry-sdk

如需控制台导出（用于 MVP 测试）： pip install opentelemetry-exporter-console

逐步实战示例（包含可直接运行的代码片段，注释清楚）下面例子展示：1) 配置 JSON 日志 2) 配置 OpenTelemetry TracerProvider（ConsoleExporter）3) 用 Filter 将 trace_id/span_id 注入每条日志。

# demo.py
import logging
import sys
from pythonjsonlogger import jsonlogger

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, ConsoleSpanExporter

# 1) 设置 TracerProvider 与 Console 导出（MVP）
trace.set_tracer_provider(TracerProvider())
tracer_provider = trace.get_tracer_provider()
tracer_provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
tracer = trace.get_tracer(__name__)

# 2) 自定义 logging Filter：从当前 span 中取 trace_id/span_id 注入 log record
class OTFilter(logging.Filter):
    def filter(self, record):
        span = trace.get_current_span()
        ctx = span.get_span_context()
        if ctx and ctx.trace_id != 0:
            record.trace_id = format(ctx.trace_id, '032x')
            record.span_id = format(ctx.span_id, '016x')
        else:
            record.trace_id = None
            record.span_id = None
        return True

# 3) 配置 logger 输出 JSON 到 stdout
logger = logging.getLogger("backend_api")
handler = logging.StreamHandler(sys.stdout)
fmt = jsonlogger.JsonFormatter('%(asctime)s %(levelname)s %(name)s %(message)s %(trace_id)s %(span_id)s')
handler.setFormatter(fmt)
logger.addHandler(handler)
logger.addFilter(OTFilter())
logger.setLevel(logging.INFO)

# 4) 使用 tracer 并在 span 内记录日志
def handle_request(user_id):
    with tracer.start_as_current_span("handle_request") as span:
        logger.info("开始处理请求", extra={"user_id": user_id})
        # simulate work
        with tracer.start_as_current_span("db_query"):
            logger.info("查询数据库", extra={"query": "select *"})
        logger.info("请求处理完成", extra={"status": "ok"})

if __name__ == "__main__":
    handle_request("user-123")

原理与实现要点（解释为什么这么做）

结构化日志（JSON）便于 ELK/EFK 等系统索引和查询；jsonlogger 直接输出 JSON 字段。
trace_id/span_id 不是由 logging 自带，需要从 OpenTelemetry 当前上下文获取并注入到 LogRecord；通过 logging.Filter 可以在记录创建时动态添加字段。
将 tracing 与日志分开管理，出口（Exporter）决定 trace 最终去向；ConsoleExporter 适合开发/调试，生产通常用 OTLP/Jaeger/Zipkin。

常见问题与注意事项（至少列出 5 项）

trace_id 为 0 或 None：说明当前没有活跃 span，可能没正确传播上下文。
多线程/异步问题：确保在新线程/协程中正确激活上下文（使用 ContextVar 或 SDK 提供的工具）。
性能开销：结构化日志和追踪会增加 CPU/IO，注意 sampling、异步导出与批处理。
敏感数据：不要把敏感字段（PII、密码）写入日志或 trace 标签。
日志字段不一致：保证结构化日志字段命名一致（trace_id、span_id、user_id），利于聚合。
版本兼容：OpenTelemetry 各库版本频繁变化，注意 SDK 与 exporter 的兼容性。
时间同步：日志与 trace 的时间戳应使用统一时钟（UTC），便于关联与排序。

最佳实践总结（要点式）

使用结构化 JSON 日志并统一字段名。
在日志中注入 trace_id/span_id，优先通过 Filter/Formatter 自动添加。
生产环境使用批量异步导出（OTLP）与采样策略。
在高频路径避免过度日志，利用指标与 traces 排查性能问题。
保证上下文传播在 RPC、队列、异步任务中正确传递。

结论与下一步建议把 logging 与 tracing 结合能显著提升线上问题定位效率。先从 ConsoleExporter + JSON logging 做 MVP，确认字段与查询链路后，迁移到生产级导出（OTLP 到后端如 Jaeger/Tempo），并配置采样与日志聚合管道。

可视化建议

调用链时序图：展示 service A -> B -> C 的 span 开始/结束时间线。
日志结构示意图：展示 JSON 日志字段（timestamp、level、message、trace_id、span_id、user_id）。

SEO 标签（tags）

python
logging
opentelemetry
tracing
结构化日志

长尾关键词（3 个）

python logging 注入 trace_id 实现
opentelemetry 日志与追踪结合示例
将 trace_id 写入 json 日志的方法

Meta 描述（≤150 字）通过示例演示如何在 Python 中将 OpenTelemetry 的 trace_id 注入结构化 JSON 日志，包含依赖安装、代码示例与常见注意事项，便于快速上线分布式追踪与日志关联。

引用与延伸阅读

Python logging 官方文档：https://docs.python.org/3/library/logging.html
OpenTelemetry Python 指南：https://opentelemetry.io/docs/instrumentation/python/
python-json-logger（GitHub）：https://github.com/madzak/python-json-logger
W3C Trace Context 规范：https://www.w3.org/TR/trace-context/

可复制示例代码运行环境与依赖版本（示例测试环境）

Python 3.8+
python-json-logger >= 2.0.2
opentelemetry-api >= 1.10.0
opentelemetry-sdk >= 1.10.0 安装命令： pip install python-json-logger opentelemetry-api opentelemetry-sdk opentelemetry-exporter-console

完整示例（复制粘贴运行）：

# demo.py
import logging
import sys
from pythonjsonlogger import jsonlogger

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, ConsoleSpanExporter

# Set up tracer provider and console exporter (MVP)
trace.set_tracer_provider(TracerProvider())
tracer_provider = trace.get_tracer_provider()
tracer_provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
tracer = trace.get_tracer(__name__)

# Logging Filter to inject trace_id/span_id
class OTFilter(logging.Filter):
    def filter(self, record):
        span = trace.get_current_span()
        ctx = span.get_span_context()
        # ctx.trace_id is integer; format to 32-hex string
        if ctx and ctx.trace_id != 0:
            record.trace_id = format(ctx.trace_id, '032x')
            record.span_id = format(ctx.span_id, '016x')
        else:
            record.trace_id = None
            record.span_id = None
        return True

# JSON logger config
logger = logging.getLogger("backend_api")
handler = logging.StreamHandler(sys.stdout)
# include trace_id/span_id in format so they appear in JSON
fmt = jsonlogger.JsonFormatter('%(asctime)s %(levelname)s %(name)s %(message)s %(trace_id)s %(span_id)s')
handler.setFormatter(fmt)
logger.addHandler(handler)
logger.addFilter(OTFilter())
logger.setLevel(logging.INFO)

# Example usage with nested spans
def handle_request(user_id):
    with tracer.start_as_current_span("handle_request") as span:
        logger.info("开始处理请求", extra={"user_id": user_id})
        with tracer.start_as_current_span("db_query"):
            logger.info("查询数据库", extra={"query": "select * from users where id=%s", "user_id": user_id})
        logger.info("请求处理完成", extra={"status": "ok"})

if __name__ == "__main__":
    handle_request("user-123")

运行： python demo.py

示例输出（控制台会同时打印 span 到 ConsoleExporter 与 JSON 日志到 stdout）。