Как правильно собирать логирование и метрики в FastAPI: единый формат событий для дебага и алертов

В продакшене FastAPI — это не столько про «как быстрее написать endpoint», сколько про управляемость системы. Когда что-то ломается, выигрывают команды, у которых есть:

1. предсказуемый формат событий в логах,
2. корреляция между логами, метриками и трассировками,
3. минимальный набор метрик, который отвечает на вопрос «что именно сломалось» и «где искать первопричину»,
4. алерты, которые не превращаются в шум.

Ниже — практическое руководство, как спроектировать единый формат событий для логирования и метрик в FastAPI, связать запросы через correlation-id и построить алерт-стратегию вокруг правильных сигналов.

Что именно логировать и мерить в проде: принцип «набора для расследования»

Логирование в веб-сервисах часто делают «как получится»: пишут всё подряд, потом находят сообщения в формате INFO something, и через неделю никто не может отличить важное от мусора. Метрики тоже превращаются в сотни графиков без ясной интерпретации.

1) Логи нужны для расследования конкретного инцидента

Хорошие логи в идеале отвечают на три вопроса:

• Какая операция? (endpoint/route, метод, источник)
• Что случилось? (статус, ошибка, причина, контекст)
• Где это видно в системе? (корреляция: correlation-id, сервис/версия, тайминг)

Для этого лог-события должны быть структурированными (JSON) и одинаковыми по ключам во всех обработчиках.

2) Метрики нужны для обнаружения проблемы и быстрой локализации

Метрики не должны пытаться заменить логи. Их задача — отследить тренды и быстро увидеть отклонения, которые указывают на первопричину.

В проде обычно достаточно метрик по нескольким слоям:

• HTTP/ASGI слой: latency, error rate, request rate
• Приложение: число ошибок по типам (валидация, бизнес-ошибки, интеграции)
• Интеграции: время/ошибки внешних запросов (БД, очередь, сторонние API)
• Ресурсы: CPU/Memory, если критично, и очередь/длина пула соединений

Важно: не пытайтесь сразу собрать метрики «про всё». Лучше выбрать маленький набор, который чаще всего помогает отвечать на вопросы продежды: «почему растёт время ответа», «где именно начали падать ответы», «это миграция или деградация БД».

Единый формат событий: как сделать, чтобы логи и метрики «склеивались»

Чтобы дебажить и строить алерты, нужен способ связывать события во времени и между слоями. В веб-сервисах таким связующим звеном обычно становится correlation-id (или trace-id, если вы используете OpenTelemetry).

correlation-id: что это и зачем оно в логах

correlation-id — идентификатор, который присваивается входящему запросу и проходит через:

• логирование на входе/выходе,
• логирование в обработчиках ошибок,
• метрики по конкретному запросу (на уровне агрегатов),
• трассировку (если вы её включите).

Он не заменяет trace-id, но прекрасно работает как минимальный слой корреляции.

Практическое требование: единый набор полей у событий должен содержать correlation_id, иначе вы не сможете быстро склеивать цепочку.

Рекомендуемая схема JSON-события для логов FastAPI

Ниже — базовый шаблон, который хорошо подходит и для отладки, и для алертов (при необходимости — алерты на базе логов):

{
  "ts": "2026-08-04T12:34:56.789Z",
  "level": "INFO",
  "service": "orders-api",
  "version": "2026.08.04",
  "env": "prod",
  "correlation_id": "f3b1c2d5-3d8a-4b2b-a4f0-12c8b7a6d1f9",
  "event": "request.completed",
  "http": {
    "method": "GET",
    "route": "/v1/orders/{order_id}",
    "status_code": 200,
    "client_ip": "203.0.113.10"
  },
  "timing_ms": {
    "duration": 42.7
  },
  "error": null,
  "context": {
    "user_id": null,
    "request_id": null
  }
}

Где это полезно:

• event показывает тип события: request.received, request.completed, request.failed
• http.route лучше, чем «сырая строка URL», чтобы видеть агрегаты по маршрутам
• error (null/объект) позволяет структурировать причину
• timing_ms даёт стабильный источник для анализа latency, даже если вы строите метрики отдельно

Как связать запросы через correlation-id в FastAPI

Стратегия: генерировать или принимать извне

Обычно клиенты (API gateway, frontend, другие сервисы) могут передавать correlation-id в заголовке, например X-Correlation-ID. Если заголовка нет — генерируем.

Плюс важная мелочь: если у вас есть возможность интегрировать с уже существующими trace-id от OpenTelemetry — лучше сделать корреляцию совместимой (иногда проще передавать trace-id как correlation-id).

Middleware для correlation-id и структурированного логирования

Ниже — пример middleware, который:

