Python 日志结构化：JSON 格式 + TraceID + 业务字段的三层规范
一、"日志里有，但查不到"的排障困境
线上用户投诉"订单支付成功但状态没更新"。查日志时发现同一笔交易的日志分散在 4 个不同的服务里：网关写了一条INFO: payment received，订单服务写了DEBUG: updating status，支付回调写了一条ERROR: timeout，消息队列写了一条WARN: retry exhausted。但日志格式全是自由文本，没有 TraceID、没有订单号、没有统一的时间格式。
用 grep 搜order_id=12345，有的服务用orderId:12345，有的用order_id: 12345，有的根本没打。这种"文字日志"在微服务架构下是排障的灾难。结构化的本质不是改格式，而是让日志从"给人看的故事"变成"机器可检索的数据"。
二、结构化日志的三层信息模型
日志应该分三层承载信息：
基础层（Base）：时间戳、日志级别、服务名、TraceID、SpanID。这些是分布式追踪的基础设施字段，每条日志都必须有。
业务层（Business）：用户ID、订单号、操作类型、请求参数、响应摘要。这些是业务排障的核心信息，由各模块按需添加。
诊断层（Diagnostic）：错误堆栈、SQL 耗时、API 调用延迟、Redis 命中率。运维和性能优化的关键数据。
flowchart TD A[请求进入] --> B[中间件层] B --> B1[生成 TraceID] B --> B2[注入基础字段] B1 --> C[业务逻辑层] B2 --> C C --> D[数据库操作] D --> D1{记录: SQL + 耗时} C --> E[缓存操作] E --> E1{记录: Key + 命中状态} C --> F[外部 API 调用] F --> F1{记录: 端点 + 延迟 + 状态码} D1 --> G[日志聚合器] E1 --> G F1 --> G G --> H[Elasticsearch / Loki] H --> I[Grafana 面板查询] H --> J[告警规则匹配]这一结构的优势在于：基础层自动注入，业务层按需添加，诊断层在关键节点记录。每一层职责清晰，不会出现"所有信息都堆在一条日志里"的情况。
三、Python 结构化日志实现
import json import logging import sys import time import uuid import traceback from contextvars import ContextVar from datetime import datetime, timezone from functools import wraps from typing import Any, Optional # ========== 上下文变量：跨函数传递 TraceID ========== _trace_id: ContextVar[str] = ContextVar("trace_id", default="") _user_id: ContextVar[str] = ContextVar("user_id", default="") _request_id: ContextVar[str] = ContextVar("request_id", default="") def set_trace_context(trace_id: str = "", user_id: str = "", request_id: str = ""): """设置请求级别的上下文信息""" if trace_id: _trace_id.set(trace_id) if user_id: _user_id.set(user_id) if request_id: _request_id.set(request_id) # ========== 结构化日志格式化器 ========== class StructuredFormatter(logging.Formatter): """JSON 格式结构化日志格式化器""" def __init__(self, service_name: str = "unknown"): super().__init__() self.service_name = service_name def format(self, record: logging.LogRecord) -> str: """将日志记录格式化为 JSON 字符串""" log_entry: dict[str, Any] = { # === 基础层：每条日志都必须有的字段 === "timestamp": datetime.fromtimestamp( record.created, tz=timezone.utc ).isoformat(), "level": record.levelname, "service": self.service_name, "logger": record.name, "trace_id": _trace_id.get(), "request_id": _request_id.get(), # === 业务层：从 extra 中提取 === "message": record.getMessage(), "module": record.module, "function": record.funcName, "line": record.lineno, } # 注入 extra 中传递的业务字段 if hasattr(record, "user_id"): log_entry["user_id"] = record.user_id or _user_id.get() if hasattr(record, "order_id"): log_entry["order_id"] = record.order_id if hasattr(record, "operation"): log_entry["operation"] = record.operation if hasattr(record, "duration_ms"): log_entry["duration_ms"] = record.duration_ms # === 诊断层：错误信息和性能数据 === if record.exc_info and record.exc_info[1]: log_entry["error"] = { "type": type(record.exc_info[1]).__name__, "message": str(record.exc_info[1]), "traceback": traceback.format_exception(*record.exc_info), } # extra 中的任意字段都合并进去 for key in ("db_query", "cache_key", "cache_hit", "api_endpoint", "http_status", "request_body", "response_summary"): if hasattr(record, key): log_entry[key] = getattr(record, key) return json.dumps(log_entry, ensure_ascii=False, default=str) # ========== 自定义 Logger Adapter ========== class BusinessAdapter(logging.LoggerAdapter): """业务日志适配器，支持一键注入业务字段""" def process(self, msg, kwargs): # 合并业务上下文 extra = kwargs.get("extra", {}) if _user_id.get(): extra.setdefault("user_id", _user_id.get()) if _request_id.get(): extra.setdefault("request_id", _request_id.get()) kwargs["extra"] = extra return msg, kwargs def with_business(self, **fields) -> "BusinessAdapter": """返回一个注入了业务字段的新 adapter""" return BusinessAdapter(self.logger, {**self.extra, **fields}) # ========== 初始化日志系统 ========== def init_logging(service_name: str, level: int = logging.INFO): """初始化结构化日志系统""" logger = logging.getLogger() logger.setLevel(level) # 移除默认的 Handler logger.handlers.clear() # JSON 格式的 stdout handler（生产环境） handler = logging.StreamHandler(sys.stdout) handler.setFormatter(StructuredFormatter(service_name)) logger.addHandler(handler) # 关闭第三方库的冗余日志 logging.getLogger("urllib3").setLevel(logging.WARNING) logging.getLogger("httpx").setLevel(logging.WARNING) # ========== 使用示例 ========== # 初始化 init_logging("order-service") logger = BusinessAdapter(logging.getLogger("order"), {}) # 模拟请求处理 def process_payment(order_id: str, amount: float): """处理支付：展示结构化日志的最佳实践""" trace_id = str(uuid.uuid4()) set_trace_context(trace_id=trace_id, user_id="user_888") # 场景 1：普通业务日志 logger.info( "开始处理支付", extra={ "order_id": order_id, "amount": amount, "operation": "payment_process", }, ) # 场景 2：数据库操作日志（带耗时） start = time.time() # db.execute(...) time.sleep(0.05) # 模拟数据库操作 duration = (time.time() - start) * 1000 logger.debug( "订单状态更新完成", extra={ "order_id": order_id, "duration_ms": round(duration, 2), "db_query": "UPDATE orders SET status='paid' WHERE id=?", }, ) # 场景 3：外部 API 调用日志 api_start = time.time() try: # response = requests.post(payment_gateway, json={...}) time.sleep(0.2) # 模拟 API 调用 api_duration = (time.time() - api_start) * 1000 logger.info( "支付网关调用成功", extra={ "order_id": order_id, "api_endpoint": "https://payment-gw/pay", "duration_ms": round(api_duration, 2), "http_status": 200, }, ) except Exception as e: logger.exception( "支付网关调用失败", extra={ "order_id": order_id, "api_endpoint": "https://payment-gw/pay", }, ) # ========== 装饰器：自动记录函数耗时 ========== def log_execution_time(func): """装饰器：自动记录函数执行时间""" @wraps(func) def wrapper(*args, **kwargs): start = time.perf_counter() try: result = func(*args, **kwargs) duration = (time.perf_counter() - start) * 1000 logger.debug( f"{func.__name__} 执行成功", extra={"duration_ms": round(duration, 2)}, ) return result except Exception: duration = (time.perf_counter() - start) * 1000 logger.exception( f"{func.__name__} 执行失败", extra={"duration_ms": round(duration, 2)}, ) raise return wrapper四、结构化日志的边界与权衡
日志量会显著增加。JSON 格式的日志比纯文本大约膨胀 30%-50%。需要在日志采集端做采样（如只记录 1% 的 DEBUG 日志）或按级别分流（ERROR 全量、INFO 采样）。
敏感信息泄露风险。结构化日志让字段变得可检索的同时，也让敏感数据更容易被暴露。手机号、身份证、Token 等敏感字段，必须在写入前做脱敏处理。可以扩展 Formatter，对标记为sensitive的字段自动哈希或截断。
字段命名规范必须统一。如果订单服务用order_id，支付服务用orderId，查询时就需要用两个字段名。建议在团队内制定字段命名规范，推荐 snake_case + 全小写英文。
本地开发不需要结构化。开发环境的结构化日志（一行 JSON 几百字符）可读性极差。在StructuredFormatter之外保留一个PlainFormatter，通过环境变量切换。
五、总结
日志从"自由文本"到"结构化 JSON"的升级，不是为了好看，而是为了让排障从 grep 变成 SQL 查询。三条核心规范：基础层字段强制统一（TraceID + 级别 + 服务名）、业务层字段按需添加但不遗漏关键信息、诊断层在关键节点自动记录性能数据。统一日志格式这件事，做得越早，代价越小——等到 20 个微服务各自风格不一的时候再改，成本就是 20 倍。
