Python 结构化日志实战：轮转、级别与上下文一次讲清

程序在本地能运行，并不代表出现故障时就容易排查。只有“请求失败”四个字，往往看不出哪个请求出错、在哪一步失败，也不知道是否影响了其他请求。本文从一个可运行的小例子出发，用 Python 标准库把日志级别、JSON 结构化输出、文件轮转和请求上下文串起来。

示例面向 Python 3.9 及以上的单进程程序，无需安装第三方依赖。它是一份便于理解和改造的基础配置，不包含真实业务数据或性能测试结论。

一、先约定日志要回答什么问题

一条有用的日志，至少应说明时间、严重程度、所属模块和发生的事情。如果一次请求经过多个函数，还应带上相同的 request_id，让这些记录可以关联起来。

例如，“处理完成，耗时 18 毫秒”适合人阅读；当需要统计大量请求时，把 event、duration_ms、status 拆成独立字段，会更容易筛选和聚合。本文采用 JSON Lines：每一行是一个完整的 JSON 对象。异常堆栈中的换行由 JSON 编码转义，不会把一条事件拆成多个物理行。

建议先保持字段稳定：timestamp 表示带时区的时间；level 表示严重程度；logger 表示来源；message 用于解释；request_id 用于关联；event 用于标识固定的事件类型。耗时应保留数值，避免把“18ms”作为字符串后再反复解析。

二、看懂 Logger、Handler、Formatter 和 Filter

Logger 是业务代码调用的入口；Handler 决定输出到控制台还是文件；Formatter 决定最终文本或 JSON 的格式；Filter 可以筛选记录，也可以补充上下文字段。本文在 Handler 上添加 Filter，使控制台和文件输出前都能取得 request_id。

级别检查需要同时看 Logger 和 Handler。本例让 Logger 接收 DEBUG 及以上记录，控制台输出 INFO 及以上，文件保存 DEBUG 及以上。因此调试细节进入文件，终端保留关键进度。如果把 Logger 本身设成 INFO，DEBUG 记录在入口就被丢弃，文件 Handler 设成 DEBUG 也无法补回。

• DEBUG：定位问题所需的细节。
• INFO：正常业务的重要节点，例如任务开始与结束。
• WARNING：出现异常迹象，但当前流程仍可继续。
• ERROR：某项操作失败，需要排查。
• CRITICAL：严重故障，通常影响服务整体运行。

三、完整示例：控制台简洁，文件保留结构

将下面代码保存为 app.py，然后在该文件所在目录执行 python app.py。日志目录以当前工作目录为基准；部署时应明确工作目录、写入权限和持久化位置。

import json
import logging
import sys
from contextvars import ContextVar
from datetime import datetime, timezone
from logging.handlers import RotatingFileHandler
from pathlib import Path

request_id_var = ContextVar("request_id", default="-")


class RequestContextFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id_var.get()
        return True


class JsonFormatter(logging.Formatter):
    def format(self, record):
        data = {
            "timestamp": datetime.fromtimestamp(
                record.created, timezone.utc
            ).isoformat(timespec="milliseconds"),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "request_id": getattr(record, "request_id", "-"),
        }
        # 只收集明确约定的业务字段，避免把整个对象写进日志。
        for key in ("event", "duration_ms", "status"):
            if hasattr(record, key):
                data[key] = getattr(record, key)
        if record.exc_info:
            data["exception"] = self.formatException(record.exc_info)
        return json.dumps(data, ensure_ascii=False, default=str)


def configure_logging(log_dir="logs"):
    logger = logging.getLogger("demo")
    logger.setLevel(logging.DEBUG)
    logger.propagate = False

    # 此函数仅在单进程程序启动时调用，不在请求处理中调用。
    # 若启动流程重复配置，先关闭本示例拥有的旧 handler。
    for handler in logger.handlers[:]:
        logger.removeHandler(handler)
        handler.close()

    Path(log_dir).mkdir(parents=True, exist_ok=True)

    console = logging.StreamHandler(sys.stdout)
    console.setLevel(logging.INFO)
    console.setFormatter(logging.Formatter(
        "%(levelname)s [%(request_id)s] %(message)s"
    ))

    file_handler = RotatingFileHandler(
        Path(log_dir) / "app.jsonl",
        maxBytes=1024 * 1024,
        backupCount=3,
        encoding="utf-8",
    )
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(JsonFormatter())

    for handler in (console, file_handler):
        handler.addFilter(RequestContextFilter())
        logger.addHandler(handler)

    return logger


def handle_request(logger, request_id):
    token = request_id_var.set(request_id)
    try:
        logger.debug("进入处理函数", extra={"event": "request.start"})
        logger.info("开始处理请求", extra={"event": "request.accepted"})
        try:
            int("invalid")
        except ValueError:
            logger.exception(
                "参数解析失败",
                extra={"event": "request.failed", "status": "invalid"},
            )
        logger.info("请求处理结束", extra={"event": "request.end"})
    finally:
        request_id_var.reset(token)


if __name__ == "__main__":
    log = configure_logging()
    handle_request(log, "req-demo-001")
    log.info("程序退出", extra={"event": "app.stop"})
    logging.shutdown()