• назначает correlation_id,
• логирует события request.received и request.completed (или request.failed),
• сохраняет корреляцию в ответе (например, в X-Correlation-ID).

Пример использует стандартный logging и JSON-логгер в приложении. Формат логов показан на уровне структуры; конкретная реализация JSON зависит от выбранной библиотеки логирования.

import json
import logging
import time
import uuid
from contextvars import ContextVar
from typing import Callable, Awaitable

from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

correlation_id_var: ContextVar[str | None] = ContextVar("correlation_id", default=None)

logger = logging.getLogger("app")

def log_event(payload: dict) -> None:
    # Минимально: печатаем JSON в stdout.
    # В реальном проекте лучше использовать JSONFormatter.
    logger.info(json.dumps(payload, ensure_ascii=False))

class CorrelationIdMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, service: str, env: str, version: str, header_name: str = "X-Correlation-ID"):
        super().__init__(app)
        self.service = service
        self.env = env
        self.version = version
        self.header_name = header_name

    async def dispatch(self, request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
        incoming = request.headers.get(self.header_name)
        correlation_id = incoming or str(uuid.uuid4())
        correlation_id_var.set(correlation_id)

        start = time.perf_counter()

        # Логируем вход (по желанию можно дебажный уровень)
        log_event({
            "ts": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime()),
            "level": "INFO",
            "service": self.service,
            "version": self.version,
            "env": self.env,
            "correlation_id": correlation_id,
            "event": "request.received",
            "http": {
                "method": request.method,
                "route": request.url.path,  # на старте может быть сырым; дальше лучше использовать route_name
                "client_ip": request.client.host if request.client else None
            },
        })

        try:
            response = await call_next(request)
            duration_ms = (time.perf_counter() - start) * 1000.0

            response.headers[self.header_name] = correlation_id

            log_event({
                "ts": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime()),
                "level": "INFO",
                "service": self.service,
                "version": self.version,
                "env": self.env,
                "correlation_id": correlation_id,
                "event": "request.completed",
                "http": {
                    "method": request.method,
                    "route": getattr(request.scope.get("route"), "path", request.url.path),
                    "status_code": response.status_code,
                    "client_ip": request.client.host if request.client else None
                },
                "timing_ms": {"duration": duration_ms},
                "error": None
            })
            return response

        except Exception as exc:
            duration_ms = (time.perf_counter() - start) * 1000.0
            # В проде не логируйте чувствительные данные из исключений без фильтра.
            log_event({
                "ts": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime()),
                "level": "ERROR",
                "service": self.service,
                "version": self.version,
                "env": self.env,
                "correlation_id": correlation_id,
                "event": "request.failed",
                "http": {
                    "method": request.method,
                    "route": request.url.path,
                    "status_code": 500,
                    "client_ip": request.client.host if request.client else None
                },
                "timing_ms": {"duration": duration_ms},
                "error": {
                    "type": type(exc).__name__,
                    "message": str(exc)
                }
            })
            raise

Подводный камень: route-поля

На входе request.url.path — это конкретный URL. Для агрегатов лучше логировать route.path (шаблон), как в примере getattr(request.scope.get("route"), "path", ...). Однако route может быть не установлен на самых ранних этапах. Поэтому часто используют:

• request.scope["route"].path (если доступно),
• либо заранее определяют route_name через обработчики роутера.

Метрики для FastAPI: какие дают быстрый ответ на первопричину

Теперь про то, что реально помогает в расследовании.

Минимальный набор метрик, который почти всегда окупается

Рекомендованный стартовый набор:

1. Request count по статус-кодам
   • измерение: http_requests_total{method, route, status_code}
2. Latency в виде histogram (P50/P95/P99)
   • http_request_duration_seconds_bucket{method, route}
3. In-flight requests (сколько запросов одновременно обрабатывается)
   • http_in_flight_requests{route}
4. Ошибки по категориям (валидация, интеграции, бизнес, unknown)
   • app_errors_total{error_type, route}

И отдельно — метрики интеграций:

• db_request_duration_seconds_*
• external_http_duration_seconds_*
• количество ошибок на интеграциях

Почему это важно для первопричины:

• Если растёт latency при стабильном error rate — чаще всего это не “падает” код, а “тормозит”: пул соединений, внешняя система, блокировки.
• Если растёт error rate и только для конкретных маршрутов — локализуем на уровне endpoint/бизнес-операции.
• Если error rate растёт совместно с ростом ошибок интеграций — первопричина внизу стека, а не в обработчике в целом.

Инструментация метрик в FastAPI: пример с Prometheus

Ниже пример, который использует prometheus_client. Идея:

• middleware измеряет длительность,
• при завершении увеличивает счётчики,
• тип ошибки можно определить по исключениям или по статусу.

