FastAPIロギング&監視入門ガイド:構造化ログ・メトリクス・トレーシングで本番運用に強いAPIに育てる
要約(最初に全体像をつかむ)
- FastAPIアプリを「本番運用」に耐えられる形にするには、機能実装だけでなく ログ・メトリクス・ヘルスチェック といった監視まわりを整えることが重要です。
- Python標準の
loggingと FastAPI のミドルウェア/依存関係を組み合わせることで、構造化ログ(JSONログ)やリクエストID付きのログを簡単に実現できます。 - レイテンシ・リクエスト数・エラー率といったメトリクスを収集し、Prometheus 等に渡すことで、Grafana などのダッシュボードで可視化できます。
- ヘルスチェック(/health)やレディネスチェック(/ready)を用意しておくと、コンテナオーケストレーションやロードバランサと連携しやすく、障害時の切り離しも自動化しやすくなります。
- 本記事では、個人開発〜小規模チーム〜SaaS開発チームまでを念頭に、段階的にログ&監視を整えていくためのロードマップと具体的なコード例をまとめてご紹介します。
誰が読んで得をするか(具体的な読者像)
個人開発・学習者さん
- FastAPIで小さなWeb APIやサービスを作っている。
- エラーが起きても「uvicorn のログをとりあえず眺める」くらいで、原因追跡に時間がかかってしまう。
- 構造化ログやメトリクスの話は聞いたことがあるけれど、「どこから触ればいいか」がピンと来ていない。
この方には、まずは 標準 logging+簡単なミドルウェア から始めて、少しずつログを整えていく入り口を一緒にたどっていきます。
小規模チームのバックエンドエンジニアさん
- 3〜5人のチームで FastAPI ベースのサービスを開発・運用していて、時々本番のトラブル対応に追われている。
- 「エラーは出ているけれど、どのユーザーのどのリクエストか分かりづらい」「どのAPIが重いのか感覚でしか把握できていない」と感じている。
- チームとして「ログのルール」「監視項目」を揃え、運用負荷を下げたい。
この方には、リクエストID付きログ・構造化ログ・メトリクス計測・ヘルスチェック を組み合わせた、現実的な監視のひな形が役立つと思います。
SaaS開発チーム・スタートアップのみなさま
- ある程度ユーザーが増え、ピーク時の障害やレイテンシ悪化が事業に直結する。
- マルチインスタンスやコンテナオーケストレーション(Kubernetes 等)環境で FastAPI を運用している、あるいはその予定がある。
- ログ集約(ELK / Loki など)、メトリクス監視(Prometheus / Cloud Monitoring)、トレーシング(OpenTelemetry 等)の基礎を FastAPI 側から整えたい。
この方には、「アプリ側で何を出すべきか」「インフラ側の監視とどう役割分担するか」という観点から、実務に持ち帰りやすい設計ポイントを整理してお伝えします。
アクセシビリティ評価(読みやすさ・理解しやすさ)
- 記事全体は、「要約 → 読者像 → なぜログ・監視が必要か → ロギング基礎 → 構造化ログとリクエストID → メトリクス → ヘルスチェック → ロードマップ」という逆三角形構造で、必要な所だけ読みとばしても流れがつかみやすいようにしています。
- 専門用語(構造化ログ、メトリクス、トレーシング、レイテンシなど)は、初出時に短い補足と一緒に紹介しています。
- コードブロックは短めに区切り、コメントも必要最低限にとどめて、画面リーダー利用者や読み上げ環境でも追いやすいよう配慮しています。
- 中見出し(##)と小見出し(###)を適切に分けることで、見出しだけ流し読みしても記事全体の構造がつかめるようにしています。
技術記事として、読みやすさ・理解のしやすさの面で、WCAG AA 程度を目標としたテキスト構成になっています。
1. なぜログ&監視がそんなに大事なのか
FastAPI は、とても軽快に REST API を作れるフレームワークですが、「動いていること」と「運用しやすいこと」は別問題です。
1.1 バグ調査・障害対応にログは欠かせない
- どのエンドポイントでエラーが出たか
- どんなリクエストパラメータが送られてきたか
- どのユーザーの操作が原因だったか
こういった情報が分からないと、障害調査はどうしても「勘と根性」になってしまいます。
ログをきちんと残しておくことで、原因の切り分けが早くなり、「再発防止策」を検討する余裕も生まれます。
1.2 体感の遅さの正体は「レイテンシ」と「エラー率」
ユーザーが「最近遅い」「たまに落ちる」と感じるとき、多くの場合、
- ある特定のエンドポイントだけレイテンシが高くなっている
- 特定の条件でエラー率が跳ね上がっている
といった現象が起きています。
メトリクスでこれらを「数字」として見られるようにしておけば、
- どのくらいの負荷でどれだけ遅くなるか
- リリース後にエラー率が増えていないか
といった判断を冷静に行えるようになります。
1.3 健全なサービス運用の3要素
本記事では、健全なサービス運用のための3要素として、
- ログ(log)
- メトリクス(metrics)
- ヘルスチェック/トレーシング(health & trace)
に分けて考え、それぞれ FastAPI からどう整えていくかを見ていきますね。
2. Python標準 logging と FastAPI の基本連携
まずは、特別なライブラリを入れなくてもできる範囲から整えていきましょう。
2.1 logging の基本セットアップ
Python には標準で logging モジュールがあります。
FastAPI アプリ用に、次のような設定ファイルを用意しておくと便利です。
# app/core/logging_config.py
import logging
from logging.config import dictConfig
LOGGING = {
"version": 1,
"disable_existing_loggers": False,
# どこに書き出すか(ここではコンソール)
"handlers": {
"default": {
"level": "INFO",
"class": "logging.StreamHandler",
"formatter": "default",
},
},
# ログのフォーマット
"formatters": {
"default": {
"format": "%(asctime)s %(levelname)s [%(name)s] %(message)s",
},
},
# ルートロガー
"root": {
"level": "INFO",
"handlers": ["default"],
},
# uvicorn 関連のロガーも整える
"loggers": {
"uvicorn": {"level": "INFO", "handlers": ["default"], "propagate": False},
"uvicorn.error": {"level": "INFO", "handlers": ["default"], "propagate": False},
"uvicorn.access": {"level": "INFO", "handlers": ["default"], "propagate": False},
"app": {"level": "INFO", "handlers": ["default"], "propagate": False},
},
}
def setup_logging():
dictConfig(LOGGING)
logging.getLogger("app").info("Logging is configured")
FastAPI の起動時にこの設定を読み込みます。
# app/main.py
from fastapi import FastAPI
from app.core.logging_config import setup_logging
import logging
setup_logging()
logger = logging.getLogger("app")
app = FastAPI(title="Logging Example")
@app.on_event("startup")
async def on_startup():
logger.info("Application startup")
@app.get("/ping")
def ping():
logger.info("Ping endpoint called")
return {"message": "pong"}
これだけでも、uvicorn のログとアプリ独自のログを一か所に揃えることができます。
2.2 ロガーの使い分け
実務では、
- モジュールごとに
logger = logging.getLogger(__name__)を定義する - 重要なイベントだけ INFO レベル以上でログを出す
- DEBUG レベルのログは開発・ステージング限定で有効化する
といった運用パターンがよく使われます。
# app/domain/users/service.py
import logging
logger = logging.getLogger(__name__)
def create_user(email: str) -> int:
logger.debug("Creating user %s", email)
# 実際の作成処理...
user_id = 1
logger.info("User created: id=%s, email=%s", user_id, email)
return user_id
3. 構造化ログとリクエストIDで「追いやすいログ」にする
プレーンテキストのログでも役に立ちますが、大規模になると「どのリクエストのログか分からない」「パースしづらい」という悩みが出てきます。
そこで、構造化ログとリクエストIDの出番です。
3.1 構造化ログって何?
構造化ログとは、単なる文字列ではなく「キーと値のセット」としてログを記録する考え方です。
例:
- プレーンログ:
2025-01-01 00:00:00 INFO [app] User created: id=1 email=foo@example.com - 構造化ログ(JSON):
{"time":"2025-01-01T00:00:00Z","level":"INFO","logger":"app.domain.users.service","msg":"User created","user_id":1,"email":"foo@example.com"}
後者であれば、
- ログ集約システムが JSON をパースしやすい
user_id=1のログだけ検索する、といったクエリが書きやすい
というメリットがあります。
3.2 JSON ログ用のフォーマッタ
標準 logging でも、自分で JSON を組み立てるフォーマッタを書けば構造化ログが扱えます。
# app/core/json_formatter.py
import json
import logging
from datetime import datetime, timezone
class JsonFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
log_record = {
"time": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
# extra で渡されたフィールドも含める
if hasattr(record, "request_id"):
log_record["request_id"] = record.request_id
if hasattr(record, "path"):
log_record["path"] = record.path
return json.dumps(log_record, ensure_ascii=False)
これを先ほどの LOGGING 設定に組み込みます。
# logging_config.LOGGING の一部を書き換え
"formatters": {
"json": {
"()": "app.core.json_formatter.JsonFormatter",
},
},
"handlers": {
"default": {
"level": "INFO",
"class": "logging.StreamHandler",
"formatter": "json",
},
},
3.3 リクエストIDを付与するミドルウェア
「このログがどの HTTP リクエストから生まれたか」を追うために、リクエストID(トレースID)を付けておくと便利です。
# app/middleware/request_id.py
import uuid
import logging
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
logger = logging.getLogger(__name__)
REQUEST_ID_HEADER = "X-Request-ID"
class RequestIDMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# 既にヘッダにIDがあればそれを使う(外部のリバースプロキシなどから)
request_id = request.headers.get(REQUEST_ID_HEADER, str(uuid.uuid4()))
# レスポンスヘッダにも乗せて返す
response: Response = await call_next(request)
response.headers[REQUEST_ID_HEADER] = request_id
# 最後にアクセスログ的なものを出す例
logger.info(
"Request completed",
extra={
"request_id": request_id,
"path": request.url.path,
"method": request.method,
"status_code": response.status_code,
},
)
return response
FastAPIアプリに組み込みます。
# app/main.py(抜粋)
from app.middleware.request_id import RequestIDMiddleware
app = FastAPI(title="Logging Example with Request ID")
app.add_middleware(RequestIDMiddleware)
アプリ内で何かログを出すときにも、この request_id を付けておくと、あとでログ集約ツール上で「1リクエスト分のログ」をまとめて追いかけやすくなります。
4. メトリクス:レイテンシ・リクエスト数・エラー率を数値で見る
次は、メトリクスです。
「どれくらい遅くなっているか」「どれくらいエラーが出ているか」を 数値 で見るための仕掛けですね。
4.1 どんなメトリクスを取るべきか
FastAPIのようなAPIサーバでは、特に次の3つが重要になります。
- リクエスト数(RPS: Requests Per Second)
- どれくらいのトラフィックが来ているか
- レイテンシ(遅延時間)
- 特に P95 / P99(上位5% / 1% の遅いリクエスト)
- エラー率
- HTTP 5xx / 4xx がどれくらい出ているか
これに加えて、
- DBクエリの時間・回数
- 外部APIへの呼び出し数・失敗数
なども取れると、ボトルネックの特定がぐっと楽になります。
4.2 Prometheus クライアントでシンプルなメトリクスを導入する
最小構成として、Python 用 Prometheus クライアントを使った例を見てみます。
pip install prometheus-client
# app/core/metrics.py
import time
from prometheus_client import Counter, Histogram
REQUEST_COUNT = Counter(
"http_requests_total",
"Total HTTP requests",
["method", "path", "status_code"],
)
REQUEST_LATENCY = Histogram(
"http_request_duration_seconds",
"HTTP request latency (seconds)",
["method", "path"],
)
ミドルウェアで計測します。
# app/middleware/metrics.py
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
from app.core.metrics import REQUEST_COUNT, REQUEST_LATENCY
class MetricsMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
method = request.method
path = request.url.path
with REQUEST_LATENCY.labels(method=method, path=path).time():
response: Response = await call_next(request)
REQUEST_COUNT.labels(
method=method,
path=path,
status_code=response.status_code,
).inc()
return response
FastAPI に組み込みます。
# app/main.py(抜粋)
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
from fastapi.responses import Response
from app.middleware.metrics import MetricsMiddleware
app.add_middleware(MetricsMiddleware)
@app.get("/metrics")
def metrics():
# Prometheus が /metrics を取得しに来る想定
data = generate_latest()
return Response(content=data, media_type=CONTENT_TYPE_LATEST)
これで、Prometheus や類似ツールから /metrics をスクレイプすることで、FastAPIアプリのメトリクスを可視化できるようになります。
4.3 専用ライブラリを使う方法
もう少し簡単に導入したい場合は、prometheus-fastapi-instrumentator のようなライブラリを使う方法もあります。
pip install prometheus-fastapi-instrumentator
from prometheus_fastapi_instrumentator import Instrumentator
app = FastAPI()
@app.on_event("startup")
async def _startup():
Instrumentator().instrument(app).expose(app)
このようなツールを使うと、メトリクスの定義やミドルウェアを自前で書かなくても、基本的な HTTP メトリクスを一通り取ってくれます。
5. ヘルスチェック&レディネスチェック:壊れたインスタンスを自動で外す
次は、コンテナ時代には欠かせない「ヘルスチェック」です。
5.1 ヘルスチェックとレディネスの違い
-
ヘルスチェック(Liveness)
- 「このプロセスは生きているか?」
- プロセスがハングしていたりクラッシュしていれば、
/healthが失敗し、オーケストレーション側が再起動を試みる。
-
レディネスチェック(Readiness)
- 「このインスタンスはリクエストを受け付けられる状態か?」
- 起動直後でキャッシュがまだ温まっていない、DBに接続できていない、などの場合には一時的にNGを返し、ロードバランサから外す。
5.2 シンプルなヘルスチェック
まずは、生死だけを確認する簡単なエンドポイントです。
# app/api/health.py
from fastapi import APIRouter
router = APIRouter(tags=["health"])
@router.get("/health")
def health():
return {"status": "ok"}
アプリに組み込みます。
# app/main.py(抜粋)
from app.api import health
app.include_router(health.router)
5.3 DB や外部サービスも含めたレディネスチェック
少し発展させて、「DB に最低1回クエリできるか」を見るレディネスチェックを作ってみます。
# app/api/health.py(続き)
from sqlalchemy.orm import Session
from app.deps.db import get_db
from fastapi import Depends, HTTPException, status
@router.get("/ready")
def ready(db: Session = Depends(get_db)):
try:
# 軽いクエリで接続を確認(例:SELECT 1)
db.execute("SELECT 1")
except Exception:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="database not available",
)
return {"status": "ready"}
Kubernetes などでは、
- livenessProbe →
/health - readinessProbe →
/ready
といった形で設定しておくと、「落ちたら再起動」「準備中はトラフィックから外す」といった振る舞いを自動化できます。
6. トレーシングの入口:リクエストごとの流れを追いかける
より高度な運用になると、分散トレーシング(Distributed Tracing)を入れたくなる場面も出てきます。
ここでは本格的な OpenTelemetry などの話には踏み込みすぎず、「FastAPI からどう準備しておくか」の入口だけ触れておきますね。
6.1 トレースIDとスパンのイメージ
- トレースID:あるユーザー操作(1リクエスト)全体を表すID
- スパン:その中の1つの処理(API呼び出し、DBクエリなど)を表す単位
これらを分散トレーシング基盤に送ることで、「この1リクエストにおいて、どこでどれだけ時間を使っているか」を可視化できます。
6.2 まずはリクエストIDをヘッダに載せるところから
先ほどの RequestIDMiddleware を少し拡張して、X-Request-ID をすべての外部API呼び出しにも渡すようにしておくと、将来的にトレーシング基盤を導入したときにも紐づけがしやすくなります。
# app/infra/http_client.py
import httpx
from contextvars import ContextVar
request_id_var: ContextVar[str | None] = ContextVar("request_id", default=None)
async def get_client() -> httpx.AsyncClient:
headers = {}
request_id = request_id_var.get()
if request_id:
headers["X-Request-ID"] = request_id
return httpx.AsyncClient(headers=headers, timeout=10.0)
RequestIDMiddleware 側で、ContextVar に値をセットしておきます。
# app/middleware/request_id.py(抜粋)
from app.infra.http_client import request_id_var
class RequestIDMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
request_id = request.headers.get(REQUEST_ID_HEADER, str(uuid.uuid4()))
request_id_var.set(request_id)
# ...
こうすることで、将来 OpenTelemetry などを導入しても、「アプリ内のトレースID」と「外部サービス側のログ」をヘッダ経由で結びつけやすくなります。
7. ログレベルと運用ルール:何をどこまで出すべきか
ログは「多ければ多いほど良い」というわけではありません。
ログが多すぎると、読みづらいだけでなく、ストレージコストやログ集約サービスの料金も増えてしまいます。
7.1 ログレベルの目安
- DEBUG
- 詳細なデバッグ情報。開発・ステージング用。
- INFO
- 正常系の重要イベント(ユーザー作成、注文確定、バッチ完了など)。本番でも基本オン。
- WARNING
- 想定外だがすぐには致命的でない状況(リトライで成功した外部API、フォールバック利用など)。
- ERROR
- 処理が失敗し、ユーザー影響がある可能性が高いもの。アラートのトリガにもしやすい。
- CRITICAL
- サービス全体に重大な影響があるもの(設定ファイル欠損、起動不能など)。
7.2 セキュリティ・プライバシーへの配慮
ログには、次のような情報を直接書かないよう気をつけてくださいね。
- パスワードやトークン、クレジットカード番号などの機密情報
- フルの個人情報(住所・電話番号など)を、不要にINFOレベルで出し続けること
必要であれば、メールアドレスやユーザーIDを匿名化して出す、あるいはハッシュ化やマスキングを行う、といった工夫も検討しましょう。
8. 読者別ロードマップ:少しずつ整えていくために
最後に、「これからログ&監視を整えていくときの道筋」を、読者像ごとに整理してみます。
8.1 個人開発・学習者さん向けステップ
- Python標準 logging を導入し、モジュールごとにロガーを定義する。
- RequestIDMiddleware を入れて、
X-Request-IDを付けたアクセスログを出してみる。 /healthエンドポイントを用意し、稼働確認をしやすくする。- 余裕があれば、
prometheus-fastapi-instrumentatorを入れて/metricsも試してみる。
8.2 小規模チームのバックエンドエンジニアさん向けステップ
- チームで logging の設定を共有し、ログフォーマットやロガー名のルールを揃える。
- 構造化ログ(JSON)+ RequestID に切り替え、ログ集約ツール(ELK / Loki 等)で検索しやすくする。
- Prometheus などを導入し、レイテンシ・エラー率・RPS をダッシュボード化する。
/readyで DB 接続チェックを行い、オーケストレーション側のヘルスチェックに連携させる。
8.3 SaaS開発チーム・スタートアップ向けステップ
- アプリ・インフラ双方で「監視ポリシー」を整理し、どの層で何を監視するか役割分担を決める。
- 分散トレーシング(OpenTelemetry 等)を検討しつつ、まずはリクエストID・ヘッダ連携でトレースIDの流れを整える。
- メトリクスのアラートルール(エラー率・レイテンシなど)を運用チームと一緒につくり、「どの閾値で誰に通知するか」を決める。
- ログ・メトリクス・トレースを組み合わせた監視ダッシュボードを用意し、リリース時や障害時に必ず参照する「共通の画面」をつくる。
9. 参考リンク(さらに深く学びたい方向け)
※ここでは一般的な参考先を挙げています。実際に利用されるときは、最新版や公式ドキュメントをご確認ください。
- FastAPI 公式ドキュメント
- Logging に関する項目(Tips & Tricks / Advanced ページなど)
- Middleware, Events, Dependencies の章は監視にも応用できます。
- Prometheus 関連
prometheus_clientの公式ドキュメントprometheus-fastapi-instrumentatorの GitHub リポジトリ
- 構造化ログ
- Python logging による JSON ログ出力のサンプル記事
structlogやloguruなどのライブラリ(チームの好みに合わせて検討をどうぞ)
- 分散トレーシング
- OpenTelemetry Python SDK の公式ドキュメント
- Jaeger / Tempo / Zipkin などのトレーシングバックエンド
おわりに
FastAPI 自体はとても軽快で、ちょっとしたAPIならすぐに動かせてしまうのが魅力です。
だからこそ、「ひとまず動くもの」ができたあとに、ログや監視に手を回す余裕がなくなってしまうことも多いのかな、と思います。
ですが、ログ・メトリクス・ヘルスチェックを少しずつ整えていくことで、
- 障害が起きても冷静に原因を追える
- 「体感が遅い」と言われたときに数字で説明できる
- 新しいメンバーが入ってきても運用の勘所を共有しやすい
といった、ちょっと心強い状態に近づいていきます。
いきなり全部を完璧に整えなくても大丈夫です。
まずは logging の設定から、あるいは /health の1エンドポイントからでも、少しずつ「運用しやすい FastAPI」に育てていっていただけたら、とてもうれしいです。
わたしも、あなたのAPIが安定して長く愛されるサービスになっていくことを、そっと応援していますね。
