如何通过Python标准logging库实现集中式10%比例日志采样

完全可以基于logging库原生能力实现，不需要引入第三方依赖，核心逻辑是利用logging的过滤器（Filter）机制+日志冒泡规则，在根日志器（root logger）上统一挂载采样过滤器，即可管控所有子logger的日志输出决策，不需要修改业务侧任何打日志的代码。

核心设计原则 #

采样必须以请求为维度，而非单条日志为维度：如果单条日志随机采样，会出现同一个请求的日志部分输出、部分丢弃的情况，碎片化的日志完全无法用于问题排查。我们需要保证命中采样的请求输出全量日志，未命中的请求不输出任何业务日志，最终整体日志量控制在10%左右。

具体实现代码 #

1. 初始化全局采样配置 #

基于Python 3.7+自带的contextvars实现跨协程/线程的请求上下文标记，避免多请求并发时的标记串扰：

import logging
import random
from contextvars import ContextVar

# 上下文变量：标记当前请求是否命中采样，默认未命中
is_sampled: ContextVar[bool] = ContextVar("is_sampled", default=False)
# 全局采样率，10%对应0.1
SAMPLE_RATE = 0.1

class RequestSamplingFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        # 直接返回当前请求的采样标记，同一个请求所有日志判定结果完全一致
        return is_sampled.get()

def init_log_config():
    root_logger = logging.getLogger()
    # 清空默认handler避免重复打印（可选，根据自己项目情况调整）
    root_logger.handlers.clear()
    # 核心：给根logger挂载采样过滤器，所有子logger冒泡上来的日志都会经过该过滤
    root_logger.addFilter(RequestSamplingFilter())
    
    # 后续正常配置你的日志handler即可，比如控制台输出、日志上报handler等
    stream_handler = logging.StreamHandler()
    stream_handler.setFormatter(logging.Formatter(
        "%(asctime)s - %(levelname)s - %(name)s - %(message)s"
    ))
    root_logger.addHandler(stream_handler)
    root_logger.setLevel(logging.INFO)

2. 请求入口添加采样判定逻辑 #

在你所用web框架的请求中间件/拦截器里，每个请求进入时先做采样判定，请求结束后重置上下文标记避免污染：

def process_request_middleware(req):
    # 请求进入时随机判定是否采样
    sampled = random.random() < SAMPLE_RATE
    token = is_sampled.set(sampled)
    try:
        # 执行后续的请求处理、业务逻辑
        return handle_request(req)
    finally:
        # 请求结束重置上下文，防止复用工作线程/协程时标记串到下一个请求
        is_sampled.reset(token)

方案优势 #

• 完全原生：基于logging标准库的Filter、日志冒泡原生机制实现，无额外依赖
• 集中管控：只需要在项目启动时执行一次init_log_config()，所有业务代码中通过logging.getLogger(__name__)获取的子logger，都会自动遵循采样规则，不需要修改任何业务侧打日志的代码
• 日志完整：命中采样的请求保留全量日志，完全满足排查问题的需求
• 灵活调整：需要修改采样率时，只需要调整全局SAMPLE_RATE常量即可，不需要散改多处配置

特殊场景说明 #

如果你的项目中有部分子logger手动设置了propagate = False（关闭日志向父级冒泡），这部分日志不会经过根logger的过滤器，只需要给这类子logger单独挂载同一个RequestSamplingFilter实例即可，一般不建议业务代码随意关闭日志冒泡，默认配置下全局一次挂载就能覆盖所有场景。

如果你不需要保留请求维度的完整日志，只想按日志条数随机采样10%，可以把过滤器逻辑简化，不需要上下文变量：

class CountBasedSamplingFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        return random.random() < SAMPLE_RATE

注意：该方式会导致日志碎片化，生产环境不推荐使用

内容的提问来源于stack exchange，提问作者Jean Carlo Machado