python-observability

来源声明：本技能内容源自 wshobson/agents （MIT License，Copyright (c) 2024 Seth Hobson；完整许可文本见同目录 LICENSE.txt）。 使用、修改与再分发须遵守该 MIT 许可条款并保留版权声明。

中文描述

本技能提供全栈开发相关的最佳实践和模式指导。

---

name: python-observability description: Python observability patterns including structured logging, metrics, and distributed tracing. Use when adding logging, implementing metrics collection, setting up tracing, or debugging production systems.

Python 可观测性

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

何时使用此技能

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

核心概念

1. 结构化日志

在生产环境中，以包含一致字段的 JSON 格式输出日志。机器可读的日志支持强大的查询和告警。在本地开发时，可考虑使用易读格式。

2. 四大黄金信号

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

3. 关联 ID

为单个请求的所有日志和追踪片段传递唯一 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 - 在所有请求和日志中传递该 ID
3. 追踪四大黄金信号 - 延迟、流量、错误、饱和度
4. 限制标签基数 - 切勿将无界值用作指标标签
5. 使用恰当的日志级别 - 不要滥用 ERROR
6. 包含上下文 - 在日志中加入用户 ID、请求 ID、操作名称
7. 使用上下文管理器 - 统一处理计时和错误
8. 分离关注点 - 可观测性代码不应污染业务逻辑
9. 测试可观测性 - 在集成测试中验证日志和指标
10. 设置告警 - 没有告警的指标毫无用处

工作流程

步骤 1：需求分析

理解项目需求和技术约束

步骤 2：架构设计

设计系统架构和组件划分

步骤 3：实现指导

提供具体的实现建议和代码模式

步骤 4：代码审查

审查代码质量和最佳实践遵循情况

步骤 5：优化建议

提供性能优化和改进建议

注意事项

1. 过度设计

避免在简单场景中使用复杂的设计模式

2. 忽视性能

不要忽视性能优化，特别是在生产环境中

3. 安全考虑

始终考虑安全性和数据保护

4. 测试覆盖

确保有足够的测试覆盖，特别是边界情况

5. 文档维护

保持文档与代码同步更新