運用で困らない可観測性：FastAPIのログ・メトリクス・トレース実装ガイド――リクエストID、JSONログ、Prometheus、OpenTelemetry、ダッシュボード設計

要約（インバーテッドピラミッド）

• 目的は、FastAPIを本番運用するための「見える化」を整え、障害対応と性能改善を素早く行えるようにすること。
• 最初にやることは、リクエストIDの付与、構造化（JSON）ログ、liveness/readinessヘルスチェック。
• 次に、アプリメトリクス（レイテンシ、スループット、エラー率＝いわゆるRED指標）をPrometheus形式で公開し、アラート条件を用意。
• 最後に、OpenTelemetryで分散トレースを入れ、外部APIやDBまでのレイテンシ内訳を可視化し、ボトルネックを特定する。

誰が得をするか（具体像）

• 学習者Aさん（大学4年、個人開発）
  APIは動くが、たまに遅くなる理由がわからない。ログとメトリクスの基本から、段階的に可視化を整えたい。
• 小規模チームBさん（受託3名）
  障害時の原因切り分けに時間がかかる。リクエストIDで追跡し、エラー率が上がったら自動通知する仕組みが欲しい。
• SaaS開発Cさん（スタートアップ）
  複数サービス構成で遅延が波及。OpenTelemetryで分散トレースを入れ、どこで時間が消えているか即座に掴みたい。

1. 可観測性の最小セット：まずはこれだけ

可観測性は「ログ」「メトリクス」「トレース」の三本柱。最初に導入したい最小セットは次の3つ。

1. リクエストID（相関ID）
2. 構造化（JSON）ログ
3. ヘルスチェック（liveness/readiness）

1.1 リクエストIDを付与するミドルウェア

# app/obsv/request_id.py
import uuid
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import ASGIApp, Receive, Scope, Send

REQUEST_ID_HEADER = "X-Request-ID"

class RequestIdMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: ASGIApp) -> None:
        super().__init__(app)

    async def dispatch(self, request, call_next):
        rid = request.headers.get(REQUEST_ID_HEADER) or str(uuid.uuid4())
        response = await call_next(request)
        response.headers[REQUEST_ID_HEADER] = rid
        # request.stateに載せるとログなどで参照しやすい
        request.state.request_id = rid
        return response

1.2 JSONログ（標準loggingのみ）

# app/obsv/logging.py
import json, logging, sys
from typing import Any

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, Any] = {
            "time": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
        }
        if hasattr(record, "request_id"):
            payload["request_id"] = record.request_id
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False)

def setup_logging(level: str = "INFO"):
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger()
    root.handlers[:] = [handler]
    root.setLevel(level.upper())

1.3 アプリ統合（ヘルスチェック付き）

# app/main.py
from fastapi import FastAPI, Request
from app.obsv.request_id import RequestIdMiddleware, REQUEST_ID_HEADER
from app.obsv.logging import setup_logging
import logging

app = FastAPI(title="Observability-Ready API")
app.add_middleware(RequestIdMiddleware)
setup_logging("INFO")
log = logging.getLogger("app")

@app.get("/health/liveness")
def liveness():
    return {"status": "alive"}

@app.get("/health/readiness")
def readiness():
    # DB疎通などを本番では確認
    return {"status": "ready"}

@app.get("/")
def root(request: Request):
    # リクエストIDをログに載せる
    extra = {"request_id": getattr(request.state, "request_id", None)}
    log.info("hello request", extra=extra)
    return {"ok": True, "request_id": request.headers.get(REQUEST_ID_HEADER)}

判断ポイント

• リクエストIDをヘッダにもレスポンスにも付与し、前段（Nginx/ALB）ログと突き合わせやすくする。
• JSONログは集約・検索が容易。テキストログに戻れなくなるほど便利。

2. メトリクス：RED指標をPrometheusで公開

メトリクスは時系列で動向を観察するための数値群。まずはRED（Rate/Errors/Duration）を揃える。

• Rate：リクエスト数
• Errors：エラー件数（4xx/5xx）
• Duration：レイテンシ分布（ヒストグラム）

2.1 Prometheusエンドポイントの追加

# app/obsv/metrics.py
import time
from typing import Callable
from starlette.middleware.base import BaseHTTPMiddleware
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from starlette.responses import Response

