python-logging-best-practices

Python 日志记录最佳实践

自我演进的技能：此技能会在使用中不断改进。如果说明有误、参数发生偏移或需要采用变通方案，请立即修复此文件，不要拖延。仅针对真实、可复现的问题进行更新。

何时使用此技能

在以下情况下使用此技能：

• 为任何服务或脚本设置 Python 日志记录
• 配置用于分析的结构化 JSONL 日志
• 实施日志轮转
• 在轻量级（零依赖）和功能齐全的日志记录方案之间进行选择
• 为容器化应用、systemd 应用或本地应用添加日志记录

概述

本指南统一整理了 Python 日志记录模式，旨在优化机器可读性（Claude Code 分析）和运行可靠性。从最轻量且可行的方案开始，仅在需要时扩展。

决策原则：从轻量方案开始，按需扩展

Is it < 5 services on a single machine, < 1 event/sec?
  YES → Lightweight Pattern (print + JSONL telemetry)
  NO  → Is it containerized / serverless?
    YES → stdout JSON (any library), no file rotation
    NO  → Is OTel tracing required?
      YES → structlog + OTel
      NO  → loguru (CLI tools) or stdlib RotatingFileHandler

方案	使用场景	优点	缺点
轻量级	小型 systemd 服务、可自托管、单一运维人员	零依赖、集成 journald、代码精简	无严重级别筛选、无法按模块控制
loguru	CLI 工具、脚本、本地服务	零配置、内置轮转、开发体验出色	外部依赖、并非真正通过架构强制约束
structlog	生产服务、OTel 集成	ContextVars、处理器链、原生支持 OTel	学习曲线较陡
stdlib	LaunchAgent 守护进程、要求零依赖	无依赖、Python 3.14 merge_extra	样板代码较多、没有结构化默认设置
Logfire	AI/LLM 可观测性、Pydantic 应用	基于 OTel 构建、令牌/成本跟踪、SQL	依赖 SaaS、生态系统较新

---

首选：轻量级模式（零依赖）

适用于：少于 5 个 systemd 服务、单台服务器、单一运维人员。已由 relay-monitor 在生产环境中经过实战检验。

此模式采用双通道架构：

• 通道 1：print(flush=True) → systemd journald（运维日志，便于人工阅读）
• 通道 2：仅追加的 JSONL 文件（结构化遥测数据，机器可读）

这符合 12-Factor App 的“将日志视为事件流”原则。journald 负责运维工作（轮转、筛选、元数据），JSONL 文件则用于记录领域遥测数据，以便进行事后分析。

架构：三个关注点分离

关注点	机制	目的	生命周期
运维日志	print() → journald	人工调试、journalctl -u service -f	由 journald 管理（自动轮转）
遥测数据	JSONL 文件（telemetry.jsonl）	结构化审计轨迹、AI/LLM 分析	仅追加，按大小轮转
状态恢复	WAL 文件（可选）	不可逆操作的崩溃恢复	临时文件，成功后删除

完整的轻量级示例

"""Append-only JSONL telemetry logger with size-based rotation.

Zero external dependencies. Works with systemd journald for ops logging
and a separate JSONL file for structured machine-readable telemetry.
"""

import json
from datetime import datetime, timezone
from pathlib import Path

TELEMETRY_PATH = Path(__file__).parent / "telemetry.jsonl"
MAX_SIZE = 10 * 1024 * 1024  # 10 MB
BACKUP_COUNT = 3             # Keep 3 rotated backups (~30MB total)


def log_event(event_type: str, data: dict) -> None:
    """Append a structured JSON line to telemetry.jsonl."""
    entry = {
        "ts": datetime.now(timezone.utc).isoformat(),
        "type": event_type,
        **data,
    }
    line = json.dumps(entry, separators=(",", ":")) + "\n"

    try:
        try:
            if TELEMETRY_PATH.stat().st_size > MAX_SIZE:
                _rotate()
        except FileNotFoundError:
            pass

        with open(TELEMETRY_PATH, "a") as f:
            f.write(line)
    except OSError as e:
        # Fallback to stderr (captured by journald)
        print(f"[telemetry] write failed: {e}", file=__import__("sys").stderr, flush=True)


