python-observability

Python 可观测性

为 Python 应用添加结构化日志、指标和追踪。当生产环境出现问题时，你需要在无需部署新代码的情况下回答“发生了什么、在哪里、为什么”。

何时使用此技能

• 为应用添加结构化日志
• 使用 Prometheus 实现指标收集
• 跨服务设置分布式追踪
• 在请求链中传播关联 ID
• 调试生产问题
• 构建可观测性仪表盘

核心概念

1. 结构化日志

在生产环境中，将日志以 JSON 格式输出，并使用一致的字段。机器可读的日志可实现强大的查询和告警。对于本地开发，可考虑使用人类可读的格式。

2. 四大黄金信号

为每个服务边界跟踪延迟、流量、错误和饱和度。

3. 关联 ID

为单个请求在所有日志和 span 中贯穿一个唯一 ID，从而实现端到端追踪。

4. 有界基数

保持指标标签值有界。无界标签（如用户 ID）会急剧增加存储成本。

快速开始

import structlog

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
)

logger = structlog.get_logger()
logger.info("Request processed", user_id="123", duration_ms=45)

基本模式

模式 1：使用 Structlog 进行结构化日志

配置 structlog 以输出具有一致字段的 JSON。

import logging
import structlog

def configure_logging(log_level: str = "INFO") -> None:
    """Configure structured logging for the application."""
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer(),
        ],
        wrapper_class=structlog.make_filtering_bound_logger(
            getattr(logging, log_level.upper())
        ),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(),
        cache_logger_on_first_use=True,
    )

# Initialize at application startup
configure_logging("INFO")
logger = structlog.get_logger()

模式 2：一致的日志字段

每条日志条目都应包含用于过滤和关联的标准字段。

import structlog
from contextvars import ContextVar

# Store correlation ID in context
correlation_id: ContextVar[str] = ContextVar("correlation_id", default="")

logger = structlog.get_logger()

def process_request(request: Request) -> Response:
    """Process request with structured logging."""
    logger.info(
        "Request received",
        correlation_id=correlation_id.get(),
        method=request.method,
        path=request.path,
        user_id=request.user_id,
    )

    try:
        result = handle_request(request)
        logger.info(
            "Request completed",
            correlation_id=correlation_id.get(),
            status_code=200,
            duration_ms=elapsed,
        )
        return result
    except Exception as e:
        logger.error(
            "Request failed",
            correlation_id=correlation_id.get(),
            error_type=type(e).__name__,
            error_message=str(e),
        )
        raise

模式 3：语义化日志级别

在整个应用中一致地使用日志级别。

级别	用途	示例
DEBUG	开发诊断	变量值、内部状态
INFO	请求生命周期、操作	请求开始/结束、任务完成
WARNING	可恢复的异常	重试尝试、使用了回退
ERROR	需要关注的失败	异常、服务不可用

# DEBUG: Detailed internal information
logger.debug("Cache lookup", key=cache_key, hit=cache_hit)

# INFO: Normal operational events
logger.info("Order created", order_id=order.id, total=order.total)

# WARNING: Abnormal but handled situations
logger.warning(
    "Rate limit approaching",
    current_rate=950,
    limit=1000,
    reset_seconds=30,
)

# ERROR: Failures requiring investigation
logger.error(
    "Payment processing failed",
    order_id=order.id,
    error=str(e),
    payment_provider="stripe",
)

切勿将预期行为记录为 ERROR。用户输入错误密码是 INFO，而不是 ERROR。

模式 4：关联 ID 传播

在入口处生成唯一 ID，并将其贯穿所有操作。

from contextvars import ContextVar
import uuid
import structlog

correlation_id: ContextVar[str] = ContextVar("correlation_id", default="")

def set_correlation_id(cid: str | None = None) -> str:
    """Set correlation ID for current context."""
    cid = cid or str(uuid.uuid4())
    correlation_id.set(cid)
    structlog.contextvars.bind_contextvars(correlation_id=cid)
    return cid

# FastAPI middleware example
from fastapi import Request

async def correlation_middleware(request: Request, call_next):
    """Middleware to set and propagate correlation ID."""
    # Use incoming header or generate new
    cid = request.headers.get("X-Correlation-ID") or str(uuid.uuid4())
    set_correlation_id(cid)

    response = await call_next(request)
    response.headers["X-Correlation-ID"] = cid
    return response

传播到出站请求：

import httpx

async def call_downstream_service(endpoint: str, data: dict) -> dict:
    """Call downstream service with correlation ID."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            endpoint,
            json=data,
            headers={"X-Correlation-ID": correlation_id.get()},
        )
        return response.json()

详细示例与模式

详细章节（以 ## Advanced Patterns 开头）位于 references/details.md。当上述导航摘要不够用时，请阅读该文件。

最佳实践总结

1. 使用结构化日志 - 具有一致字段的 JSON 日志
2. 传播关联 ID - 贯穿所有请求和日志
3. 跟踪四大黄金信号 - 延迟、流量、错误、饱和度
4. 限制标签基数 - 切勿将无界值用作指标标签
5. 在适当的级别记录日志 - 不要用 ERROR 虚张声势
6. 包含上下文 - 日志中包含用户 ID、请求 ID、操作名称
7. 使用上下文管理器 - 一致的计时和错误处理
8. 分离关注点 - 可观测性代码不应污染业务逻辑
9. 测试你的可观测性 - 在集成测试中验证日志和指标
10. 设置告警 - 没有告警的指标毫无用处