一、为什么需要全链路可观测性
线上服务就像一个大型小区，里面有很多住户。当用户报修说“我家的灯不亮了”，你不可能直接冲进每一户去查。你要拿着一张图纸，找到用户对应的那栋楼、那个单元、那间房。这个图纸就是我们的日志系统。但很遗憾，过去的日志往往是一行行文字，杂乱无章。比如某条日志写着“用户访问失败”，另一条写着“数据库连接超时”。当多个用户同时访问时，这些日志混在一起，我们根本分不清哪条日志属于哪个用户。这就像整个小区的住户都跑到楼下喊“我家出事了”，你根本听不出谁是谁。
要解决这个问题，我们需要做两件事：第一，让日志本身变得规整，也就是结构化；第二，给每个请求一个独一无二的编号，也就是请求ID。有了这两样，我们把整个调用链看清楚就有了基础。
实际上，可观测性这个概念并不神秘。它包含三个支柱：日志、指标、链路追踪。本文聚焦在日志和请求ID的配合上。当我们把日志从一行行自由文本变成一条条带字段的JSON，再给每次请求分配ID，然后让这条请求产生的所有日志都带上这个ID，那么排查问题时，只需要按ID一搜，就能把散落在各处的信息串成一条完整的线。这就是用低成本换来清晰追踪效果的办法。
下面我们动手实现。
二、准备工作：搭建一个最简单的FastAPI应用
2.1 技术栈说明
本文使用FastAPI框架，配合Python标准库的logging模块和python-json-logger库来生成结构化日志。所有示例代码基于Python 3.10环境。如果你有基本的Python开发经验，应该能很快跟上。
2.2 创建项目文件
我们先写一个最简单的FastAPI应用。创建一个main.py文件，内容如下：
# 技术栈：Python 3.10 + FastAPI + python-json-logger
# 文件：main.py
from fastapi import FastAPI
# 创建FastAPI实例，这是整个服务的入口
app = FastAPI()
# 定义根路径接口，用户访问时返回一句问候
@app.get("/")
def read_root():
return {"message": "hello world"}
# 定义带参数的接口，模拟真正的业务逻辑
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
# 直接使用uvicorn启动服务，这样我们不需要额外敲命令
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
这里有两个接口。现在运行这个文件，服务会启动，但日志还是默认的格式。为了让日志结构化，我们需要对日志系统进行配置。
2.3 项目结构建议
随着代码变多，建议把日志配置独立到一个文件里，比如logger_config.py。这样主程序保持干净，日志逻辑可以单独维护。接下来我们就来写这个配置。
三、结构化日志：让日志变成一列列数据
3.1 为什么要结构化
普通文本日志像手写的便签，虽然记了内容，但很难批量处理。JSON格式的日志像一份表格，每一列都有字段名，机器能读，人也能读。它能被日志平台直接解析，也方便在命令行里用jq等工具查询。
3.2 安装python-json-logger
在本文的示例中，我们需要用到一个第三方库python-json-logger。它能把标准库logging输出的日志格式化成JSON。你可以在终端中用pip安装它，这里就不展示安装命令了，毕竟那只依赖你的Python环境。
3.3 创建日志配置
在logger_config.py文件中，我们定义一个安装日志的函数。先看代码：
# 文件：logger_config.py
# 技术栈：Python 3.10 + FastAPI + python-json-logger
import logging
import sys
from pythonjsonlogger.json import JsonFormatter
def setup_logging():
# 创建一个控制台处理器，把日志输出到标准输出
handler = logging.StreamHandler(sys.stdout)
# 使用JSON格式器，并且把标准字段名转换成更友好的名字
formatter = JsonFormatter(
fmt="%(asctime)s %(levelname)s %(name)s %(message)s",
rename_fields={
"asctime": "time", # 时间字段改为time
"levelname": "level", # 级别字段改为level
},
)
handler.setFormatter(formatter)
# 获取根日志器，挂上处理器并设置日志级别
root_logger = logging.getLogger()
root_logger.addHandler(handler)
root_logger.setLevel(logging.INFO)
在应用启动时调用setup_logging()，日志就会变成JSON格式。比如我们在一个接口里打印一条消息：
import logging
# 记录一条INFO级别的测试日志
logging.info("这是一条测试日志")
输出的日志会类似下面这样：
{"time": "2025-03-21 10:00:00,123", "level": "INFO", "name": "logger_config", "message": "这是一条测试日志"}
注意，这里的输出里包含了time、level、name、message四个字段。这就是结构化日志的最小形态。在此基础上，我们还可以继续添加更多自定义字段。
四、请求ID：给每个请求发一个唯一编号
4.1 什么是请求ID
请求ID就是一次HTTP请求的唯一标识。用户可以不了解，但后端必须知道。当用户遇到问题时，我们把响应头里的X-Request-ID告诉用户，然后运维在日志系统里一搜，所有相关日志都出来了。
4.2 如何生成请求ID
FastAPI中可以使用中间件来拦截请求。中间件可以理解成一道安检门，请求进来时先过一遍，响应出去时再过一遍。我们在这道门里生成一个UUID，把它存到Python的上下文变量中。
4.3 上下文变量的作用
Python标准库的contextvars模块提供了一种在异步任务中传递状态的机制。它类似于给每个任务发一个小口袋，任务内部任何地方都能掏出这个口袋里的东西，任务之间互不干扰。这正好用来存放请求ID。
我们来看生成请求ID的中间件代码：
# 文件：request_id.py
# 技术栈：Python 3.10 + FastAPI
import uuid
from contextvars import ContextVar
from starlette.middleware.base import BaseHTTPMiddleware
# 声明一个上下文变量，默认值是短横线，表示还没有请求ID
request_id_var: ContextVar[str] = ContextVar("request_id", default="-")
class RequestIDMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
# 为当前请求生成一个全局唯一的ID
request_id = str(uuid.uuid4())
# 把ID放入上下文变量
request_id_var.set(request_id)
# 继续执行后续的路由处理
response = await call_next(request)
# 把ID放进响应头，方便前端或用户回传给我们
response.headers["X-Request-ID"] = request_id
return response
4.4 把中间件注册到应用
在main.py中添加中间件注册：
# 在main.py中导入并注册中间件
from request_id import RequestIDMiddleware
app.add_middleware(RequestIDMiddleware)
五、让请求ID自动出现在所有日志中
5.1 日志过滤器
logging模块提供了Filter类，用来在日志输出前对日志记录做手脚。我们可以创建一个过滤器，从上下文变量中取出请求ID，把它塞进日志记录里，这样JSON格式器就能把它输出出来。
代码：
# 文件：logger_config.py（在原有基础上增加Filter）
# 技术栈：Python 3.10 + FastAPI + python-json-logger
import logging
import sys
from pythonjsonlogger.json import JsonFormatter
from request_id import request_id_var
class RequestIDFilter(logging.Filter):
"""把请求ID加到每一条日志上"""
def filter(self, record: logging.LogRecord) -> bool:
# 从上下文变量里取出当前请求ID，存到日志记录里
record.request_id = request_id_var.get()
# 返回True表示日志继续正常处理
return True
def setup_logging():
handler = logging.StreamHandler(sys.stdout)
formatter = JsonFormatter(
fmt="%(asctime)s %(levelname)s %(name)s %(message)s",
rename_fields={
"asctime": "time",
"levelname": "level",
},
)
handler.setFormatter(formatter)
# 把这个过滤器添加到处理器上
handler.addFilter(RequestIDFilter())
root_logger = logging.getLogger()
root_logger.addHandler(handler)
root_logger.setLevel(logging.INFO)
这样，当我们在路由函数中打印日志时，每一条日志都会带上request_id字段。比如我们在某个接口里写：
import logging
# 在某个接口里记录日志
logging.info("用户查询商品的请求已收到")
输出就会变成：
{"time": "2025-03-21 10:00:00,123", "level": "INFO", "name": "main", "message": "用户查询商品的请求已收到", "request_id": "6f9b2e5a-34d1-4f1a-9a5b-1c8e4d7f2a66"}
注意，这个request_id是从上下文变量中取出来的，和当前请求完全绑定。
下面我们把所有部分拼起来，得到一个完整的main.py：
# 文件：main.py（完整示例）
# 技术栈：Python 3.10 + FastAPI + python-json-logger
import logging
from fastapi import FastAPI
from logger_config import setup_logging
from request_id import RequestIDMiddleware
# 启动时配置日志
setup_logging()
app = FastAPI()
# 注册请求ID中间件
app.add_middleware(RequestIDMiddleware)
@app.get("/")
def read_root():
# 打印一条普通日志
logging.info("根接口被访问")
return {"message": "hello world"}
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
# 打印一条业务日志
logging.info("查询商品信息")
return {"item_id": item_id, "q": q}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
5.3 为什么不用日志参数传递
有的同学可能想，直接在每行日志里把请求ID作为参数传进去，不就行了吗？那样虽然可行，但太麻烦。你每写一条日志都要手动带上ID，容易漏，也容易写错。用Filter自动注入，能确保所有日志都规规矩矩带上ID，而且不需要改动业务代码。这就是自动化带来的好处。
六、收集更多关键信息：路径、状态码、耗时
仅有请求ID还不够，我们经常还要知道这次请求访问了哪个路径、返回了什么状态码、花了多长时间。这些信息可以帮助我们快速定位性能问题和访问异常。
6.1 在中间件中记录耗时
我们可以在中间件里记录开始时间，在响应返回前计算耗时，然后把相关信息写进一条日志。注意，我们可以顺便把请求方法、路径、状态码都记录下来。
扩展RequestIDMiddleware如下：
# 文件：request_id.py（增强版）
# 技术栈：Python 3.10 + FastAPI + python-json-logger
import logging
import time
import uuid
from contextvars import ContextVar
from starlette.middleware.base import BaseHTTPMiddleware
request_id_var: ContextVar[str] = ContextVar("request_id", default="-")
class RequestIDMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
# 生成请求ID
request_id = str(uuid.uuid4())
request_id_var.set(request_id)
# 记录开始处理的时间
start_time = time.perf_counter()
try:
response = await call_next(request)
except Exception:
# 出异常时也要记录日志，并带上请求ID
logging.exception("请求处理异常")
raise
# 计算处理耗时（单位：毫秒）
duration_ms = round((time.perf_counter() - start_time) * 1000, 2)
# 把请求ID写入响应头
response.headers["X-Request-ID"] = request_id
# 记录一条访问日志，包含关键信息
logging.info(
"访问日志",
extra={
"method": request.method,
"path": request.url.path,
"status_code": response.status_code,
"duration_ms": duration_ms,
},
)
return response
这样，每次请求结束后，日志系统都会自动输出一条包含关键信息的JSON日志。请求ID由之前挂载的RequestIDFilter自动注入，不需要手动写。输出会类似：
{"time": "...", "level": "INFO", "name": "request_id", "message": "访问日志", "request_id": "abc-123", "method": "GET", "path": "/items/1", "status_code": 200, "duration_ms": 13.45}
6.2 业务日志中也可以包含更多字段
比如用户ID、订单号、商品ID等，都可以通过日志的extra参数来补充。但要注意，不要让日志字段无限膨胀，字段太多反而会增加维护成本。我们只需要记录对排查问题最有用的信息即可。
七、应用场景、优缺点分析
7.1 应用场景
最适合这种方案的应用场景有这些：
第一，面向用户的线上服务。当用户遇到问题，我们只需要拿到他的请求ID，就能把他那次请求的完整日志链捞出来，从入口到数据库，所有环节都能看到。
第二，微服务架构。虽然本文只演示了单个FastAPI服务，但请求ID的思想可以横向扩展。每个服务都生成自己的请求ID，或者在网关层生成一个统一的追踪ID，然后透传到下游服务，就能实现跨服务的完整链路追踪。
第三，数据分析与监控。结构化的JSON日志可以直接被日志平台解析，然后根据字段进行统计，比如按状态码统计请求量，按耗时计算P95等。
7.2 优点
这种做法的优点非常明显。首先，日志可读性和可查询性大大提升，人眼能看懂，脚本也能处理。其次，因为每条日志都有请求ID，排查问题时不再需要猜测日志之间的关联。再者，结构化日志是自动化的基础，后面接上日志采集、告警、可视化管理平台，都非常顺畅。
7.3 缺点
当然，任何方案都有代价。第一，日志量变大。JSON格式比纯文本长得多，存储成本会上升。第二，加中间件、加过滤器，多少会影响一点性能，尤其在高并发场景下。第三，如果团队没有统一的日志规范，每个人往日志里塞不同的字段，时间一长会变得混乱。所以我们在引入结构化日志时，最好提前定义好字段标准，并且控制日志输出量。
八、注意事项
8.1 中间件的执行顺序
在FastAPI中添加中间件时，注意它的执行顺序是后添加的先执行。如果我们有多个中间件，要确保请求ID中间件在最外层，这样请求ID才能覆盖内部所有环节。
8.2 异常情况下的日志输出
如果请求在处理过程中抛出了异常，中间件里的call_next会抛出异常，此时我们可能拿不到response。我们需要在中间件中捕获异常，并记录一条包含请求ID和异常信息的日志。否则异常日志容易丢失请求ID。增强版中间件里已经用try-except处理了这种情况，关键代码是这样的：
try:
response = await call_next(request)
except Exception:
logging.exception("请求处理异常")
raise
注意这里的request_id依然会被自动注入，因为上下文变量在异常发生时还没有被清除。
8.3 日志脱敏
永远不要在日志中记录密码、Token、银行卡号等敏感信息。如果一定要记录某些用户数据，先做脱敏处理，比如手机号中间四位用星号代替。
8.4 不要过度记录
每个请求都打日志虽然方便，但如果日志量过大，会影响系统性能。可以根据需要调整日志级别。比如生产环境只记录WARNING及以上，或者采样记录一批请求。也可以使用异步日志处理器来减轻IO压力。
8.5 与分布式追踪系统的关系
请求ID是最朴素的追踪方案，但它只适合单服务或手工透传的场景。如果系统很复杂，建议使用专业的分布式追踪系统，例如OpenTelemetry。它会自动生成trace ID和span ID，并且能跨服务传递上下文。我们可以把请求ID当作trace ID的雏形，理解它的原理有助于后面学习更复杂的工具。
九、总结
回到最初的问题：日志杂乱无章，排查问题痛苦。我们通过两步来解决。第一步，用JSON格式器把日志变成结构化数据；第二步，用中间件为每次请求生成唯一ID，并通过上下文变量和日志过滤器，让这个ID自动出现在每条日志里。同时，我们还可以在中间件里补充路径、状态码、耗时等信息，让一条日志变得更有价值。
这套方案不需要引入额外的大件系统，只用了FastAPI和Python标准库，以及一个小小的python-json-logger。但它已经能解决大多数中小型服务的日志追踪问题。当你觉得服务越来越复杂、日志查询越来越难时，不妨先把这个基础打好，再考虑引入更完整的可观测性平台。希望这篇文章能给你带来启发，让你在排查问题时不再像无头苍蝇一样乱翻日志。
Comments