REQ_COUNTER = Counter("http_requests_total", "request count", ["method", "path", "code"])
REQ_ERRORS = Counter("http_requests_error_total", "error count", ["method", "path", "code"])
REQ_LATENCY = Histogram(
    "http_request_duration_seconds",
    "request latency",
    ["method", "path", "code"],
    buckets=[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5]
)

class PrometheusMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next: Callable):
        start = time.perf_counter()
        response = await call_next(request)
        dur = time.perf_counter() - start
        labels = {
            "method": request.method,
            "path": request.url.path,  # 本番はパスパラメータをテンプレ名に正規化推奨
            "code": str(response.status_code),
        }
        REQ_COUNTER.labels(**labels).inc()
        REQ_LATENCY.labels(**labels).observe(dur)
        if response.status_code >= 400:
            REQ_ERRORS.labels(**labels).inc()
        return response

def metrics_endpoint():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

2.2 アプリ統合

# app/main.py（追記）
from app.obsv.metrics import PrometheusMiddleware, metrics_endpoint

app.add_middleware(PrometheusMiddleware)

@app.get("/metrics")
def metrics():
    return metrics_endpoint()

2.3 ダッシュボードの最小式（PromQL）

• レイテンシP90

  histogram_quantile(
    0.90,
    sum(rate(http_request_duration_seconds_bucket[5m])) by (le)
  )
  

• エラー率

  sum(rate(http_requests_error_total[5m])) /
  sum(rate(http_requests_total[5m]))
  

• QPS

  sum(rate(http_requests_total[1m]))
  

判断ポイント

• ルートパラメータ（/users/123）は正規化してカードinality爆発を防ぐ。Starletteのルート名や自前マッピングを利用。
• バケットはP50/P90/P99が安定して読める粒度に調整。

3. アプリログをもっと使える形に：リクエストサマリーログ

レイテンシとステータス、パス、リクエストID、ユーザーID（あれば）を1行でまとめると、検索が楽。

# app/obsv/access_log.py
import time, logging
from starlette.middleware.base import BaseHTTPMiddleware
from typing import Callable

log = logging.getLogger("access")

class AccessLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next: Callable):
        start = time.perf_counter()
        response = await call_next(request)
        dur_ms = int((time.perf_counter() - start) * 1000)
        rid = getattr(request.state, "request_id", None)
        extra = {"request_id": rid}
        log.info(
            f'{request.client.host} {request.method} {request.url.path} '
            f'{response.status_code} {dur_ms}ms',
            extra=extra
        )
        return response

アプリ統合：

# app/main.py（追記）
from app.obsv.access_log import AccessLogMiddleware
app.add_middleware(AccessLogMiddleware)

判断ポイント

• アクセスログは1リクエスト1行で完結する形が良い。
• 迷ったらNCSA風＋追加フィールド（request_id/bytes等）で十分。

4. OpenTelemetryで分散トレース

分散トレースは、リクエストの各処理区間（span）を時系列で可視化する技術。ボトルネック特定に有効。

4.1 依存関係（例）

opentelemetry-distro
opentelemetry-exporter-otlp
opentelemetry-instrumentation-fastapi
opentelemetry-instrumentation-requests

4.2 ワンライナー実行（最小）

opentelemetry-instrument \
  --traces_exporter otlp \
  --service_name fastapi-app \
  uvicorn app.main:app --host 0.0.0.0 --port 8000

環境変数でエンドポイントやサンプリング率を設定（例）：

OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4317
OTEL_TRACES_SAMPLER=parentbased_traceidratio
OTEL_TRACES_SAMPLER_ARG=0.1  # 10%サンプリング

4.3 コードから手動でspanを作る例

# app/obsv/trace.py
from opentelemetry import trace
tracer = trace.get_tracer("app")

def heavy_calc(x: int) -> int:
    with tracer.start_as_current_span("heavy_calc"):
        # 実処理（例）
        s = 0
        for i in range(x):
            s += i * i
        return s

4.4 外部HTTP/DBも可視化

• opentelemetry-instrumentation-requests を入れると、requests経由のHTTP呼び出しに自動的にspanが付与される。
• DB用インスツルメンテーション（SQLAlchemy等）を足すとクエリ時間も追える。

判断ポイント

• 本番はサンプリングを使う（固定率、またはテールサンプリング）。
• request_idをトレースIDに揃える必要はないが、ログ⇔トレースの相互検索導線は用意する。

5. ダッシュボードとアラート：見るべき指標

見る頻度が高いのは次のセット。