这里使用专用的 demo Logger，并关闭向父 Logger 传播，避免同一条记录同时被自己的 Handler 和上层 Handler 输出。模块拆分后，可以使用 demo.worker 等子 Logger，并让它们传播到 demo，由应用入口统一配置。

初始化函数会关闭并替换 demo 自身的 Handler，因此只适合这个独立示例所拥有的 Logger。接入已有框架时，应先确认框架的日志配置，不要直接清空根 Logger，也不要在每次请求中重新配置日志。

四、运行后应该检查哪些结果

控制台会看到带有 req-demo-001 的请求日志，参数解析失败时还会显示异常堆栈。DEBUG 记录不会出现在控制台，但应出现在 logs/app.jsonl 中。最后的“程序退出”记录，其 request_id 应恢复为“-”，说明请求上下文已经清理。

文件中的一条记录形如下面这样，时间仅为格式示意，实际值以运行结果为准：

{"timestamp":"2026-10-25T00:00:00.000+00:00","level":"INFO","logger":"demo","message":"开始处理请求","request_id":"req-demo-001","event":"request.accepted"}

可以再运行下面这段检查脚本，逐行解析日志。如果文件中有格式损坏或被截断的行，json.loads 会抛出异常。断言通过只说明本次生成记录满足这些基本条件，不代表已经完成并发或生产环境验证。

import json
from pathlib import Path

rows = [
    json.loads(line)
    for line in Path("logs/app.jsonl").read_text(
        encoding="utf-8"
    ).splitlines()
    if line.strip()
]
assert any(row["level"] == "DEBUG" for row in rows)
assert any(
    row.get("event") == "request.failed" and "exception" in row
    for row in rows
)
assert rows[-1]["request_id"] == "-"
print("基本检查通过，记录数：", len(rows))

五、轮转解决文件增长，保留策略解决磁盘占用

示例使用 RotatingFileHandler，maxBytes 为 1 MiB，backupCount 为 3。达到轮转条件后，当前文件会归档为 app.jsonl.1，更旧的备份依次后移，只保留约定数量。这里的容量是控制目标，不应把它理解为严格的磁盘配额；特别大的单条日志仍可能突破预期大小。

为了快速观察效果，可以在测试目录把 maxBytes 临时改成 1024，再连续写入多条日志，检查备份文件是否生成。测试后恢复配置。若按日期管理更方便，可考虑 TimedRotatingFileHandler；它通常在新日志到来时检查是否需要轮转，并非独立运行的定时归档任务。

轮转不等于长期归档。需要长期保存时，还要设计采集、压缩、过期清理和磁盘告警。容器部署则可以只输出 JSON 到标准输出，由运行环境统一采集，避免应用内部轮转与外部轮转同时管理同一文件。

六、请求上下文为什么要在 finally 中恢复

把 request_id 放在普通全局变量里，并发请求可能相互覆盖。ContextVar 为当前执行上下文保存值，在 asyncio 场景中可以用来关联任务内的日志。示例先保存 set 返回的 token，最后在 finally 中 reset，即使中途抛出异常，也能恢复进入请求前的状态。

实际 Web 服务可在中间件入口设置 request_id，在请求结束时恢复。来自外部的请求标识应校验长度和字符；缺失时可以生成新的标识。不要把它当作身份认证依据。

上下文也有边界：新进程、任意线程池或跨服务调用不会自动获得完整请求链路。跨服务应通过协议明确传递关联字段；切换到 QueueHandler 后，应在日志进入队列前把上下文写入 LogRecord，不能等后台监听线程再读取请求线程的 ContextVar。

七、接入项目时最容易踩的坑

• 一条日志出现两遍：检查是否重复添加 Handler，以及当前 Logger 与父 Logger 是否同时输出。统一由应用入口配置，业务模块只获取 Logger。
• 异常只有一句话：在 except 内使用 logger.exception，或使用 logger.error(..., exc_info=True)，保留堆栈。应由明确的一层负责记录同一次失败，避免层层重复。
• extra 字段冲突：不要覆盖 name、msg、levelname 等 LogRecord 保留属性。本文只输出约定好的业务字段，新增字段时同时更新 Formatter 的白名单。
• 多进程争用轮转文件：本例适用于单进程。多个 worker 不应直接共用这个 RotatingFileHandler 写同一个文件。可由 QueueHandler 与单独的监听端集中落盘，或输出到标准输出后集中采集。
• 日志变成业务瓶颈：先测量日志量、磁盘延迟和单条大小。需要队列解耦时，应一并考虑队列积压、丢弃策略与退出时排空，不能只把写入动作移到后台。
• 把敏感内容写进日志：不要直接记录密码、Token、Authorization 请求头、完整请求体或个人资料。结构化字段应先约定范围，对异常消息也要检查是否包含敏感输入。

八、从这个示例继续改造

小项目可以先固定字段与事件名，确保出错时能凭 request_id 还原一次处理过程；接着再按部署方式选择文件或标准输出，并用实际日志量决定轮转与保留策略。当程序扩展为多进程或分布式服务时，再补上集中采集、链路传递与告警。日志配置的价值，最终体现在一次真实故障是否能被快速解释清楚。