Back to Details

python-observability-patterns

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Python Observability Patterns

Python 可观测性模式

Logging, metrics, and tracing for production applications.
面向生产环境应用的日志、指标与追踪方案。

Structured Logging with structlog

基于structlog的结构化日志

python
import structlog
python
import structlog

Configure structlog

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

Usage

Usage

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

Output: {"event": "user_created", "user_id": 123, "email": "test@example.com", "level": "info", "timestamp": "2024-01-15T10:00:00Z"}

Output: {"event": "user_created", "user_id": 123, "email": "test@example.com", "level": "info", "timestamp": "2024-01-15T10:00:00Z"}

undefined
undefined

Request Context Propagation

请求上下文传递

python
import structlog
from contextvars import ContextVar
from uuid import uuid4

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

def bind_request_context(request_id: str | None = None):
    """Bind request ID to logging context."""
    rid = request_id or str(uuid4())
    request_id_var.set(rid)
    structlog.contextvars.bind_contextvars(request_id=rid)
    return rid
python
import structlog
from contextvars import ContextVar
from uuid import uuid4

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

def bind_request_context(request_id: str | None = None):
    """Bind request ID to logging context."""
    rid = request_id or str(uuid4())
    request_id_var.set(rid)
    structlog.contextvars.bind_contextvars(request_id=rid)
    return rid

FastAPI middleware

FastAPI middleware

@app.middleware("http") async def request_context_middleware(request, call_next): request_id = request.headers.get("X-Request-ID") or str(uuid4()) bind_request_context(request_id) response = await call_next(request) response.headers["X-Request-ID"] = request_id structlog.contextvars.clear_contextvars() return response
undefined
@app.middleware("http") async def request_context_middleware(request, call_next): request_id = request.headers.get("X-Request-ID") or str(uuid4()) bind_request_context(request_id) response = await call_next(request) response.headers["X-Request-ID"] = request_id structlog.contextvars.clear_contextvars() return response
undefined

Prometheus Metrics

Prometheus 指标

python
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi import FastAPI, Response
python
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi import FastAPI, Response

Define metrics

Define metrics

REQUEST_COUNT = Counter( "http_requests_total", "Total HTTP requests", ["method", "endpoint", "status"] )
REQUEST_LATENCY = Histogram( "http_request_duration_seconds", "HTTP request latency", ["method", "endpoint"], buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0] )
ACTIVE_CONNECTIONS = Gauge( "active_connections", "Number of active connections" )
REQUEST_COUNT = Counter( "http_requests_total", "Total HTTP requests", ["method", "endpoint", "status"] )
REQUEST_LATENCY = Histogram( "http_request_duration_seconds", "HTTP request latency", ["method", "endpoint"], buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0] )
ACTIVE_CONNECTIONS = Gauge( "active_connections", "Number of active connections" )

Middleware to record metrics

Middleware to record metrics

@app.middleware("http") async def metrics_middleware(request, call_next): ACTIVE_CONNECTIONS.inc() start = time.perf_counter()
response = await call_next(request)

duration = time.perf_counter() - start
REQUEST_COUNT.labels(
    method=request.method,
    endpoint=request.url.path,
    status=response.status_code
).inc()
REQUEST_LATENCY.labels(
    method=request.method,
    endpoint=request.url.path
).observe(duration)
ACTIVE_CONNECTIONS.dec()

return response
@app.middleware("http") async def metrics_middleware(request, call_next): ACTIVE_CONNECTIONS.inc() start = time.perf_counter()
response = await call_next(request)

duration = time.perf_counter() - start
REQUEST_COUNT.labels(
    method=request.method,
    endpoint=request.url.path,
    status=response.status_code
).inc()
REQUEST_LATENCY.labels(
    method=request.method,
    endpoint=request.url.path
).observe(duration)
ACTIVE_CONNECTIONS.dec()

return response

Metrics endpoint

Metrics endpoint

@app.get("/metrics") async def metrics(): return Response( content=generate_latest(), media_type="text/plain" )
undefined
@app.get("/metrics") async def metrics(): return Response( content=generate_latest(), media_type="text/plain" )
undefined

OpenTelemetry Tracing

OpenTelemetry 追踪

python
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
python
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

Setup

Setup

provider = TracerProvider() processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="localhost:4317")) provider.add_span_processor(processor) trace.set_tracer_provider(provider)
tracer = trace.get_tracer(name)
provider = TracerProvider() processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="localhost:4317")) provider.add_span_processor(processor) trace.set_tracer_provider(provider)
tracer = trace.get_tracer(name)

Manual instrumentation

Manual instrumentation

async def process_order(order_id: int): with tracer.start_as_current_span("process_order") as span: span.set_attribute("order_id", order_id)
    with tracer.start_as_current_span("validate_order"):
        await validate(order_id)

    with tracer.start_as_current_span("charge_payment"):
        await charge(order_id)
undefined
async def process_order(order_id: int): with tracer.start_as_current_span("process_order") as span: span.set_attribute("order_id", order_id)
    with tracer.start_as_current_span("validate_order"):
        await validate(order_id)

    with tracer.start_as_current_span("charge_payment"):
        await charge(order_id)
undefined

Quick Reference

速查参考

LibraryPurpose
structlogStructured logging
prometheus-clientMetrics collection
opentelemetryDistributed tracing
Metric TypeUse Case
CounterTotal requests, errors
HistogramLatencies, sizes
GaugeCurrent connections, queue size
库用途
structlog结构化日志
prometheus-client指标收集
opentelemetry分布式追踪
指标类型适用场景
Counter请求总数、错误数
Histogram延迟、大小
Gauge当前连接数、队列大小

Additional Resources

额外资源

  • ./references/structured-logging.md
    - structlog configuration, formatters
  • ./references/metrics.md
    - Prometheus patterns, custom metrics
  • ./references/tracing.md
    - OpenTelemetry, distributed tracing
  • ./references/structured-logging.md
    - structlog配置、格式化器
  • ./references/metrics.md
    - Prometheus模式、自定义指标
  • ./references/tracing.md
    - OpenTelemetry、分布式追踪

Assets

资源文件

  • ./assets/logging-config.py
    - Production logging configuration
  • ./assets/logging-config.py
    - 生产环境日志配置

See Also

另请参阅

Prerequisites:
  • python-async-patterns
    - Async context propagation
Related Skills:
  • python-fastapi-patterns
    - API middleware for metrics/tracing
  • python-cli-patterns
    - CLI logging patterns
Integration Skills:
  • python-database-patterns
    - Database query tracing
前置要求：
  • python-async-patterns
    - 异步上下文传递
相关技能：
  • python-fastapi-patterns
    - 用于指标/追踪的API中间件
  • python-cli-patterns
    - CLI日志模式
集成技能：
  • python-database-patterns
    - 数据库查询追踪