• RED指標：QPS、エラー率、P90レイテンシ
• リソース：CPU、メモリ、GC回数、FD数
• DB：接続数、スロークエリ件数、クエリP95
• 外部API：呼び出し失敗率、リトライ回数
• キュー（Celery等）：待ち行列長、遅延、失敗率

アラート例

• エラー率 > 2% が5分継続
• P90 > 300ms が10分継続
• readinessが失敗（=ローリング時の新バージョン停止）

判断ポイント

• 単発のスパイクで鳴り続けないよう、持続時間を設定。
• アラートは少数精鋭から始めて、誤検知を減らす。

6. 例：一気通貫の雛形（最小構成）

# app/main.py（まとめ）
from fastapi import FastAPI, Request
from app.obsv.request_id import RequestIdMiddleware
from app.obsv.logging import setup_logging
from app.obsv.metrics import PrometheusMiddleware, metrics_endpoint
from app.obsv.access_log import AccessLogMiddleware
import logging

setup_logging("INFO")
app = FastAPI(title="Obs API")
app.add_middleware(RequestIdMiddleware)
app.add_middleware(PrometheusMiddleware)
app.add_middleware(AccessLogMiddleware)
log = logging.getLogger("app")

@app.get("/health/liveness")
def liveness():
    return {"status": "alive"}

@app.get("/health/readiness")
def readiness():
    return {"status": "ready"}

@app.get("/metrics")
def metrics():
    return metrics_endpoint()

@app.get("/calc")
def calc(n: int, request: Request):
    s = sum(i*i for i in range(n))
    log.info("calc done", extra={"request_id": getattr(request.state, "request_id", None)})
    return {"n": n, "sum": s}

7. 運用の工夫：本番で効く小さな設計

• ログのPII（個人情報）を記録しない。メッセージとIDのみにする。
• ログレベルはINFO起点、例外はERROR、想定内の検証失敗はWARN。
• ステージングと本番でサンプリング率・ログ詳細度を変える。
• 重要なビジネスイベント（注文作成など）は監査ログとメトリクスの両方に記録。
• デプロイ時はreadinessでの外形監視を活用し、ローリングの踏み抜きを防ぐ。

8. セキュリティとプライバシー

• トレースやログにトークンやパスワードを載せない。ヘッダログはマスク。
• request/response本文の自動記録は危険（PII混入）。サマリーのみにする。
• メトリクスにも個人特定情報を含めない（ラベル設計に注意）。
• 監視UI（Grafana/Jaeger等）は認証とネットワーク制限を必須に。

9. よくある落とし穴と回避策

10. 対象読者別サンプル（導入プラン）

• 個人開発：
  1. リクエストID＋JSONログ → 2) /metrics → 3) P90/エラー率だけのダッシュボード
• 小規模チーム：
  1. 監視基盤（Prometheus/Grafana） → 2) アラート2本（エラー率/P90） → 3) トレース10%サンプリング
• 成長中SaaS：
  1. サービス横断の相関ID方針 → 2) 主要ドメインのカスタムspan → 3) ビジネスKPIメトリクスの可視化

11. ロードマップ（段階導入）

1. 最小セット：リクエストID、JSONログ、ヘルスチェック。
2. メトリクス：RED指標をPrometheusで公開し、簡易ダッシュボード。
3. アラート：エラー率・P90の2本から開始。
4. トレース：OpenTelemetryを10%サンプリングで導入、主要外部APIとDBを可視化。
5. チューニング：パス正規化・ラベル整理、ダッシュボード洗練、ビジネスイベントの監査ログ整備。

12. 参考リンク

• FastAPI
  • FastAPI Official Docs
  • Middleware
• Starlette
  • Starlette Middleware
• Prometheus / Grafana
  • prometheus_client (Python)
  • Prometheus Docs
  • Grafana Docs
• OpenTelemetry
  • OpenTelemetry Python
  • Instrumentation for FastAPI
  • OTLP Exporter
• ログ
  • Python logging

まとめ

• 可観測性は「最小セット」から始めるのが成功の近道。リクエストID、JSONログ、ヘルスチェックをまず整える。
• 次に、PrometheusでRED指標を公開し、P90とエラー率のダッシュボード・アラートを用意する。
• 最後に、OpenTelemetryで分散トレースを取り入れ、外部APIやDBの遅延内訳を把握して優先度高く対策する。
• カーディナリティの管理、PIIの取り扱い、アラートのメンテを怠らず、チームで「見える化」を共通言語にする。運用が落ち着き、改善サイクルが速くなる。