Python 日志结构化：JSON 格式 + TraceID + 业务字段的三层规范

Python 日志结构化JSON 格式 TraceID 业务字段的三层规范一、日志里有但查不到的排障困境线上用户投诉订单支付成功但状态没更新。查日志时发现同一笔交易的日志分散在 4 个不同的服务里网关写了一条INFO: payment received订单服务写了DEBUG: updating status支付回调写了一条ERROR: timeout消息队列写了一条WARN: retry exhausted。但日志格式全是自由文本没有 TraceID、没有订单号、没有统一的时间格式。用 grep 搜order_id12345有的服务用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, tztimezone.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_asciiFalse, defaultstr) # 自定义 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_idtrace_id, user_iduser_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 statuspaid 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 倍。