FastAPIの可観測性入門：ログ、メトリクス、トレースの実装と運用

FastAPIでAPIを公開した後に必要になるのが、遅延やエラーの原因を追える仕組みです。

可観測性とは、システムが出す情報から内部の状態を調べられる性質を指します。

最初から大規模な監視基盤を作る必要はありません。

リクエストID、JSONログ、ヘルスチェックを整え、次にメトリクス、最後に分散トレースを加えると、導入効果を確かめながら範囲を広げられます。

ログ、メトリクス、トレースの役割

可観測性の三要素は、同じ情報を別形式で保存する仕組みではありません。

それぞれが答える問いを分けると、何を実装すべきか判断しやすくなります。

要素	答えられる問い	主な識別情報
ログ	その時、何が起きたか	時刻、ログレベル、リクエストID、処理結果
メトリクス	どの程度の頻度や遅さで起きているか	ルート、HTTPメソッド、ステータスコード
トレース	一つの処理がどこを通り、どこで時間を使ったか	トレースID、スパンID、サービス名

ログで個別事象を調べ、メトリクスで異常の広がりを把握し、トレースで処理経路をたどるという役割分担です。

最初に整える三つの仕組み

• リクエストID：一つのHTTPリクエストに付ける識別子です。
• 構造化ログ：項目名と値を持つJSON形式のログです。
• ヘルスチェック：プロセスが動いているか、リクエストを受けられるかを判定するエンドポイントです。

リクエストIDを先に設定する

リクエストIDは、API本体の処理を呼ぶ前にrequest.stateへ保存します。

API本体の処理後に保存すると、エンドポイントや内側のミドルウェアは生成されたIDを参照できません。

# ​app/obsv/request_id.py
import ​uuid
from ​starlette.middleware.base ​import ​BaseHTTPMiddleware

REQUEST_ID_HEADER ​= ​"X-Request-ID"

class ​RequestIdMiddleware(BaseHTTPMiddleware):
 ​ ​ ​ ​async ​def ​dispatch(self, ​request, ​call_next):
 ​ ​ ​ ​ ​ ​ ​ ​request_id ​= ​request.headers.get(REQUEST_ID_HEADER) ​or ​str(uuid.uuid4())
 ​ ​ ​ ​ ​ ​ ​ ​request.state.request_id ​= ​request_id

 ​ ​ ​ ​ ​ ​ ​ ​response ​= ​await ​call_next(request)
 ​ ​ ​ ​ ​ ​ ​ ​response.headers[REQUEST_ID_HEADER] ​= ​request_id
 ​ ​ ​ ​ ​ ​ ​ ​return ​response

この順序なら、アプリケーションのログとレスポンスヘッダーに同じIDを載せられます。

FastAPIの公式ドキュメントにも、リクエスト処理の前後で動作するミドルウェアの基本が示されています。

検索できるJSONログを出す

JSONログでは、後から絞り込みたい項目を独立したフィールドにします。

次のフォーマッターは、アクセスログで使う項目だけを明示的に出力します。

# ​app/obsv/logging.py
import ​json
import ​logging
import ​sys

LOG_FIELDS ​= ​(
 ​ ​ ​ ​"request_id",
 ​ ​ ​ ​"method",
 ​ ​ ​ ​"path",
 ​ ​ ​ ​"status_code",
 ​ ​ ​ ​"duration_ms",
)

class ​JsonFormatter(logging.Formatter):
 ​ ​ ​ ​def ​format(self, ​record: ​logging.LogRecord) ​-> ​str:
 ​ ​ ​ ​ ​ ​ ​ ​payload ​= ​{
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"time": ​self.formatTime(record, ​"%Y-%m-%dT%H:%M:%S"),
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"level": ​record.levelname,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"logger": ​record.name,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"message": ​record.getMessage(),
 ​ ​ ​ ​ ​ ​ ​ ​}
 ​ ​ ​ ​ ​ ​ ​ ​for ​field ​in ​LOG_FIELDS:
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​if ​hasattr(record, ​field):
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​payload[field] ​= ​getattr(record, ​field)
 ​ ​ ​ ​ ​ ​ ​ ​if ​record.exc_info:
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​payload["exception"] ​= ​self.formatException(record.exc_info)
 ​ ​ ​ ​ ​ ​ ​ ​return ​json.dumps(payload, ​ensure_ascii=False)

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

ログの文章へ値を埋め込むだけでは、ステータスコードや処理時間を条件に検索しにくくなります。