import time
from prometheus_client import Counter, Histogram, Gauge
from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

from typing import Callable, Awaitable

REQUESTS = Counter(
    "http_requests_total",
    "Total HTTP requests",
    ["method", "route", "status_code"],
)

DURATION = Histogram(
    "http_request_duration_seconds",
    "Request duration in seconds",
    ["method", "route"],
    buckets=(0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10)
)

IN_FLIGHT = Gauge(
    "http_in_flight_requests",
    "In-flight HTTP requests",
    ["route"]
)

class MetricsMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
        route = request.scope.get("route")
        # route.path обычно шаблон, например /v1/orders/{order_id}
        route_name = getattr(route, "path", request.url.path)

        IN_FLIGHT.labels(route=route_name).inc()
        start = time.perf_counter()
        try:
            response = await call_next(request)
            status = str(response.status_code)
            REQUESTS.labels(method=request.method, route=route_name, status_code=status).inc()
            return response
        except Exception:
            # Если исключение не было обработано и даст 500,
            # можно отнести к 500 или к отдельной категории.
            REQUESTS.labels(method=request.method, route=route_name, status_code="500").inc()
            raise
        finally:
            duration = time.perf_counter() - start
            DURATION.labels(method=request.method, route=route_name).observe(duration)
            IN_FLIGHT.labels(route=route_name).dec()

app = FastAPI()
app.add_middleware(MetricsMiddleware)

@app.get("/health")
def health():
    return {"ok": True}

Подводный камень: cardinality

Если вы добавите в label значение типа user_id, order_id, client_ip — получите взрыв количества временных рядов. Это убивает и Prometheus, и Grafana, и ваш бюджет.

Правило:

• labels должны быть низко-кардинальными: метод, route-шаблон, статус, тип ошибки, иногда уровень/тенант (если он небольшой).
• Всё остальное — в логах, а не в метриках.

Как связать логи и метрики единой “линией расследования”

Метрики и логи часто живут раздельно. Чтобы дебаг был быстрым, добавьте общий идентификатор и единый формат полей.

Практика: correlation-id в логах + метрики по маршрутам

Метрики обычно агрегируются, и включать correlation_id в labels не стоит (из-за cardinality). Вместо этого:

• корреляция через correlation_id работает в логах (и трассировках),
• метрики локализуют проблему по route/method/status_code,
• после обнаружения аномалии вы в логах фильтруете по route и корреляционным признакам (время интервала, пользователь/тенант если безопасно и низко-кардинально, но лучше по correlation_id если вы знаете конкретный запрос).

Если вы используете OpenTelemetry, то обычно корреляцию строят на trace context. Но в минимальном варианте correlation-id тоже решает половину задач.

Ошибки и алерты: как проектировать правила, чтобы они были “действующими”

Сделать алерт — это не “повесить красную линию”. Это подобрать условие так, чтобы:

• оно срабатывало редко на нормальных колебаниях,
• оно срабатывало быстро на реальные деградации,
• оно помогало ответить: “почему” и “где”.

Сигналы, которые лучше всего подходят для алертов

1) Ошибка по статус-коду

Классический алерт: доля 5xx растёт.

Логика:

• если 5xx растут — проблема на стороне сервиса/интеграций
• если 4xx растут — чаще проблема во входных данных, но иногда и безопасность/маршрутизация

Пример пром-подобного правила (псевдоформат, зависит от системы алертинга):

• Alert: http_5xx_rate_high
• Условие: rate(http_requests_total{status_code=~"5.."}[5m]) / rate(http_requests_total[5m]) > 0.02

Плюс: добавляйте фильтр по route, если сервис большой.

2) Latency (P95) выше порога

Если вы фиксируете histogram, проще алертить по P95/P99, а не по среднему.

Почему P95 лучше: среднее может оставаться нормальным, пока небольшая часть запросов “умирает” (например, медленные запросы к БД или timeouts на внешних API).

3) In-flight рост без соответствующего падения throughput

Если in-flight растёт, а request rate или completed не растут — есть риск зависаний, блокировок, исчерпания пула соединений.

4) Метрики интеграций

Если у вас есть отдельные метрики для БД/внешних вызовов, именно они часто являются “первопричиной”.

Например:

• растёт доля timeout в external_http
• параллельно растёт latency конкретного route
• значит алерт по route — вторичный, а первичную причину дают интеграции

Алгоритм: от алерта к причине за 3 шага

Чтобы расследование не растягивалось, заранее зафиксируйте “маршрут”:

1. Посмотреть route и метод по алерту (из алерта или дашборда)
2. Сверить корреляцию по типу: 5xx vs 4xx, какие ошибки в логах
3. Проверить интеграции: БД/очереди/внешние API на интервалы совпадения по времени

Если вы делаете лог-с