日志采集实战：用 Python logging 模块输出结构化 SDK 日志

日志采集，是整个异常检测体系的“地基”。地基不稳，后面再花哨的算法都是白搭。说实话，我见过太多团队花大价钱搞了一套检测模型，结果因为日志字段缺胳膊少腿、时间戳格式不对、或者干脆没打日志，模型跑起来跟没头苍蝇一样。

这一章，我们就聚焦一个最实际的问题：怎么用 Python 自带的 logging 模块，把 SDK 日志打得既规范又好解析。

为什么非得用 logging 模块？print 不够吗？

你想想看，生产环境的 SDK 每分钟可能打印成千上万条日志。print 只能往控制台写，没法控制级别、没法区分输出目的地、也没法做格式化。更关键的是——print 是同步阻塞的，日志量一大，你的业务代码性能直接崩。

logging 模块天生支持：

• 日志级别控制——DEBUG、INFO、WARNING、ERROR、CRITICAL，按需过滤。
• 多输出通道——同时写文件、控制台、远程收集器。
• 格式化能力——统一加时间戳、进程号、线程号、调用位置。
• 非阻塞 IO——配合 QueueHandler 甚至不会拖慢主流程。

标准结构化日志该长什么样？

我先给你一个“及格线”的日志格式。这种格式在 ELK、Splunk 或者自建日志平台里都能被自动解析。

你看，每个字段都有明确含义，机器解析起来完全没歧义。我个人习惯把 extra 字段放成 JSON 字符串，这样无论是 grep 还是 JSON 解析器都能直接用。

实战：配置一个生产级别的 Logger

下面这段配置，我从第一个 SDK 项目用到现在，基本就改改文件名和格式。你拿过去就能用。

import logging
import logging.handlers
import json
from datetime import datetime, timezone

class StructuredFormatter(logging.Formatter):
    """输出 JSON 格式的结构化日志"""
    
    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "module": record.module,
            "funcName": record.funcName,
            "lineno": record.lineno,
            "message": record.getMessage(),
        }
        # 把 extra 参数合并进日志
        if hasattr(record, "extra_fields"):
            log_entry.update(record.extra_fields)
        return json.dumps(log_entry, ensure_ascii=False)

def setup_sdk_logger(name: str = "sdk.access") -> logging.Logger:
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)

    # 控制台输出（开发用）
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)
    console_formatter = StructuredFormatter()
    console_handler.setFormatter(console_formatter)

    # 文件输出（生产用，按天轮转）
    file_handler = logging.handlers.TimedRotatingFileHandler(
        filename="logs/sdk-access.log",
        when="midnight",
        backupCount=30,
        encoding="utf-8"
    )
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(StructuredFormatter())

    logger.addHandler(console_handler)
    logger.addHandler(file_handler)
    
    return logger

这段代码做了几件事：

• 自定义了一个 StructuredFormatter，每条日志打出来都是一行 JSON。这是为了能被 Logstash 或 Fluentd 直接吃掉。
• 控制台开 DEBUG 级别，方便本地调试时看全量日志。
• 文件只记 INFO 及以上，避免磁盘被调试日志灌满。
• 文件按天轮转，保留 30 天。嗯，我之前忘了配 backupCount，结果一个月后磁盘报警……这坑我替你踩过了。

如何在 SDK 里传递上下文？

一个请求从进入 SDK 到返回，中间可能经过好几个函数调用。要是每个函数都自己拼一个日志字段，字段名早晚乱成一锅粥。我习惯在入口处生成一个 trace_id，然后一路传下去。

import uuid
import logging

logger = logging.getLogger("sdk.access")

def handle_request(request):
    trace_id = str(uuid.uuid4())
    context = {"trace_id": trace_id, "client_ip": request.get("ip")}
    
    logger.info("Request received", extra={"extra_fields": context})
    # 处理逻辑...
    result = process(request, context)
    logger.info("Request completed", extra={"extra_fields": context})
    return result

def process(request, context):
    logger.debug("Processing started", extra={"extra_fields": context})
    # ...

你看，只要在每个 log 调用里挂上 extra={"extra_fields": context}，所有相关日志就自动带上了同一个 trace_id。将来根据 trace_id 一查，整条调用链一目了然。

日志级别应该怎么定？

级别定错了，要么信息太多，要么漏掉关键线索。我一般这么约定：

你还需要一套日志配置规范

代码写好了，还得有人用对。我见过最头疼的情况是——同一个 SDK 里有人用 logger.info，有人用 print，还有人直接把字符串拼到文件里。

不妨在项目 README 里加一个“日志规范”章节：

1. 不允许直接使用 print 或 logging.basicConfig。
2. 所有日志必须通过 setup_sdk_logger 获取 logger。
3. 关键业务操作（鉴权、限流、降级）必须打 INFO 日志，并带 trace_id。
4. 异常栈信息用 logger.exception，不要自己拼 str(e)。
5. 禁止在日志里记录密码、密钥、完整身份证号等敏感字段。

说实话，没有规范的团队，早晚会在某个半夜被乱糟糟的日志折腾得欲哭无泪。

本章小结：你手里的“原材料”准备好了

这一章我们只做了一件事——把 SDK 日志从“随便打打”升级成“结构化的、可解析的、带上下文的日志”。别小看这一步。后面的所有异常检测算法，都是建立在干净、一致的日志数据之上的。

我把它画成一张图，帮你理清整体流程：

这张图讲清楚了：业务代码经过 logging 拦截点，送入自定义格式化器转成 JSON，再同时分发给本地文件、控制台和远程收集器。过程中记得注入上下文和链路追踪 ID。每个环节单独可控，互不干扰。