フィールドを固定すると、ログ収集先が変わっても検索条件を保ちやすくなります。

一つのリクエストを一行にまとめる

アクセスログには、調査に使う最小限の情報を一行で記録します。

クライアントIPやリクエスト本文は個人情報を含む可能性があるため、この例には入れていません。

# ​app/obsv/access_log.py
import ​logging
import ​time
from ​starlette.middleware.base ​import ​BaseHTTPMiddleware

log ​= ​logging.getLogger("access")

class ​AccessLogMiddleware(BaseHTTPMiddleware):
 ​ ​ ​ ​async ​def ​dispatch(self, ​request, ​call_next):
 ​ ​ ​ ​ ​ ​ ​ ​started ​= ​time.perf_counter()
 ​ ​ ​ ​ ​ ​ ​ ​response ​= ​await ​call_next(request)
 ​ ​ ​ ​ ​ ​ ​ ​duration_ms ​= ​round((time.perf_counter() ​- ​started) ​* ​1000, ​1)

 ​ ​ ​ ​ ​ ​ ​ ​route ​= ​request.scope.get("route")
 ​ ​ ​ ​ ​ ​ ​ ​path ​= ​getattr(route, ​"path", ​"__unmatched__")
 ​ ​ ​ ​ ​ ​ ​ ​log.info(
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"request_completed",
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​extra={
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"request_id": ​getattr(request.state, ​"request_id", ​None),
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"method": ​request.method,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"path": ​path,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"status_code": ​response.status_code,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"duration_ms": ​duration_ms,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​},
 ​ ​ ​ ​ ​ ​ ​ ​)
 ​ ​ ​ ​ ​ ​ ​ ​return ​response

livenessとreadinessを分ける

livenessはプロセスが応答できるか、readinessはリクエストを受けてもよい状態かを判定します。

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

@app.get("/health/readiness")
def ​readiness():
 ​ ​ ​ ​# ​本番では、サービスの受け入れ可否に必要な依存先を確認する
 ​ ​ ​ ​return ​{"status": ​"ready"}

常にreadyを返すサンプルは配線確認にしか使えません。

本番では、停止中のインスタンスへ新しい通信を送らないための条件を決めます。

コンテナ環境での扱いは、FastAPIとKubernetesのヘルスチェック設計も参考になります。

ミドルウェアの順序をそろえる

複数のミドルウェアでは、最後に追加したものが最も外側で動きます。

アクセスログからリクエストIDを参照するには、リクエストID用ミドルウェアを最後に追加します。

from ​fastapi ​import ​FastAPI, ​Request
from ​app.obsv.access_log ​import ​AccessLogMiddleware
from ​app.obsv.logging ​import ​setup_logging
from ​app.obsv.metrics ​import ​PrometheusMiddleware, ​metrics_endpoint
from ​app.obsv.request_id ​import ​RequestIdMiddleware

setup_logging("INFO")
app ​= ​FastAPI(title="Observable ​API")

app.add_middleware(AccessLogMiddleware)
app.add_middleware(PrometheusMiddleware)
app.add_middleware(RequestIdMiddleware)

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

@app.get("/")
def ​root(request: ​Request):
 ​ ​ ​ ​return ​{
 ​ ​ ​ ​ ​ ​ ​ ​"ok": ​True,
 ​ ​ ​ ​ ​ ​ ​ ​"request_id": ​request.state.request_id,
 ​ ​ ​ ​}

リクエスト時はリクエストID、メトリクス、アクセスログ、エンドポイントの順に進み、レスポンス時は逆順に戻ります。

PrometheusでRED指標を計測する

RED指標は、HTTPサービスの状態をリクエスト数、エラー、処理時間で確認する考え方です。

指標	意味	確認例
Rate	一定時間あたりのリクエスト数	急増、急減、通常時との差
Errors	失敗と数えるレスポンスの割合	4xxと5xxをどう扱うか
Duration	処理時間の分布	P50、P90、P99

このサンプルは4xxと5xxをエラーとして数えます。

入力誤りによる4xxをサービス障害から除外したい場合は、運用上の定義に合わせてカウンターを分けます。

高カーディナリティを避ける

カーディナリティは、ラベル値の組み合わせ数です。

/users/123のような実URLをそのままラベルにすると、利用者やデータの数だけ時系列が増えます。

/users/{user_id}のようなルート定義へ正規化し、ユーザーIDやリクエストIDをメトリクスのラベルにしない設計が必要です。

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

