python-observability

出典の明記：このスキルの内容は wshobson/agents （MIT License, Copyright (c) 2024 Seth Hobson）を元にしています。完全なライセンス条文は同じディレクトリの LICENSE.txt をご覧ください。 使用、変更、再配布の際は、該当する MIT ライセンス条項に従い、著作権表示を保持してください。

日本語の説明

このスキルは、フルスタック開発に関するベストプラクティスとパターンのガイダンスを提供します。

---

name: python-observability description: Python の可観測性パターン（構造化ログ、メトリクス、分散トレーシングなど）。ログの追加、メトリクス収集の実装、トレーシングの設定、本番システムのデバッグ時に使用します。

Python の可観測性

構造化ログ、メトリクス、トレースを使って Python アプリケーションを計測します。本番環境で問題が発生したとき、新しいコードをデプロイすることなく「何が、どこで、なぜ」起きたのかを把握する必要があります。

このスキルを使用する場面

• アプリケーションへの構造化ログの追加
• Prometheus を使ったメトリクス収集の実装
• サービス間の分散トレーシングの設定
• リクエストチェーン全体への相関 ID の伝播
• 本番環境の問題のデバッグ
• 可観測性ダッシュボードの構築

基本概念

1. 構造化ログ

本番環境では、一貫したフィールドを持つ JSON 形式でログを出力します。機械可読なログを使うと、高度なクエリやアラートを活用できます。ローカル開発では、人が読みやすい形式を検討してください。

2. 4 つのゴールデンシグナル

すべてのサービス境界で、レイテンシ、トラフィック、エラー、飽和度を追跡します。

3. 相関 ID

1 つのリクエストに対して、一意の 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 による構造化ログ

一貫したフィールドを含む 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 として記録しないでください。ユーザーが誤ったパスワードを入力した場合は、ERROR ではなく INFO です。

パターン 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 を伝播させる - すべてのリクエストとログに引き回す
3. 4 つのゴールデンシグナルを追跡する - レイテンシ、トラフィック、エラー、飽和度
4. ラベルのカーディナリティを制限する - メトリクスのラベルに無制限な値を決して使用しない
5. 適切なレベルでログを記録する - ERROR で狼少年にならない
6. コンテキストを含める - ログにユーザー ID、リクエスト ID、操作名を含める
7. コンテキストマネージャーを使用する - 一貫した時間計測とエラー処理
8. 関心事を分離する - 可観測性のコードでビジネスロジックを汚さない
9. 可観測性をテストする - 統合テストでログとメトリクスを検証する
10. アラートを設定する - アラートがなければメトリクスは役に立たない

ワークフロー

Step 1: 要件分析

プロジェクトの要件と技術的な制約を理解します

Step 2: アーキテクチャ設計

システムアーキテクチャとコンポーネントの分割を設計します

Step 3: 実装ガイダンス

具体的な実装の提案とコードパターンを提示します

Step 4: コードレビュー

コード品質とベストプラクティスへの準拠状況をレビューします

Step 5: 最適化の提案

パフォーマンスの最適化と改善案を提示します

注意点

1. 過剰設計

単純な場面で複雑な設計パターンを使用しないでください

2. パフォーマンスの軽視

特に本番環境では、パフォーマンスの最適化を軽視しないでください

3. セキュリティへの配慮

常にセキュリティとデータ保護を考慮してください

4. テストカバレッジ

特に境界条件について、十分なテストカバレッジを確保してください

5. ドキュメントの保守

ドキュメントをコードと同期した状態に保ってください