Python Observability

作者 wshobson46891e7e60da無授權條款收錄於 2026年10月8日更新於 2026年10月8日

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

僅含說明Software Development DevOps & Cloud

AI 產生的概覽

指導為 Python 應用程式加入結構化日誌、指標與分散式追蹤，方便排查正式環境問題。

功能: 提供模式與程式碼範例，用於加入以 structlog 為基礎的結構化 JSON 日誌、一致的日誌欄位、語意化日誌等級，以及跨服務傳遞關聯 ID。也涵蓋指標概念，例如四大黃金訊號與標籤基數控制，以及分散式追蹤的設定。詳細的完整範例放在獨立的參考檔案中。
適用情境: 適用於加入日誌、實作指標收集、設定追蹤、傳遞關聯 ID，或排查正式環境系統問題的情境。目標是讓 Python 服務具備可觀測性，不必重新部署程式碼就能定位正式環境問題。
執行需求: 需要 Python 與 structlog 函式庫；範例中也提到 FastAPI 與 httpx。不隨附指令碼，僅為說明與程式碼片段，另有一個選用參考檔案。

Python Observability

Instrument Python applications with structured logs, metrics, and traces. When something breaks in production, you need to answer "what, where, and why" without deploying new code.

When to Use This Skill

Adding structured logging to applications
Implementing metrics collection with Prometheus
Setting up distributed tracing across services
Propagating correlation IDs through request chains
Debugging production issues
Building observability dashboards

Core Concepts

1. Structured Logging

Emit logs as JSON with consistent fields for production environments. Machine-readable logs enable powerful queries and alerts. For local development, consider human-readable formats.

2. The Four Golden Signals

Track latency, traffic, errors, and saturation for every service boundary.

3. Correlation IDs

Thread a unique ID through all logs and spans for a single request, enabling end-to-end tracing.

4. Bounded Cardinality

Keep metric label values bounded. Unbounded labels (like user IDs) explode storage costs.

Quick Start

python

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)

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)

Fundamental Patterns

Pattern 1: Structured Logging with Structlog

Configure structlog for JSON output with consistent fields.

python

import loggingimport 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 startupconfigure_logging("INFO")logger = structlog.get_logger()

import loggingimport 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 startupconfigure_logging("INFO")logger = structlog.get_logger()

Pattern 2: Consistent Log Fields

Every log entry should include standard fields for filtering and correlation.

python

import structlogfrom contextvars import ContextVar
# Store correlation ID in contextcorrelation_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

import structlogfrom contextvars import ContextVar
# Store correlation ID in contextcorrelation_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

Pattern 3: Semantic Log Levels

Use log levels consistently across the application.

Level	Purpose	Examples
`DEBUG`	Development diagnostics	Variable values, internal state
`INFO`	Request lifecycle, operations	Request start/end, job completion
`WARNING`	Recoverable anomalies	Retry attempts, fallback used
`ERROR`	Failures needing attention	Exceptions, service unavailable

python

# DEBUG: Detailed internal informationlogger.debug("Cache lookup", key=cache_key, hit=cache_hit)
# INFO: Normal operational eventslogger.info("Order created", order_id=order.id, total=order.total)
# WARNING: Abnormal but handled situationslogger.warning(    "Rate limit approaching",    current_rate=950,    limit=1000,    reset_seconds=30,)
# ERROR: Failures requiring investigationlogger.error(    "Payment processing failed",    order_id=order.id,    error=str(e),    payment_provider="stripe",)

# DEBUG: Detailed internal informationlogger.debug("Cache lookup", key=cache_key, hit=cache_hit)
# INFO: Normal operational eventslogger.info("Order created", order_id=order.id, total=order.total)
# WARNING: Abnormal but handled situationslogger.warning(    "Rate limit approaching",    current_rate=950,    limit=1000,    reset_seconds=30,)
# ERROR: Failures requiring investigationlogger.error(    "Payment processing failed",    order_id=order.id,    error=str(e),    payment_provider="stripe",)

Never log expected behavior at ERROR. A user entering a wrong password is INFO, not ERROR.

Pattern 4: Correlation ID Propagation

Generate a unique ID at ingress and thread it through all operations.

python

from contextvars import ContextVarimport uuidimport 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 examplefrom 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

from contextvars import ContextVarimport uuidimport 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 examplefrom 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

Propagate to outbound requests:

python

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

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

Detailed worked examples and patterns

Detailed sections (starting with ## Advanced Patterns) live in references/details.md. Read that file when the navigation summary above is insufficient.

Best Practices Summary

Use structured logging - JSON logs with consistent fields
Propagate correlation IDs - Thread through all requests and logs
Track the four golden signals - Latency, traffic, errors, saturation
Bound label cardinality - Never use unbounded values as metric labels
Log at appropriate levels - Don't cry wolf with ERROR
Include context - User ID, request ID, operation name in logs
Use context managers - Consistent timing and error handling
Separate concerns - Observability code shouldn't pollute business logic
Test your observability - Verify logs and metrics in integration tests
Set up alerts - Metrics are useless without alerting

來源與署名

來源：wshobson/agents位於plugins/python-development/skills/python-observability提交46891e7

授權條款：無授權條款

內容歸原作者所有。SourceWeft 從公開儲存庫中收錄這些內容。

檢舉或申請下架

Python Observability

Python Observability

When to Use This Skill

Core Concepts

1. Structured Logging

2. The Four Golden Signals

3. Correlation IDs

4. Bounded Cardinality

Quick Start

Fundamental Patterns

Pattern 1: Structured Logging with Structlog

Pattern 2: Consistent Log Fields

Pattern 3: Semantic Log Levels

Pattern 4: Correlation ID Propagation

Detailed worked examples and patterns

Best Practices Summary

來源與署名

更多來自 wshobson/agents 的技能

Uv Package Manager

Web Component Design

Python Type Safety

Visual Design Foundations

Responsive Design

Python Resource Management

更多Software Development技能

Playground

M5 Onboard

Cardputer Buddy

Testing Strategy

Tech Debt

System Design