Python · logging 日志体系

场景 1 · dictConfig 一份配置管全服务: 控制台人读 + 文件 JSON

服务要同时满足"运维 tail 控制台"和"采集器吃 JSON 文件"两个消费方, 一张 dictConfig 声明整棵树:

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,      # 不杀库先建的 logger
    "formatters": {
        "human": {"format": "%(asctime)s %(levelname)-7s %(name)s: %(message)s"},
        "json":   {"()": "app.JsonFormatter"},        # 文件侧机器可读
    },
    "handlers": {
        "console": {"class": "logging.StreamHandler",
                    "stream": "ext://sys.stdout", "formatter": "human"},
        "file": {"class": "logging.handlers.RotatingFileHandler",
                "formatter": "json", "filename": "logs/app.log",
                "maxBytes": 104857600, "backupCount": 5},
    },
    "root": {"handlers": ["console", "file"], "level": "INFO"},
}
logging.config.dictConfig(LOGGING)         # 入口处一次调用, 全服务生效

场景 2 · logger.exception: except 块里一行带全栈

支付失败排障时, 只有 "timeout" 两个字没有栈, 等于没日志; exception() 自动附带当前异常完整链:

def charge(order_id: str):
    try:
        gw = PaymentGateway()               # 初始化也可能失败
        return gw.charge(order_id)
    except TimeoutError:
        log.warning("gateway timeout, will retry: order=%s", order_id)
        raise                             # 瞬时错误: 上层有重试
    except Exception:
        log.exception("charge failed order=%s", order_id)  # 自动带 40 行栈
        raise PaymentError(order_id) from None
# str(e) 只有一行消息; exception() = 消息 + 完整调用栈 + cause 链

事故复盘时间从"猜半天"变成"看栈即知": 报错行、调用链、cause 一屏全齐。

场景 3 · 灰度只拉一个模块到 DEBUG

计费对账差一分钱, 要看 SQL 明细但不能全服务 DEBUG(日志量 100 倍), 按模块精确调级:

# 发版前: dictConfig 里只给 app.billing 开 DEBUG
LOGGING["loggers"] = {
    "app.billing": {"level": "DEBUG", "propagate": True},
    "app":         {"level": "INFO"},
}

# 线上不停机临时开 (改内存, 不落盘):
logging.getLogger("app.billing").setLevel(logging.DEBUG)
logging.getLogger("app.billing.sqlalchemy.engine").setLevel(logging.INFO)
# 排查完记得调回去 —— DEBUG 的 SQL 明细 1 分钟能写 2GB

场景 4 · RotatingFileHandler 给磁盘上保险丝

裸 FileHandler 写了三周把 2GB 数据盘写满, 服务整个挂掉; 按大小轮转 + 备份数封顶:

from logging.handlers import RotatingFileHandler

h = RotatingFileHandler(
    "logs/app.log",
    maxBytes=100 * 1024 * 1024,           # 单文件 100MB 即轮转
    backupCount=5,                          # app.log.1 ... app.log.5
    encoding="utf-8",                       # 中文不乱码
)
log = logging.getLogger("app")
log.setLevel(logging.INFO)                   # 两级闸门都要开
log.addHandler(h)
# 磁盘占用上限 = 100MB × (1 + 5) = 600MB, 永远打不满 2GB 数据盘
# 注意: 多进程共用此 handler 会互相踩轮转 → 场景 5

场景 5 · QueueHandler + QueueListener: 脱离主线程、多 worker 汇聚

gunicorn 4 worker 各自轮转同一文件, 互相 rename 踩脚丢单; 换成"worker 只入队, 主进程一个 listener 独家写文件":

from logging.handlers import QueueHandler, QueueListener
import multiprocessing as mp

q = mp.Manager().Queue()                     # 跨进程队列 (spawn/fork 都行)

def worker_logger(name: str):
    lg = logging.getLogger(name)
    lg.handlers = [QueueHandler(q)]          # 业务侧只入队, 不做慢 I/O
    lg.setLevel(logging.INFO)
    return lg

listener = QueueListener(q, rotating_file_handler, console_handler,
                         respect_handler_level=True)
