我要评论摘要
一个接口返回 500，并不等于开发者已经知道哪里出了问题。请求来自哪个用户、传了什么业务参数、执行到哪一层失败、是否调用了下游服务、同一条请求在多个服务之间经历了什么，这些信息需要靠一套可检索、可关联的可观测性设计来回答。
日志、异常处理和链路追踪是 python 后端走向生产环境的三项基础能力：日志记录事实，异常处理稳定地表达失败，链路追踪把一次跨服务请求串成完整路径。本文以 fastapi 为例，完成 request_id 上下文、json 结构化日志、统一错误码、下游请求透传与 opentelemetry 追踪的最小闭环，并讨论脱敏、采样、异步任务和告警等生产实践。
读完本文后，你应该能够：
- 设计适合线上检索的结构化日志；
- 区分业务异常、参数异常、依赖异常和系统异常；
- 为每个请求注入并贯穿 request_id；
- 在 fastapi 中实现统一异常响应；
- 向 http 下游服务透传关联标识；
- 理解 trace、span、日志与指标如何配合排障；
- 避免敏感信息泄露、重复报错和日志风暴。
一、背景与问题
1. 线上排障真正缺少的是什么
开发环境里，print() 一句变量，浏览器重试一次接口，通常就能定位问题。生产环境却不同：请求并发到来，应用往往运行多个副本；一次下单可能经过网关、订单、库存、支付等服务；错误还可能只在特定用户、特定时间或某个下游超时时出现。
如果日志只是下面这样：
error: order failed
它几乎无法回答任何问题。一个有价值的错误记录至少应让我们知道：什么时候、哪台实例、哪个请求、哪个接口、哪个用户或订单（经过脱敏或使用内部 id）、错误类型、耗时和异常堆栈。
2. 三类常见的失败设计
第一类是日志散落。路由、service、dao 都各自 logging.info，字段格式不统一，检索时无法按请求聚合。第二类是异常泄漏。把数据库异常、堆栈和内部 sql 原样返回给客户端，既不安全，也让调用方无法依据稳定错误码处理。第三类是跨服务断链。订单服务记录了“调用支付失败”，支付服务却找不到同一次调用，排查只能靠时间范围猜测。
3. 可观测性不是“多打点日志”
生产系统通常同时需要日志、指标和链路追踪：
| 能力 | 主要回答的问题 | 典型载体 |
|---|---|---|
| 日志（logs） | 当时具体发生了什么？ | 请求字段、错误堆栈、业务事件 |
| 指标（metrics） | 系统整体是否健康？ | qps、p95 延迟、错误率、队列积压 |
| 链路追踪（traces） | 这次请求经过了哪里、慢在哪里？ | trace、span、调用依赖 |
三者并非替代关系。指标负责快速发现“错误率升高”，追踪帮助定位“慢在库存服务”，日志则提供“库存服务拒绝了哪个业务条件”的细节。
二、核心概念
1. 日志级别与使用边界
python 标准库 logging 提供多个级别。它们不是语气强弱，而是后续告警、筛选和存储策略的依据。
| 级别 | 适合记录 | 是否通常在线上开启 |
|---|---|---|
| debug | 调试路径、局部变量、开发诊断 | 否，短时按需开启 |
| info | 服务启动、请求完成、关键业务事件 | 是 |
| warning | 可恢复异常、配置降级、即将耗尽的资源 | 是 |
| error | 当前操作失败且需要关注 | 是 |
| critical | 服务不可用、数据风险等紧急故障 | 是，并应触发高优告警 |
不要把一切都写成 error。验证码输错、库存不足等预期业务结果，通常返回明确的业务错误码即可；只有系统无法按预期完成操作时才需要 error 级别日志。
2. 结构化日志
结构化日志把每条记录写成固定字段，而不是拼接自然语言。json 是最常见格式，便于 elasticsearch、loki、clickhouse 等平台建立索引和聚合。
{
"timestamp": "2026-09-16t10:24:11.625+08:00",
"level": "info",
"event": "request_completed",
"request_id": "e3ee2d4c8df84c1c",
"method": "post",
"path": "/api/orders",
"status_code": 201,
"duration_ms": 38.4
}其中 event 应描述稳定事件名，例如 request_completed、payment_timeout，而把会变化的信息放在独立字段中。这样可以按 event=payment_timeout 聚合，而不是依赖模糊文本搜索。
3. request_id、trace_id 与 span_id
request_id 是应用层请求编号，通常由入口服务生成并通过 x-request-id 传递。它实现简单，适合日志检索。
分布式追踪使用更标准的层次：一次端到端请求对应一个 trace_id；每次 http、数据库或内部函数调用可形成一个 span，并拥有自己的 span_id。下游 span 通过父子关系构成调用树。w3c trace context 约定用 traceparent 请求头传递这些信息。
客户端
-> api 网关 [trace_id: 4bf..., span: a1]
-> 订单服务 [trace_id: 4bf..., span: b2]
-> 库存服务 [trace_id: 4bf..., span: c3]
-> 支付服务 [trace_id: 4bf..., span: d4]
实践中可同时保留两者：业务人员用 request_id 快速查日志，运维平台依靠 trace_id 观察跨服务耗时。
4. 异常与错误码
异常是程序内部的控制流，错误码是对外契约。不要把某个库的 integrityerror、keyerror 直接交给 api 客户端，而应先归类：
| 分类 | 示例 | http 状态码 | 对外信息 |
|---|---|---|---|
| 参数错误 | 字段缺失、格式错误 | 422 / 400 | 指出可修正字段 |
| 认证授权 | token 无效、无权访问 | 401 / 403 | 不暴露鉴权实现 |
| 业务错误 | 库存不足、订单状态不允许取消 | 409 / 400 | 稳定业务错误码 |
| 依赖错误 | redis、支付服务超时 | 503 / 502 | 请稍后重试 |
| 未知系统错误 | 代码缺陷、未预期异常 | 500 | 通用错误和 request_id |
三、工作原理
1. 从请求进入到响应返回
一条 http 请求的观测闭环可以设计为：
请求进入 -> 中间件读取或生成 request_id -> 写入 contextvar 日志上下文 -> 记录 request_started -> 路由 / service / repository 执行业务 -> 下游 http 请求透传 x-request-id 与 traceparent -> 正常响应：记录 request_completed（状态码、耗时） -> 已知业务异常：统一转换为稳定错误响应 -> 未知异常：记录完整堆栈，返回通用 500 + request_id -> 清理 contextvar
中间件适合处理与业务无关的横切逻辑。把 request_id 放在 contextvar 中，service 层无需层层传递参数，也能让日志 filter 自动补充该字段。对于 python 的 asyncio 任务，contextvar 会随当前任务上下文传播；但提交到 celery、线程池或独立进程时，仍需显式传递。
2. 为什么要统一处理异常
统一异常处理将“记录什么”和“返回什么”分开。业务异常的调用方可以根据 code 做下一步操作；未知异常只向服务器日志记录堆栈，客户端得到不泄漏内部细节的响应。这样也避免每个路由重复写 try/except。
一个建议的错误响应如下：
{
"code": "order_not_cancellable",
"message": "当前订单状态不允许取消",
"request_id": "e3ee2d4c8df84c1c"
}message 可以面向用户，code 必须稳定、可被程序处理；request_id 则让用户反馈问题时可以被迅速定位。
3. 关联日志与追踪
理想状态下，每条应用日志都附带 request_id、trace_id 与 span_id。告警从指标中发现后，先跳到异常 trace，再由 trace 中的 id 跳转到对应日志。即使暂未部署完整追踪平台，仅通过 request_id 也能完成第一阶段的请求级检索。
四、实战示例
下面构建一个小型订单 api。示例重点是可观测性结构，数据库操作以简单的内存数据代替。
1. 安装依赖
pip install fastapi "uvicorn[standard]" httpx python-json-logger pip install opentelemetry-api opentelemetry-sdk pip install opentelemetry-instrumentation-fastapi opentelemetry-instrumentation-httpx
2. 用 contextvar 保存请求上下文
创建 observability.py：
from __future__ import annotations
import contextvars
import json
import logging
import sys
from datetime import datetime, timezone
request_id_var: contextvars.contextvar[str] = contextvars.contextvar(
"request_id", default="-"
)
class contextfilter(logging.filter):
def filter(self, record: logging.logrecord) -> bool:
record.request_id = request_id_var.get()
return true
class jsonformatter(logging.formatter):
"""只保留稳定且易查询的基础字段；业务字段使用 extra 添加。"""
def format(self, record: logging.logrecord) -> str:
payload = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getmessage(),
"request_id": getattr(record, "request_id", "-"),
}
for key in ("event", "method", "path", "status_code", "duration_ms", "code"):
value = getattr(record, key, none)
if value is not none:
payload[key] = value
if record.exc_info:
payload["exception"] = self.formatexception(record.exc_info)
return json.dumps(payload, ensure_ascii=false, default=str)
def configure_logging() -> none:
handler = logging.streamhandler(sys.stdout)
handler.addfilter(contextfilter())
handler.setformatter(jsonformatter())
root = logging.getlogger()
root.handlers.clear()
root.addhandler(handler)
root.setlevel(logging.info)
logger = logging.getlogger("app")
这里没有把用户邮箱、token、完整请求体放入通用日志字段。日志往往会被集中采集并长期保留，默认收集越少，泄露风险越低。
3. 定义业务异常与统一响应
创建 errors.py：
from dataclasses import dataclass
@dataclass
class apperror(exception):
code: str
message: str
status_code: int = 400
log_level: str = "warning"
class inventorynotenough(apperror):
def __init__(self, sku_id: str) -> none:
super().__init__(
code="inventory_not_enough",
message="库存不足，请调整购买数量",
status_code=409,
)
再创建 main.py：
from __future__ import annotations
import logging
import time
import uuid
from contextlib import asynccontextmanager
from fastapi import fastapi, request
from fastapi.exceptions import requestvalidationerror
from fastapi.responses import jsonresponse
from pydantic import basemodel, field
from errors import apperror, inventorynotenough
from observability import configure_logging, logger, request_id_var
class createorderrequest(basemodel):
sku_id: str = field(min_length=1, max_length=64)
count: int = field(gt=0, le=10)
@asynccontextmanager
async def lifespan(app: fastapi):
configure_logging()
logger.info("service started", extra={"event": "service_started"})
yield
logger.info("service stopped", extra={"event": "service_stopped"})
app = fastapi(lifespan=lifespan)
@app.middleware("http")
async def request_logging_middleware(request: request, call_next):
request_id = request.headers.get("x-request-id") or uuid.uuid4().hex
token = request_id_var.set(request_id)
started = time.perf_counter()
try:
response = await call_next(request)
duration_ms = round((time.perf_counter() - started) * 1000, 2)
response.headers["x-request-id"] = request_id
logger.info(
"request completed",
extra={
"event": "request_completed",
"method": request.method,
"path": request.url.path,
"status_code": response.status_code,
"duration_ms": duration_ms,
},
)
return response
except exception:
# 未被异常处理器接住的中间件级异常，必须保留堆栈。
logger.exception(
"request crashed",
extra={"event": "request_crashed", "method": request.method, "path": request.url.path},
)
raise
finally:
request_id_var.reset(token)
@app.exception_handler(apperror)
async def app_error_handler(_: request, exc: apperror):
log = getattr(logger, exc.log_level, logger.warning)
log(
"business request rejected",
extra={"event": "business_error", "code": exc.code},
)
return jsonresponse(
status_code=exc.status_code,
content={"code": exc.code, "message": exc.message, "request_id": request_id_var.get()},
)
@app.exception_handler(requestvalidationerror)
async def validation_error_handler(_: request, exc: requestvalidationerror):
# errors() 仅保留字段和规则，不回显整个原始请求体。
logger.warning("invalid request", extra={"event": "validation_error", "code": "validation_error"})
return jsonresponse(
status_code=422,
content={
"code": "validation_error",
"message": "请求参数不符合要求",
"details": exc.errors(),
"request_id": request_id_var.get(),
},
)
@app.exception_handler(exception)
async def unhandled_error_handler(_: request, exc: exception):
logger.exception("unhandled server error", extra={"event": "unhandled_error", "code": "internal_error"})
return jsonresponse(
status_code=500,
content={
"code": "internal_error",
"message": "服务暂时不可用，请稍后重试",
"request_id": request_id_var.get(),
},
)
@app.post("/api/orders", status_code=201)
async def create_order(payload: createorderrequest):
if payload.sku_id == "sold-out":
raise inventorynotenough(payload.sku_id)
logger.info("order created", extra={"event": "order_created"})
return {"order_id": uuid.uuid4().hex, "status": "created"}
运行服务：
uvicorn main:app --reload --port 8000
请求库存不足的商品：
curl -i -x post http://127.0.0.1:8000/api/orders \
-h "content-type: application/json" \
-h "x-request-id: demo-order-001" \
-d '{"sku_id":"sold-out","count":1}'
响应会包含可反馈的关联标识：
{
"code": "inventory_not_enough",
"message": "库存不足，请调整购买数量",
"request_id": "demo-order-001"
}
4. 调用下游服务时透传 request_id
使用 httpx 调用库存服务时，显式传递当前 id：
import httpx
from observability import logger, request_id_var
async def reserve_inventory(sku_id: str, count: int) -> none:
headers = {"x-request-id": request_id_var.get()}
timeout = httpx.timeout(connect=1.0, read=2.0, write=2.0, pool=1.0)
try:
async with httpx.asyncclient(timeout=timeout) as client:
response = await client.post(
"http://inventory.internal/reservations",
json={"sku_id": sku_id, "count": count},
headers=headers,
)
response.raise_for_status()
except httpx.timeoutexception as exc:
logger.warning("inventory timeout", extra={"event": "inventory_timeout", "code": "dependency_timeout"})
raise apperror("dependency_timeout", "库存服务繁忙，请稍后重试", 503) from exc
except httpx.httpstatuserror as exc:
logger.error("inventory rejected request", extra={"event": "inventory_http_error"})
raise apperror("inventory_unavailable", "库存服务暂不可用", 503) from exc
不要无限重试所有下游异常。写操作重试前要确认幂等性，例如携带 idempotency-key，否则网络超时后重试可能造成重复扣库存。
5. 接入 opentelemetry
request_id 解决基础关联，opentelemetry 则提供标准 trace 上下文与 span 数据。最小初始化示例如下：
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import otlpspanexporter
from opentelemetry.sdk.resources import resource
from opentelemetry.sdk.trace import tracerprovider
from opentelemetry.sdk.trace.export import batchspanprocessor
from opentelemetry.instrumentation.fastapi import fastapiinstrumentor
from opentelemetry.instrumentation.httpx import httpxclientinstrumentor
def configure_tracing(app) -> none:
provider = tracerprovider(
resource=resource.create({"service.name": "order-api", "deployment.environment": "production"})
)
provider.add_span_processor(batchspanprocessor(otlpspanexporter(endpoint="http://otel-collector:4318/v1/traces")))
trace.set_tracer_provider(provider)
fastapiinstrumentor.instrument_app(app)
httpxclientinstrumentor().instrument()
实际部署中通常把 trace 发往 opentelemetry collector，再由 collector 导出到 jaeger、tempo 或云端观测平台。自动埋点已经能覆盖 fastapi 和 httpx；对“创建订单”“扣减积分”这类关键业务动作，可以再手动创建 span，并只添加低基数、无敏感信息的属性。
五、常见问题与实践建议
1. 不要记录敏感信息
密码、访问令牌、cookie、银行卡号、身份证号和完整地址不应写入普通应用日志。必要字段应脱敏，例如只记录手机号后四位；请求头应设置白名单，而不是将所有 header 序列化。日志平台的访问权限、保留期限和删除策略也是安全设计的一部分。
2. 不要吞掉异常
以下写法会让程序在错误状态下继续运行，也丢失排障线索：
try:
await save_order()
except exception:
pass
如果异常确实可恢复，应记录原因、采取明确的补偿动作，并保持异常链：raise apperror(...) from exc。from exc 能保留原始异常，便于后续查看完整堆栈。
3. 不要把堆栈返回给客户端
开发模式中 fastapi 的调试信息很便利，生产模式却可能暴露文件路径、依赖版本、sql 或密钥片段。对外始终返回统一错误响应；完整堆栈只进入受控日志系统。用户报障时提供 request_id，客服和研发即可精确检索。
4. 控制日志量与字段基数
访问日志、成功请求日志通常量最大。高并发接口可对成功日志采样，例如只保留 5%；错误日志不采样或设置单独策略。不要把订单号、用户 id、url 全参数作为 prometheus label，这会造成高基数指标爆炸。日志字段的高基数一般可接受，指标标签则应谨慎。
5. 异步任务需要显式传递上下文
celery 任务、消息队列消息、run_in_executor 线程池与当前 web 请求并非同一执行上下文。提交任务时把 request_id、traceparent 写入任务头或消息元数据；worker 启动后恢复这些字段，再写日志。否则 http 请求和后台处理会在观测平台中断开。
6. 建立可执行的告警规则
告警应面向用户影响，而不是只监控机器状态。例如：5 分钟内 5xx 比例超过 2%、接口 p95 延迟持续超过 800ms、任务队列积压超过阈值、下游超时比例突然升高。每条告警都应附带仪表盘、查询条件和处置说明，避免半夜只收到一句“服务异常”。
六、进阶思考
1. 将可观测性纳入接口契约
错误码、request_id 响应头、审计事件和关键指标不应在事故后临时添加，而应像请求字段和响应字段一样在 api 设计阶段确定。为每个对外 api 定义成功、可预期失败与不可预期失败的观测策略，能显著降低后期治理成本。
2. 区分业务日志、审计日志与运行日志
运行日志用于排障，例如超时、异常和耗时；业务日志用于分析关键事件，例如订单创建；审计日志则回答“谁在什么时候修改了什么”。三者的字段、保留期限、权限和不可篡改要求不同，不宜混在一张索引或一个文件中。
3. 用 sli、slo 连接技术指标与服务承诺
sli 是可量化指标，例如“成功响应比例”或“p95 响应时间”；slo 是目标，例如“月度订单创建成功率不低于 99.9%”。有了 slo，告警不再只基于瞬时波动，而可以围绕错误预算消耗速度设计。日志与 trace 用于解释 slo 变差的根因。
4. 在多服务环境中统一约定
团队应统一 json 字段名、错误码命名、时间格式、请求 id 请求头和 trace 采样规则。否则即使各服务都“有日志”，跨团队排障仍会遇到字段不一致、时区不同和 id 无法关联的问题。opentelemetry 是追踪标准，但采样率、资源属性、数据脱敏和 collector 路由仍需要团队治理。
结论
可靠的 python 后端不只要能正确处理成功请求，也要能在失败时留下可关联、可检索、可行动的证据。本文通过 fastapi 示例建立了从 request_id 中间件、contextvar、json 日志、统一异常响应到 http 下游透传的基础闭环，并介绍了 opentelemetry 如何将单服务日志扩展为分布式链路追踪。
下一步可以把日志输出接入 elk 或 loki，把指标接入 prometheus 和 grafana，把 trace 接入 opentelemetry collector 与 jaeger/tempo；再为关键接口定义 slo 和演练告警响应流程。至此，后端服务才真正具备持续运行和高效排障的基础。
以上就是一文详解python后端服务的日志、异常处理与链路追踪的详细内容，更多关于python日志、异常处理与链路追踪的资料请关注代码网其它相关文章！
发表评论