REQUESTS ​= ​Counter(
 ​ ​ ​ ​"http_requests_total",
 ​ ​ ​ ​"HTTP ​request ​count",
 ​ ​ ​ ​["method", ​"path", ​"code"],
)
ERRORS ​= ​Counter(
 ​ ​ ​ ​"http_requests_error_total",
 ​ ​ ​ ​"HTTP ​error ​count",
 ​ ​ ​ ​["method", ​"path", ​"code"],
)
LATENCY ​= ​Histogram(
 ​ ​ ​ ​"http_request_duration_seconds",
 ​ ​ ​ ​"HTTP ​request ​duration",
 ​ ​ ​ ​["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):
 ​ ​ ​ ​ ​ ​ ​ ​started ​= ​time.perf_counter()
 ​ ​ ​ ​ ​ ​ ​ ​status_code ​= ​500
 ​ ​ ​ ​ ​ ​ ​ ​try:
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​response ​= ​await ​call_next(request)
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​status_code ​= ​response.status_code
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​return ​response
 ​ ​ ​ ​ ​ ​ ​ ​finally:
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​route ​= ​request.scope.get("route")
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​path ​= ​getattr(route, ​"path", ​"__unmatched__")
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​labels ​= ​{
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"method": ​request.method,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"path": ​path,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"code": ​str(status_code),
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​}
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​REQUESTS.labels(**labels).inc()
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​LATENCY.labels(**labels).observe(time.perf_counter() ​- ​started)
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​if ​status_code ​>= ​400:
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ERRORS.labels(**labels).inc()

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

finallyで計測するため、アプリケーションで未処理例外が起きた場合も、このミドルウェアを通った処理は500として記録されます。

/metricsは監視基盤から到達できればよく、インターネットへ無制限に公開しない構成にします。

最初のPromQL

まずはP90レイテンシ、エラー率、QPSの三つがあれば、遅延と失敗と負荷の変化を追えます。

P90は、観測したリクエストの90％が収まる処理時間の目安です。

# ​ルート別のP90レイテンシ
histogram_quantile(
 ​ ​0.90,
 ​ ​sum ​by ​(le, ​path) ​(rate(http_request_duration_seconds_bucket[5m]))
)

# ​全体のエラー率
sum(rate(http_requests_error_total[5m]))
/
sum(rate(http_requests_total[5m]))

# ​全体のQPS
sum(rate(http_requests_total[1m]))

ヒストグラムのバケットは、実測した処理時間の分布を見て調整します。

上限が小さすぎると遅い処理の差が見えず、細かすぎると管理する時系列が増えます。

OpenTelemetryで処理経路をたどる

分散トレースは、一つのリクエストがAPI、外部HTTP、データベースなどを通る流れを時系列で記録する仕組みです。

各処理区間をスパンと呼び、複数のスパンを一つのトレースとして結び付けます。

自動計装を起動する

OpenTelemetryの自動計装では、必要なパッケージを導入したうえで、通常の起動コマンドをopentelemetry-instrumentから実行します。

pip ​install ​opentelemetry-distro ​opentelemetry-exporter-otlp
opentelemetry-bootstrap ​-a ​install

export ​OTEL_SERVICE_NAME=fastapi-app
export ​OTEL_TRACES_EXPORTER=otlp
export ​OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4317
export ​OTEL_TRACES_SAMPLER=parentbased_traceidratio
export ​OTEL_TRACES_SAMPLER_ARG=0.1

opentelemetry-instrument ​\
 ​ ​uvicorn ​app.main:app ​--host ​0.0.0.0 ​--port ​8000

0.1は10％を採取する設定例であり、すべての環境に適した既定値ではありません。

通信量、調査に必要な粒度、保存コストを見ながら変更します。

調べたい処理に手動スパンを付ける

自動計装だけでは処理の意味が分からない箇所に、手動スパンを追加します。

from ​opentelemetry ​import ​trace

tracer ​= ​trace.get_tracer("app")

def ​heavy_calc(x: ​int) ​-> ​int:
 ​ ​ ​ ​with ​tracer.start_as_current_span("heavy_calc"):
 ​ ​ ​ ​ ​ ​ ​ ​total ​= ​0
 ​ ​ ​ ​ ​ ​ ​ ​for ​i ​in ​range(x):
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​total ​+= ​i ​* ​i
 ​ ​ ​ ​ ​ ​ ​ ​return ​total

スパン名には、変動するIDではなく処理の役割を表す名前を使います。

外部HTTPやデータベースの遅延を分離できると、アプリケーション内部と依存先のどちらを先に調べるか判断できます。

外部サービスの失敗を検知した後の設計は、サーキットブレーカとフォールバックで補えます。

ダッシュボードとアラートの組み立て方

ダッシュボードは、利用者への影響から原因候補へ順にたどれる構成にします。

1. QPS、エラー率、P90レイテンシでサービス全体の変化を見る。
2. CPU、メモリ、ファイルディスクリプタ数などのリソースを確認する。
3. データベース、外部API、キューなど依存先の指標を確認する。
4. トレースとリクエストIDを使い、影響を受けた処理を個別に追う。

SLOは、サービスが満たすべき信頼性の目標です。

アラートの閾値はSLOと通常時の実測値から決めます。

「エラー率が2％を超えて5分続く」「P90が300ミリ秒を超えて10分続く」といった値は出発点の例であり、根拠なしに本番へ流用する値ではありません。

単発の揺れで通知が連続しないよう持続時間を設け、対応不要だった通知は条件を見直します。

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

• 認証トークン、パスワード、Cookieをログやスパンへ記録しない。
• リクエスト本文とレスポンス本文を無条件に保存しない。
• ユーザーIDやリクエストIDをPrometheusのラベルへ入れない。
• Grafana、トレース閲覧画面、/metricsへの接続を認証やネットワークで制限する。
• 注文作成などの監査記録と、性能調査用のアクセスログを同じ目的で扱わない。

調査材料を増やすほど安全になるわけではありません。

誰が何の目的で参照するかを決め、必要な項目だけを残します。

よくある不具合と見直し方

症状	原因	見直す点
メトリクスの時系列が増え続ける	実URLや利用者IDをラベルにしている	ルートをテンプレートへ正規化し、識別子を除く
同じ障害のログを追えない	リクエストIDを処理前に設定していない	外側のミドルウェアでIDを設定し、後続処理へ渡す
ログ検索で条件を指定できない	値がメッセージ本文に埋め込まれている	項目名を固定したJSONログへ分ける
readinessが常に成功する	サンプルの固定応答を本番でも使っている	通信を受けられる条件を定義し、実際に判定する
トレースの保存量が多い	すべてのリクエストを採取している	調査目的に合わせてサンプリング方針を決める
通知が多く対応できない	閾値と持続時間が通常時の変動に合っていない	SLO、実測値、対応手順を一緒に見直す

段階導入の手順

1. リクエストID、JSONログ、liveness、readinessを実装する。
2. PrometheusでRED指標を公開し、ダッシュボードで実測値を確認する。
3. SLOと対応手順に基づくアラートを少数から設定する。
4. OpenTelemetryを導入し、外部HTTPやデータベースを含む処理時間を確認する。
5. ログ項目、メトリクスのラベル、サンプリング率、通知条件を定期的に見直す。

個人開発では最初の二段階だけでも、遅いリクエストとエラーの発生傾向を追いやすくなります。

複数サービスを運用するチームでは、サービス間でリクエストIDの受け渡し方とトレースのサービス名をそろえる作業まで計画します。

公開前の確認項目

• 生成したリクエストIDをエンドポイントとアクセスログから参照できるか。
• レスポンスにも同じリクエストIDが返るか。
• URL内のIDがメトリクスのラベルに残っていないか。
• 未処理例外を500として計測できるか。
• readinessが実際の受け入れ可否を表しているか。
• ログとトレースに秘密情報や本文が入っていないか。
• アラートを受けた担当者が次に見る画面と手順を決めているか。

公式資料

• FastAPI公式ドキュメント
• FastAPIのミドルウェア
• Starletteのミドルウェア
• Prometheus ​Python ​Client
• Prometheusの概要
• Grafana公式ドキュメント
• OpenTelemetry ​Python
• OpenTelemetry ​Pythonの計装ライブラリ
• OTLP仕様
• Python ​logging

まとめ

FastAPIの可観測性は、リクエストIDと構造化ログで個別事象を追える状態を作り、Prometheusで全体の傾向を測り、OpenTelemetryで処理経路を調べる順に整えます。

実装後は、ラベル数、秘密情報、ヘルスチェックの判定条件、サンプリング率、アラートの実効性を運用の中で見直します。

この記事に関連する株式会社greedenの取り組み

障害時に追えるログと指標を設計段階から組み込むと、APIは改善を続けやすくなります。株式会社greedenは、要件定義から実装、監視を含む保守運用まで、Webシステム開発を一気通貫で支援します。