def _rotate() -> None:
    """Rotate telemetry files: .jsonl → .jsonl.1 → .jsonl.2 → .jsonl.3"""
    for i in range(BACKUP_COUNT, 1, -1):
        src = TELEMETRY_PATH.with_suffix(f".jsonl.{i - 1}")
        dst = TELEMETRY_PATH.with_suffix(f".jsonl.{i}")
        if src.exists():
            dst.unlink(missing_ok=True)
            src.rename(dst)
    backup = TELEMETRY_PATH.with_suffix(".jsonl.1")
    backup.unlink(missing_ok=True)
    TELEMETRY_PATH.rename(backup)


# === Ops logging (goes to journald via stdout) ===

def log(msg: str) -> None:
    """Human-readable operational log line. Captured by journald."""
    ts = datetime.now(timezone.utc).strftime("%H:%M:%S")
    print(f"[{ts}] {msg}", flush=True)

用法：

# Operational (human reads via journalctl -u myservice -f)
log("Refreshing token for account X")
log("Switch: account A → account B (reason: 5h breach)")

# Telemetry (machine reads via jq/DuckDB/Claude Code)
log_event("token_refresh", {"account": "X", "expires_in_h": 8.0, "token_fp": "abc12345"})
log_event("account_switch", {"from": "A", "to": "B", "reason": "5h_breach"})

安全性：令牌指纹识别（而非正则表达式脱敏）

切勿让密钥经过日志管道。 仅记录不可逆片段：

def _token_fingerprint(token: str) -> str:
    """Extract uniquely identifiable chars from a token's mid-section.

    The prefix (sk-ant-oat01-) and suffix (...AA) are common across tokens.
    Chars 14-22 (after the prefix) are the most unique per-token.
    Middle-slice avoids leaking type-prefix metadata that prefix-based
    approaches expose.
    """
    if len(token) > 25:
        return token[14:22]
    return token[:8] if token else ""

# Usage: log the fingerprint, never the token
log_event("token_refresh", {"account": name, "token_fp": _token_fingerprint(token)})

为何这比正则表达式脱敏过滤器更好：

方式	安全性	维护成本	失效模式
令牌指纹识别（仅记录一段）	密钥永远不会进入日志管道	零——适用于任何令牌格式	不会失效——无需脱敏
正则表达式脱敏过滤器	密钥会经过管道，仅在输出时过滤	新增令牌格式时必须更新正则表达式	静默漏检 = 密钥出现在日志中

这符合 OWASP Logging Cheat Sheet 的要求：“确保日志条目中不包含任何敏感数据。” AWS、Stripe、GitHub 等主要平台都使用独立的非敏感标识符或仅显示部分令牌——绝不会依赖正则表达式清理完整令牌。

正则表达式过滤器仍可作为纵深防御的兜底措施，但不应作为主要控制手段。

将健康端点用作可观测性工具

对于小型部署，丰富的 JSON 健康端点可以替代日志聚合：

@app.get("/api/status")
def status():
    """White-box monitoring — current state on demand."""
    return {"active_account": ..., "accounts": [...], "polled_at": ...}

@app.get("/api/vault-health")
def vault_health():
    """Token health for all accounts."""
    return {name: {"status": "healthy", "expires_in": "7.5h", ...} for ...}

这是 Health Endpoint Monitoring Pattern（Microsoft Azure Architecture Center）/ Health Check API Pattern（microservices.io）。仪表板本身就是监控工具——无需 Grafana/Prometheus。

当服务以结构化 JSON 提供自身运行状态时，你可以获得：

• 实时当前状态（不会因日志采集管道而延迟）
• 零基础设施（无需日志传输器、存储或查询引擎）
• 可供 AI 解析（Claude Code 可以直接执行 curl 并分析）

使用 FOSS 命令行工具进行事后分析

无需日志聚合技术栈。这些单二进制文件工具可以直接处理 JSONL：

# DuckDB — SQL analytics on JSONL (most powerful)
duckdb -c "SELECT type, count(*) FROM read_json_auto('telemetry.jsonl') GROUP BY 1 ORDER BY 2 DESC"

# jq — ad-hoc JSON filtering
jq 'select(.type == "token_refresh")' telemetry.jsonl

# journalctl — already exports JSONL natively
journalctl -u ccmax-switcher -o json --since "1h ago" | jq 'select(.PRIORITY == "3")'

