python-observability

用途

Python Observability 用于为 Python 应用添加可观测性能力，帮助在生产环境出现问题时无需部署新代码即可回答“发生了什么、在哪里发生、为什么发生”。适用场景包括：为应用添加结构化日志、使用 Prometheus 实现指标采集、在各服务间设置分布式追踪、在请求链中传播关联 ID、调试生产问题以及构建可观测性仪表盘。

主要能力

结构化日志

以 JSON 形式输出日志，并保持生产环境中字段一致，使日志可被机器读取，从而支持强大的查询与告警。本地开发时可考虑使用人类可读的格式。

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)

四大黄金信号

针对每个服务边界追踪延迟（latency）、流量（traffic）、错误（errors）和饱和度（saturation）。

关联 ID

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

有界基数

保持指标标签值有界。诸如用户 ID 之类的无界标签会急剧增加存储成本。

基本模式

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

为 JSON 输出配置 structlog，并保持字段一致。

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()

高级模式

更详细的工作示例与模式位于 references/details.md。当上述导航摘要不足时，请阅读该文件。

使用方式

该技能以结构化模式与代码示例的形式提供，直接参考并应用于 Python 项目。安装命令、依赖版本与运行环境要求暂未提供。

最佳实践摘要

使用结构化日志——JSON 日志并保持字段一致
传播关联 ID——贯穿所有请求与日志
追踪四大黄金信号——延迟、流量、错误、饱和度
限制标签基数——绝不将无界值用作指标标签
按适当级别记录日志——不要动辄使用 ERROR
包含上下文——日志中的用户 ID、请求 ID、操作名称
使用上下文管理器——保持一致的计时与错误处理
分离关注点——可观测性代码不应污染业务逻辑
测试可观测性——在集成测试中验证日志与指标
设置告警——没有告警的指标毫无用处

限制

资料未提供具体的指标后端、追踪系统、仪表盘工具的配置细节与集成步骤，也未给出安装命令、版本要求、平台限制或收费信息。