本节目标：用 structlog 26.1.0 把日志输出成机器可解析的 JSON，用
contextvars绑定请求级上下文，并让trace_id贯穿从 HTTP 入口到业务代码的每一行日志。
适用版本：Python 3.12+（实测 3.14.6）；structlog 26.1.0
3.2 结构化日志与链路追踪
上一节的配置解决了「进程启动时知道什么」，本节解决「进程运行起来后发生了什么」。标准库 logging 把日志拼成一行给人看的文本，可一旦日志量上去、要按字段聚合或告警，纯文本就成了负担。我们把它升级成「每个事件是一行 JSON」，再给每个请求打上唯一 trace_id。
3.2.1 文本日志的三个天花板
标准库 logging 输出形如 2026-10-09 07:38:02,104 INFO app: 服务已启动 的文本行。这种格式对人友好，但撞上三个天花板：
| 天花板 | 后果 |
|---|---|
| 不可靠解析 | 消息里混入空格、冒号、换行，正则就会解析错位 |
| 无法按字段检索 | 想查「user_id=42 的所有请求」只能靠 grep，容易误匹配 |
| 上下文靠拼串 | 每条日志手动 f"... request_id={rid}"，漏一条就断链 |
结构化日志的做法是：把每个字段作为独立键值对输出，渲染成 JSON 后交给采集系统（Loki、ELK、Datadog）索引。structlog 是 Python 里最成熟的实现——它把「日志是一组键值对」当作一等公民。
3.2.2 最小配置：一行一条 JSON
import logging
import structlog
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.JSONRenderer(),
],
wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
logger_factory=structlog.PrintLoggerFactory(),
cache_logger_on_first_use=True,
)
log = structlog.get_logger()
log.info("user_login", user_id=42, ip="10.0.0.1")
log.debug("this_is_filtered")
{"user_id": 42, "ip": "10.0.0.1", "event": "user_login", "level": "info", "timestamp": "2026-10-09T03:32:58.874786Z"}
四个配置项各司其职：
processors：处理链，每个处理器对事件字典做一次变换，顺序决定语义。wrapper_class：绑定日志级别过滤，make_filtering_bound_logger(logging.INFO)让debug直接短路。logger_factory：决定底层往哪写，PrintLoggerFactory走print，生产常换成structlog.stdlib.LoggerFactory。cache_logger_on_first_use：第一次取 logger 时固化处理链，避免每次调用重复组装。
注意 log.info("user_login", user_id=42) 的调用形状：第一个位置参数是事件名，其余是关键字字段。这与标准库 logging.info("%s 登录", user) 的 % 风格完全不同，字段天然分离。
3.2.3 processors 链：顺序即语义
处理链按顺序执行，每个处理器拿到事件字典、返回（通常是修改后的）字典：
| 处理器 | 作用 | 位置建议 |
|---|---|---|
merge_contextvars | 把 contextvar 里的上下文并入事件 | 最前 |
add_log_level | 补 level 字段 | 中间 |
TimeStamper(fmt="iso") | 补 ISO 时间戳 | 中间 |
format_exc_info | 把异常转成字符串 | 渲染前 |
JSONRenderer | 渲染成 JSON 字符串 | 最后 |
JSONRenderer 必须是最后一个——它把字典变成字符串，之后的处理器就拿不到字段了。而 merge_contextvars 必须靠前，否则它补进来的上下文字段可能被后续逻辑覆盖。
3.2.4 上下文绑定：一次绑定，整条链路携带
给每条日志手动加 request_id 既啰嗦又易漏。structlog.contextvars 基于 contextvars，在请求开始时绑定一次，之后同一条协程链路里所有日志自动带上：
from structlog.contextvars import bind_contextvars, clear_contextvars
log = structlog.get_logger()
bind_contextvars(request_id="req-abc123", trace_id="trace-9f2c")
log.info("request_started", path="/api/orders")
log.info("db_query", table="orders", rows=3)
req_log = log.bind(user_id=7) # 只作用于这个 logger 对象
req_log.info("order_created", order_id=1001)
clear_contextvars()
log.info("after_clear")
{"path": "/api/orders", "event": "request_started", "request_id": "req-abc123", "trace_id": "trace-9f2c", "level": "info", "timestamp": "2026-10-09T03:33:09.184743Z"}
{"table": "orders", "rows": 3, "event": "db_query", "request_id": "req-abc123", "trace_id": "trace-9f2c", "level": "info", "timestamp": "2026-10-09T03:33:09.187726Z"}
{"user_id": 7, "order_id": 1001, "event": "order_created", "request_id": "req-abc123", "trace_id": "trace-9f2c", "level": "info", "timestamp": "2026-10-09T03:33:09.187995Z"}
{"event": "after_clear", "level": "info", "timestamp": "2026-10-09T03:33:09.188092Z"}
三点值得注意：request_id/trace_id 出现在前三条而最后一条没有——clear_contextvars() 清掉了上下文；bind() 返回一个新 logger，绑定的 user_id 只影响它自己，不污染全局；contextvars 是协程安全的，在 asyncio 下各请求的上下文天然隔离，不会串味。
3.2.5 与标准库 logging 桥接
现实是：你的业务代码用 structlog，但第三方库（uvicorn、httpx、SQLAlchemy）只会用标准库 logging。两套日志混在一起会一半 JSON 一半文本。用 ProcessorFormatter 把 stdlib 的记录也拉进 structlog 的处理链：
import logging
import structlog
from structlog.stdlib import ProcessorFormatter
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
ProcessorFormatter.wrap_for_formatter, # 交给下面的 formatter 渲染
],
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
formatter = ProcessorFormatter(
foreign_pre_chain=[ # 第三方库的 stdlib 记录走这里
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
],
processors=[
ProcessorFormatter.remove_processors_meta,
structlog.processors.JSONRenderer(),
],
)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
root = logging.getLogger()
root.handlers = [handler]
root.setLevel(logging.INFO)
structlog.get_logger().info("from_structlog", order_id=1001)
logging.getLogger("some.library").warning("from_stdlib %s", "warning")
{"order_id": 1001, "event": "from_structlog", "level": "info", "timestamp": "2026-10-09T03:33:28.959543Z"}
{"event": "from_stdlib warning", "level": "warning", "timestamp": "2026-10-09T03:33:28.960498Z"}
关键在 foreign_pre_chain：它只作用于「外来的」标准库记录，把它们的 levelname、msg 补齐成统一字段，再走同一套 JSONRenderer。这样整个进程——你的代码和第三方库——输出的都是同一格式的 JSON，采集端无需区分两套解析规则。
3.2.6 request_id 贯穿：一个中间件搞定
把上下文绑定放进 HTTP 中间件，就能让每个请求拥有唯一 ID，并在响应头里回传，方便用户报障时直接给出：
import uuid
from fastapi import FastAPI, Request
from structlog.contextvars import bind_contextvars, clear_contextvars
app = FastAPI()
@app.middleware("http")
async def request_context(request: Request, call_next):
rid = request.headers.get("x-request-id") or uuid.uuid4().hex[:12]
clear_contextvars() # 防止复用上一个请求的上下文
bind_contextvars(request_id=rid, path=request.url.path)
response = await call_next(request)
response.headers["x-request-id"] = rid
return response
@app.get("/orders/{order_id}")
def get_order(order_id: int):
log.info("handling_order", order_id=order_id)
log.info("order_loaded", order_id=order_id, source="db")
return {"order_id": order_id}
{"order_id": 7, "event": "handling_order", "path": "/orders/7", "request_id": "req-fixed-001", "level": "info", "timestamp": "2026-10-09T03:36:04.730744Z"}
{"order_id": 7, "source": "db", "event": "order_loaded", "path": "/orders/7", "request_id": "req-fixed-001", "level": "info", "timestamp": "2026-10-09T03:36:04.731375Z"}
响应头 x-request-id: req-fixed-001
当请求头带了 x-request-id 就用它，否则自动生成——这实现了跨服务透传：上游网关生成的 ID 一路传到下游，整条调用链的日志可以用同一个 ID 串起来。clear_contextvars() 那行很关键，否则在长驻进程里可能读到上一个请求残留的上下文。
3.2.7 与 OpenTelemetry 打通：日志跳转链路
request_id 只在单个服务内有效。要做跨服务的分布式追踪，得用 trace_id——而它应该来自 OpenTelemetry 的当前 span，而不是自己生成。写一个自定义 processor，把 span 的 ID 注入每条日志：
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer("demo")
def add_otel_ids(logger, method_name, event_dict):
"""把当前 span 的 trace_id/span_id 注入日志，实现日志↔链路互跳"""
ctx = trace.get_current_span().get_span_context()
if ctx.is_valid:
event_dict["trace_id"] = format(ctx.trace_id, "032x")
event_dict["span_id"] = format(ctx.span_id, "016x")
return event_dict
把它插进处理链（放在渲染之前），然后：
with tracer.start_as_current_span("checkout"):
log.info("payment_started", amount=99.9)
log.info("payment_done", amount=99.9)
log.info("no_span_here")
{"amount": 99.9, "event": "payment_started", "trace_id": "1953cca62d538b766c92cfef12a6cbae", "span_id": "596df3df721c3d0b", "level": "info", "timestamp": "2026-10-09T03:36:53.270647Z"}
{"amount": 99.9, "event": "payment_done", "trace_id": "1953cca62d538b766c92cfef12a6cbae", "span_id": "596df3df721c3d0b", "level": "info", "timestamp": "2026-10-09T03:36:53.277081Z"}
{"event": "no_span_here", "level": "info", "timestamp": "2026-10-09T03:36:53.277154Z"}
同一个 span 内的两条日志共享 trace_id 和 span_id，第三条在 span 之外所以没有这两个字段。有了它，日志系统里点一下 trace_id 就能跳到追踪面板看完整调用链——日志与追踪不再是两个割裂的系统。format(ctx.trace_id, "032x") 把它转成 W3C 标准的 32 位十六进制字符串，方便与其它语言的追踪系统对齐。
3.2.8 异常日志：让堆栈也结构化
异常堆栈不能直接塞进 JSON 字段（会破坏单行结构）。format_exc_info 处理器把它压成字符串，保持一行一条：
try:
{}["missing"]
except KeyError:
log.exception("读取配置失败")
{"event": "读取配置失败", "level": "error", "timestamp": "2026-10-09T03:34:12.674833Z", "exception": "Traceback (most recent call last):\n File \"app.py\", line 17, in <module>\n {}[\"missing\"]\n ~~^^^^^^^^^^^\nKeyError: 'missing'"}
（File 后的路径为便于阅读省略了目录前缀。）log.exception() 只在 except 块里用，自动抓取当前异常。如果采集端要结构化地分析堆栈帧，可换成 structlog.processors.dict_tracebacks，它把堆栈解析成数组，代价是输出体积更大（默认还会带上每一帧的 locals，生产上要评估隐私与体积）。
3.2.9 开发看颜色，生产看 JSON
同一个处理链，只换最后一个渲染器就能切换观感：
# 开发期：ConsoleRenderer 输出彩色、对齐、易读的文本
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.dev.ConsoleRenderer(),
],
wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
logger_factory=structlog.PrintLoggerFactory(),
)
log = structlog.get_logger()
bind_contextvars(trace_id="trace-9f2c")
log.error("payment_failed", reason="timeout")
2026-10-09T03:40:42.302714Z [error ] payment_failed reason=timeout trace_id=trace-9f2c
（终端里级别与键有颜色，这里去掉了 ANSI 转义；字段之间会被对齐填充空格。）生产环境换成 JSONRenderer，采集端直接索引。用环境变量（正是上一节的 Settings）在两者间切换，是最常见的落地方式：
renderer = (structlog.processors.JSONRenderer()
if settings.log_format == "json"
else structlog.dev.ConsoleRenderer())
3.2.10 日志字段的约定
结构化日志的价值取决于字段命名的一致性。团队内部约定几个通用字段，采集端的查询与仪表盘才能复用：
| 字段 | 含义 | 何时绑定 |
|---|---|---|
event | 事件名（动词短语） | 每次调用 |
level | 级别 | add_log_level |
timestamp | 时间戳 | TimeStamper |
request_id | 请求标识 | 中间件绑定 |
trace_id / span_id | 链路标识 | OTel processor 注入 |
user_id / tenant_id | 主体标识 | 鉴权后绑定 |
duration_ms | 耗时 | 操作结束时 |
事件名用 payment_started 这类「名词+动词过去式」比 "payment" 更好检索。想先看调试与日志的完整工具箱，可延伸阅读 Python 调试与日志工程
。
小结
- 结构化日志把每个事件输出成一行 JSON，字段可被采集系统索引；
structlog用「事件名 + 关键字字段」替代%风格格式化。 processors链顺序即语义：merge_contextvars靠前，JSONRenderer必须最后。structlog.contextvars在请求开始时bind_contextvars一次，同一条协程链路里所有日志自动携带上下文，clear_contextvars防止上下文泄漏。ProcessorFormatter的foreign_pre_chain把第三方库的标准库日志拉进同一处理链，全进程输出统一格式。- HTTP 中间件绑定
request_id并在响应头回传，实现跨服务透传；自定义 processor 注入 OpenTelemetry 的trace_id，让日志与追踪互相跳转。 - 异常用
log.exception()，开发期用ConsoleRenderer、生产用JSONRenderer，靠配置切换。
日志记录的是「离散事件」，而「系统整体健康度」需要连续的时间序列——下一节我们讲指标，并把健康检查与告警接进来。
阅读导航：上一节：分层配置与 pydantic-settings · 下一节：指标、健康检查与告警接入 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。