---

name: python-observability-ops description: "Python 应用的可观测性模式。触发条件：logging、metrics、tracing、opentelemetry、prometheus、observability、monitoring、structlog、correlation id。" license: MIT compatibility: "Python 3.10+。需要 structlog、opentelemetry-api、prometheus-client。" allowed-tools: "Read Write" metadata: author: claude-mods depends-on: python-async-ops related-skills: python-fastapi-ops, python-cli-ops

Python 可观测性模式

生产应用的日志、指标和追踪。

使用 structlog 的结构化日志

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
)

logger = structlog.get_logger()

# 使用
logger.info("user_created", user_id=123, email="test@example.com")

请求上下文传播

import structlog
from contextvars import ContextVar
from uuid import uuid4

request_id_var: ContextVar[str] = ContextVar("request_id", default="")

async def middleware(request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid4()))
    request_id_var.set(request_id)
    structlog.contextvars.bind_contextvars(request_id=request_id)
    response = await call_next(request)
    structlog.contextvars.unbind_contextvars("request_id")
    return response

Prometheus 指标

from prometheus_client import Counter, Histogram, Gauge, start_http_server

# 计数器
requests_total = Counter(
    "http_requests_total",
    "Total HTTP requests",
    ["method", "endpoint", "status"]
)

# 直方图
request_duration = Histogram(
    "http_request_duration_seconds",
    "Request duration in seconds",
    ["method", "endpoint"],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)

# 仪表
active_connections = Gauge(
    "active_connections",
    "Number of active connections"
)

# 使用
requests_total.labels(method="GET", endpoint="/users", status="200").inc()
with request_duration.labels(method="GET", endpoint="/users").time():
    result = await handle_request()

OpenTelemetry 追踪

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

@tracer.start_as_current_span("process_order")
async def process_order(order_id: str):
    span = trace.get_current_span()
    span.set_attribute("order.id", order_id)
    # 处理逻辑...

快速参考

| 信号 | 库 | 用途 | |------|-----|------| | 日志 | structlog | 结构化事件日志 | | 指标 | prometheus-client | 数值时间序列 | | 追踪 | opentelemetry | 分布式请求追踪 |

附加资源

• ./references/structlog-patterns.md - 高级 structlog 配置
• ./references/otel-patterns.md - OpenTelemetry 完整集成
• ./references/fastapi-observability.md - FastAPI 可观测性中间件