# lnav — interactive terminal log viewer with SQL
lnav telemetry.jsonl

# llm (Simon Willison) — pipe to LLM for AI post-mortem
journalctl -u myservice --since "2h ago" --priority=err -o json | llm "analyze root cause"

何时升级为更轻量方案之外的配置

出现以下任一情况时，升级到 loguru/structlog：

• 跨多台主机的服务数 > 5（需要使用跟踪 ID 进行关联）
• 持续事件数 > 10 次/秒（需要异步接收器、orjson）
• 多个运维人员需要按模块筛选日志级别
• 合规要求规定必须使用带签名的结构化审计跟踪
• 容器/K8s 部署（stdout JSON 是标准做法）

---

功能齐全：Loguru + JSONL 模式

适用于可从日志库中获益的命令行工具、脚本和服务：

日志轮转（本地/命令行应用务必配置）

from loguru import logger

logger.add(
    log_path,
    rotation="10 MB",
    retention="7 days",
    compression="gz"
)

# stdlib alternative (zero-dep)
from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler(
    log_path,
    maxBytes=100 * 1024 * 1024,  # 100MB
    backupCount=5
)

容器/无服务器应用：完全跳过文件轮转。将 JSON 日志写入 stdout/stderr，由容器运行时负责采集和轮转。

JSONL 格式（机器可读）

# One JSON object per line - jq-parseable
{"timestamp": "2026-01-14T12:45:23.456Z", "level": "info", "message": "..."}

文件扩展名：始终使用 .jsonl（不要使用 .json 或 .log）

性能：对于 >10k 条记录/秒的场景，使用 orjson 替代 json.dumps()：

import orjson

def json_formatter(record) -> str:
    log_entry = { ... }
    return orjson.dumps(log_entry).decode()

正则表达式脱敏（纵深防御）

将其作为令牌指纹识别的兜底措施，而非主要控制手段：

import re

REDACT_PATTERNS = [
    (re.compile(r'AKIA[0-9A-Z]{16}'), '[REDACTED_AWS_KEY]'),
    (re.compile(r'sk-[a-zA-Z0-9]{48}'), '[REDACTED_API_KEY]'),
    (re.compile(r'(?i)bearer\s+[a-zA-Z0-9._~+/=-]+'), '[REDACTED_BEARER]'),
]

def redact_filter(record):
    for pattern, replacement in REDACT_PATTERNS:
        record["message"] = pattern.sub(replacement, record["message"])
    return True

logger.add(sink, filter=redact_filter)

关闭时始终刷新队列中的消息

import asyncio
from loguru import logger

async def main():
    logger.add("app.jsonl", enqueue=True)
    await logger.complete()

asyncio.run(main())
# Sync: logger.remove()

完整的 Loguru + JSONL 示例

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.14"
# dependencies = ["loguru", "orjson"]
# ///

import re
import sys
from pathlib import Path
from uuid import uuid4

import orjson
from loguru import logger

REDACT_PATTERNS = [
    (re.compile(r'AKIA[0-9A-Z]{16}'), '[REDACTED_AWS_KEY]'),
    (re.compile(r'sk-[a-zA-Z0-9]{48}'), '[REDACTED_API_KEY]'),
]


def json_formatter(record) -> str:
    log_entry = {
        "timestamp": record["time"].strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
        "level": record["level"].name.lower(),
        "component": record["function"],
        "operation": record["extra"].get("operation", "unknown"),
        "operation_status": record["extra"].get("status", None),
        "trace_id": record["extra"].get("trace_id"),
        "message": record["message"],
        "context": {k: v for k, v in record["extra"].items()
                   if k not in ("operation", "status", "trace_id", "metrics")},
        "metrics": record["extra"].get("metrics", {}),
        "error": None
    }

    if record["exception"]:
        exc_type, exc_value, _ = record["exception"]
        log_entry["error"] = {
            "type": exc_type.__name__ if exc_type else "Unknown",
            "message": str(exc_value) if exc_value else "Unknown error",
        }

    return orjson.dumps(log_entry).decode()


def redact_filter(record):
    for pattern, replacement in REDACT_PATTERNS:
        record["message"] = pattern.sub(replacement, record["message"])
    return True


