本节目标：把 structlog 输出的 JSON 日志接进「采集 → 存储 → 查询 → 告警」链路，理解日志级别与采样策略，写出可降噪的告警规则，并让日志通过
trace_id关联到追踪。
适用版本：Python 3.12+（实测 3.14.6）；structlog 26.1.0
17.2 日志聚合与告警
上一节的追踪能告诉我们「一次请求慢在哪一段」，但要回答「为什么报错、什么时候开始报」，仍然得回到日志。第 3.2 节讲过怎么把日志写成 JSON，本节往前走一步——这些 JSON 到底怎么被采集、存储、查询，并最终变成一条会叫醒人的告警。
17.2.1 单机日志的三堵墙
服务一旦多实例部署，tail -f app.log 就彻底失效：
| 墙 | 现象 |
|---|---|
| 分散 | 请求可能落到任意实例，日志散在 N 台机器上 |
| 易失 | 容器一销毁，日志随之蒸发 |
| 不可查 | 想统计「过去 1 小时 5xx 有多少」只能逐台 grep |
日志聚合要解决的就是这三件事：把分散的日志汇到一处、持久化、并支持按字段检索。前提是日志必须结构化——非结构化的文本无法被高效索引，这也是 3.2 节先讲 JSON 的原因。
17.2.2 结构化输出：让每条日志带 trace_id
聚合系统的查询能力取决于字段。下面这份配置输出 JSON，并把当前 span 的 trace_id/span_id 注入每条日志，实现日志与追踪互跳：
import logging
import structlog
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
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
add_otel_ids,
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()
with tracer.start_as_current_span("checkout"):
log.info("order_created", order_id=1001, amount=99.9)
log.warning("cache_miss", key="order:1001")
{"order_id": 1001, "amount": 99.9, "event": "order_created", "level": "info", "timestamp": "2026-10-09T04:00:48.153404Z", "trace_id": "fd50db94bf245935215fa0651861f016", "span_id": "782e9101fdf0ba35"}
{"key": "order:1001", "event": "cache_miss", "level": "warning", "timestamp": "2026-10-09T04:00:48.154266Z"}
第一条日志落在 span 内，带上了 trace_id；第二条在 span 之外，没有这两个字段。有了 trace_id，在日志系统里点一下就能跳到 17.1 的追踪面板看完整调用链——日志与追踪不再是两个割裂的系统。
17.2.3 聚合链路：采集 → 存储 → 查询
JSON 日志只是一行文本流。要变成可查询的数据，需要一条四段式链路：
| 阶段 | 做什么 | 典型组件 |
|---|---|---|
| 采集 | 从 stdout/文件/容器收集并转发 | Promtail、Fluent Bit、Filebeat |
| 存储 | 建立倒排索引，持久化 | Loki、Elasticsearch、ClickHouse |
| 查询 | 按标签与字段检索、聚合 | LogQL、KQL、SQL |
| 告警 | 命中规则时触发通知 | Alertmanager、Grafana |
两个流派值得区分：全文索引派（ELK）把整条日志分词建索引，查询灵活但存储成本高；标签 + 扫描派（Loki）只为标签建索引、日志体压缩存储，查询时扫描，成本低很多。日志量大时，Loki 这种「重标签、轻索引」的方案更划算——前提还是字段命名要规范。
本地没有采集器，但我们用纯 Python 模拟「采集 → 聚合 → 查询」这三步，逻辑与真实系统一致：
import json, io, logging, structlog
from collections import Counter
buf = io.StringIO()
structlog.configure(
processors=[
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(file=buf), # 模拟写入日志文件/stdout 管道
)
log = structlog.get_logger()
log.info("order_created", order_id=1)
log.warning("cache_miss", key="order:1")
log.error("payment_failed", order_id=2, reason="timeout")
lines = [json.loads(l) for l in buf.getvalue().splitlines()] # 采集：逐行解析
print("采集到日志行数:", len(lines))
print("按级别聚合:", dict(Counter(r["level"] for r in lines)))
hits = [r for r in lines if r["level"] == "error" and r["event"] == "payment_failed"]
print("查询结果:", hits) # 查询：按字段过滤
采集到日志行数: 3
按级别聚合: {'info': 1, 'warning': 1, 'error': 1}
查询结果: [{'order_id': 2, 'reason': 'timeout', 'event': 'payment_failed', 'level': 'error', 'timestamp': '2026-10-09T04:02:03.155189Z'}]
「按级别聚合」就是采集端建索引后仪表盘上的柱状图，「按字段过滤」等价于 Loki 的 {level="error"} | json | event="payment_failed"。只要日志是 JSON，任何后端都能用同一套字段查询，换后端不用改一行应用代码。
17.2.4 把标准库日志也拉进同一条链
现实是：业务代码用 structlog，但 uvicorn、httpx、SQLAlchemy 只会用标准库 logging。两套格式混在一起，采集端就得维护两套解析规则。用 ProcessorFormatter 把第三方库的日志也渲染成同样的 JSON：
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,
],
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
)
formatter = ProcessorFormatter(
foreign_pre_chain=[
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("app_started", version="1.4.0")
logging.getLogger("uvicorn.error").info("Uvicorn running on http://127.0.0.1:8000")
{"version": "1.4.0", "event": "app_started", "level": "info", "timestamp": "2026-10-09T04:02:17.556962Z"}
{"event": "Uvicorn running on http://127.0.0.1:8000", "level": "info", "timestamp": "2026-10-09T04:02:17.558075Z"}
foreign_pre_chain 只作用于「外来的」标准库记录，把它们补齐成统一字段后走同一个渲染器。全进程输出同一种格式，采集端只需一套解析规则。
17.2.5 日志级别与采样：控成本、降噪音
日志的成本和噪音随量线性增长，两个手段缺一不可。
级别决定「哪些日志被写出来」。生产环境通常设 INFO，DEBUG 只在排障时临时开。级别是编译期短路——make_filtering_bound_logger(logging.INFO) 让 debug() 直接返回，连处理器都不进。
采样决定「同类日志留多少」。像「缓存探测」这种高频、低信息量的日志，全量写会淹没真正重要的信号。写一个按比例丢弃的 processor：
import random, structlog
def sample_debug(sample_rate: float):
def processor(logger, method_name, event_dict):
if method_name == "debug" and random.random() > sample_rate:
raise structlog.DropEvent # 抛出即丢弃这条日志
return event_dict
return processor
structlog.configure(
processors=[
structlog.processors.add_log_level,
sample_debug(0.1), # 只保留约 10% 的 debug
structlog.processors.JSONRenderer(),
],
wrapper_class=structlog.make_filtering_bound_logger(0), # 0 表示允许 debug
logger_factory=structlog.PrintLoggerFactory(),
)
log = structlog.get_logger()
for i in range(1000):
log.debug("cache_probe", i=i)
实际调用 1000 次 debug，采样后输出 108 行（约 10.8%）
structlog.DropEvent 是 structlog 提供的「丢弃信号」异常，抛出后该条日志被静默丢弃，不影响业务代码。实测 1000 次调用留下 108 行，接近设定的 10%。采样策略要小心：错误日志绝不能采样，否则会漏掉关键故障——只对 DEBUG/TRACE 这类高频低价值日志采样。
17.2.6 告警规则：表达式与降噪
日志进聚合系统后，就该让它「主动喊人」。下面是一条 Loki 的 LogQL 告警规则（本机无 Loki，未实测，仅示意）：
groups:
- name: orders-api-logs
rules:
- alert: PaymentErrorsSpiking
expr: |
sum(count_over_time({app="orders-api"} |= "payment_failed" [5m])) > 20
for: 5m
labels:
severity: critical
annotations:
summary: "支付失败日志 5 分钟内超过 20 条"
runbook: "https://runbook.example.com/payment-errors"
count_over_time(...[5m]) 统计时间窗口内的行数，|= 是「包含子串」过滤，for: 5m 要求异常持续 5 分钟才触发。告警设计的核心是降噪：
| 噪音来源 | 降噪手段 |
|---|---|
| 瞬时毛刺 | for 持续时长（如 5m） |
| 重复告警 | 按 group_by 分组聚合，或静默窗口 |
| 低价值告警 | 对症状（用户可见）而非原因告警 |
| 无头苍蝇 | 附 runbook 链接，写清第一步做什么 |
| 阈值拍脑袋 | 基于 SLO 错误预算定阈值 |
一个高频反模式是「对每一条 error 日志告警」。一条 error 不等于一次故障——可能只是某个用户输错了参数。告警应该对「错误率 / 错误量超过阈值」这类症状告警，而不是对单条日志。这正是 3.3 节指标告警的补充：指标看趋势，日志看细节。
17.2.7 采集端配置样例（未实测）
应用侧只要把 JSON 打到 stdout，剩下的交给采集器。下面是 Promtail 从容器 stdout 抓取并送 Loki 的配置（本机无采集器，未实测，仅示意）：
scrape_configs:
- job_name: orders-api
static_configs:
- targets: [localhost]
labels:
app: orders-api
__path__: /var/log/containers/orders-api-*.log
pipeline_stages:
- json:
expressions:
level: level
trace_id: trace_id
- labels:
level: # 把 JSON 里的 level 提升为可过滤标签
json 阶段解析出字段，labels 阶段把 level 提升成索引标签——这样 {app="orders-api", level="error"} 就能只扫错误日志。注意只把低基数、高频过滤的字段提升为标签（level、app），高基数的 trace_id、order_id 留在日志体里按需扫描，否则索引会膨胀。
17.2.8 三支柱闭环
把 17.1 和本节串起来，一个完整的排障闭环是：
- 指标告警触发——错误率或支付失败量超阈值。
- 看日志——按
trace_id或字段过滤，找到具体报错事件。 - 看追踪——点开
trace_id，看这次请求慢/错在哪一段。
三者的黏合剂就是 trace_id：它同时出现在日志字段和追踪里，才能一键互跳。这也是为什么本节要让日志 processor 主动注入 span 的 trace_id。想更系统地了解日志工程，可延伸阅读 Python 调试与日志工程
。
小结
- 日志聚合解决多实例下的「分散、易失、不可查」，前提是日志结构化为 JSON。
- 链路四段：采集（Promtail/Fluent Bit）→ 存储（Loki/ES）→ 查询（LogQL/KQL）→ 告警（Alertmanager/Grafana）；Loki 重标签轻索引，成本更低。
ProcessorFormatter.foreign_pre_chain把第三方库的标准库日志统一成同一 JSON 格式，采集端只需一套解析规则。- 级别是编译期短路，采样用
structlog.DropEvent按比例丢弃；只对 DEBUG/TRACE 采样，错误日志绝不采样。 - 告警对「错误率/错误量」等症状而非单条日志，用
for持续时长、分组聚合、runbook 链接降噪。 - 采集端只把低基数、高频过滤的字段（
level、app）提升为标签，高基数 ID 留在日志体里。
日志和追踪解决了「系统跑起来后看得见」，但「怎么安全地把新版本推上去、出事怎么退回来」是另一套工程。下一节讲灰度发布、回滚与故障演练。
阅读导航：上一节：OpenTelemetry 追踪 · 下一节：灰度发布、回滚与故障演练 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。