listener.start()                             # 一个进程独家轮转/格式化/落盘

副作用收益: 业务线程的日志调用从"同步写文件"变"入队即返回", P99 尾部抖动同步消失。

场景 6 · 结构化 JSON: extra 字段 + request_id 贯穿

排障要按请求串联所有日志, 人读格式没法 grep; LoggerAdapter 把 req_id 注进每条记录:

class JsonFormatter(logging.Formatter):
    def format(self, r):
        d = {"ts": utc_iso(r.created), "lvl": r.levelname,
             "logger": r.name, "msg": r.getMessage(),
             "req_id": getattr(r, "req_id", None)}   # extra 注入的字段
        if r.exc_info: d["exc"] = self.formatException(r.exc_info)
        return json.dumps(d, ensure_ascii=False)

log = logging.LoggerAdapter(logging.getLogger("app"),
                            {"req_id": ctx.request_id})
log.info("order paid")                     # 同一请求的全部日志同 req_id
# Loki/ES: req_id=abc123 一搜, 这单支付 17 条日志按序全出

场景 7 · 高频日志 1/100 采样降噪

健康检查探活每秒 2000 条 INFO, 告警平台被打爆; 用 Filter 采样, 只留百分之一:

class Sampler(logging.Filter):
    def __init__(self, rate=100):
        self.rate, self.n = rate, 0
    def filter(self, record):                # 必须显式 return bool
        self.n += 1
        keep = self.n % self.rate == 0       # 100 条留 1 条
        if keep: record.sampled = 1       # 标记: 知道这是采样
        return keep

logging.getLogger("app.health").addFilter(Sampler(100))
# 2000/s → 20/s; ERROR 级别走独立 logger 不采样, 异常一条不丢

场景 8 · 第三方库噪音治理

一开 DEBUG, urllib3 连接池和 uvicorn.access 每秒几万条, 业务日志被淹; 库日志单独压级:

LOGGING["loggers"] = {
    "urllib3":          {"level": "WARNING"},   # 连接池 DEBUG 刷屏
    "uvicorn.access":  {"level": "WARNING", "propagate": False},
    "sqlalchemy.engine": {"level": "WARNING"},
    "celery":          {"level": "INFO"},
    "app":              {"level": "INFO"},     # 自己的业务不变
}
# 纪律: 第三方默认 WARNING, 只有自己的代码吃 INFO/DEBUG 配额

治理后日志量降 85%, 真正的业务信号不再被基建噪音稀释。

场景 9 · 日志脱敏 Filter: 密码/token 进日志前洗掉

合规审计发现手机号和 Authorization 原文进了日志文件; 在源头统一清洗, 而不是靠每个开发自觉:

import re

SENSITIVE = ("password", "token", "authorization", "phone")

class MaskFilter(logging.Filter):
    def filter(self, record):
        msg = record.msg
        if isinstance(msg, str):
            for k in SENSITIVE:            # 匹配 JSON 形态与 k=v 形态
                msg = re.sub(rf'("{k}"\s*:\s*")[^"]+', r'\1***', msg)
                msg = re.sub(rf'({k}=)[^\s,]+', r'\1***', msg)
            record.msg = msg                # 源头改掉, Formatter 只见干净数据
        return True                          # 别忘了放行

logging.getLogger().handlers[0].addFilter(MaskFilter())

场景 10 · 聚合到 ELK/Loki 的字段约定

五个服务各写各的格式, 平台没法统一索引; 约定四个字段全公司对齐:

# ts       UTC ISO8601 带毫秒: 2026-09-26T08:01:02.123Z (不要本地时区!)
# svc      服务名: payment / order ...    与部署单元对齐
# lvl      级别名: INFO / ERROR           统一大写
# trace_id W3C trace id: 跨服务串联      中间件注入 contextvar

log.info("refund done", extra={"svc": "payment",
                                 "trace_id": tid})
# Loki 一条查询链穿三个服务:
#   {svc="payment"} |= "trace_id=9f3c..."  →  order → risk → db
# 字段不齐 = 检索失效: "格式即接口", 日志也逃不掉