def setup_logger(app_name: str, log_dir: Path | None = None):
    logger.remove()
    logger.add(sys.stderr, format=json_formatter, filter=redact_filter, level="INFO")
    if log_dir is not None:
        log_dir.mkdir(parents=True, exist_ok=True)
        logger.add(
            str(log_dir / f"{app_name}.jsonl"),
            format=json_formatter,
            filter=redact_filter,
            rotation="10 MB",
            retention="7 days",
            compression="gz",
            level="DEBUG"
        )
    return logger

语义字段参考

字段	类型	用途
timestamp / ts	ISO 8601	事件排序（毫秒精度为最低要求）
level / type	string	严重性或事件类型
component / svc	string	模块、函数或服务名称
operation	string	正在执行的操作
operation_status	string	已开始／成功／失败／已跳过
trace_id	UUID4 or OTel	关联 ID（生产服务使用 OTel trace ID）
message	string	人类可读的描述
context	object	特定于操作的元数据
metrics	object	定量数据（计数、持续时间）
error	object/null	失败时的异常详情

相关资源

• 健康端点监控模式 - Microsoft Azure 架构中心
• OWASP 日志记录速查表 - 安全最佳实践
• 预写日志模式 - Martin Fowler
• DuckDB JSON 支持 - 对 JSONL 进行 SQL 分析
• lnav - 终端日志文件浏览器，支持 SQL
• llm CLI - 将日志传递给大模型进行分析
• structlog 文档 - 面向生产服务的结构化日志记录
• Pydantic Logfire - 基于 OTel 构建的 AI／大模型可观测性工具
• Langfuse - 开源大模型可观测性工具（可自托管）

应避免的反模式

1. 无界日志 - 始终配置轮转（本地）或标准输出（容器）
2. 记录完整密钥 - 使用令牌指纹；正则表达式脱敏是兜底措施，而非首选方案
3. 为少于 5 个低流量服务添加 loguru／structlog - print + JSONL 已足够；依赖并非没有代价
4. 裸用 except 且不记录日志 - 捕获特定异常并记录日志
5. 静默失败 - 抑制错误前先记录日志
6. 使用 enqueue=True 却不调用 logger.complete() - 关闭时会静默丢失日志
7. 将 enqueue=True 与慢速接收端搭配使用 - 会导致内存无限增长
8. 以超过每秒 10k 个事件的速率使用 json.dumps() - 使用 orjson 可提升 2–10 倍速度
9. 在 OTel 服务中使用 UUID4 trace ID - 使用由 OTel 传播的 trace ID
10. 为少于 5 个服务使用 Prometheus／Grafana - 健康端点 + Uptime Kuma 已足够
11. 混淆 WAL 与遥测 - WAL 用于崩溃恢复（临时），遥测用于审计（永久）

---

故障排查

问题	原因	解决方案
找不到 loguru	未安装	运行 uv add loguru
未显示日志	日志级别错误	排查时将级别设为 DEBUG
日志轮转未生效	缺少轮转配置	向 logger.add() 添加 rotation 参数
JSONL 解析错误	日志行格式错误	检查是否存在未转义的特殊字符
enqueue=True 时发生 OOM	内部队列无界	监控 RSS；使用 structlog 或避免使用速度较慢的接收器
关闭时丢失日志	缺少 logger.complete()	调用 await logger.complete() 或 logger.remove()
JSONL 序列化速度慢	高流量时使用标准库 json	改用 orjson.dumps().decode()
日志中包含密钥	未进行指纹处理	记录令牌片段，而非完整值
journald 未捕获输出	缺少刷新操作	使用 print(..., flush=True) 或 PYTHONUNBUFFERED=1
服务崩溃时没有告警	缺少外部监控	添加 Uptime Kuma 或 Gatus，以轮询健康检查端点

执行后反思

此技能执行完成后，请在结束前检查：

1. 命令是否成功？ — 如果没有，请修复导致失败的指令或错误表。
2. 参数或输出是否发生变化？ — 如果底层工具的接口有所变动，请更新 Usage 示例和 Parameters 表，使其保持一致。
3. 是否需要变通处理？ — 如果你不得不临时变通（使用不同标志、增加额外步骤），请更新此 SKILL.md，以便下次调用无需重复采用相同的变通方案。

仅当问题确实存在且可复现时才更新，不要基于猜测更新。