Как реализовать логирование запросов и трейсинг в FastAPI?

Структурированное логирование с structlog

Стандартный logging выводит текст — сложно парсить в ELK/Loki. structlog формирует JSON-строки с произвольными полями, включая request_id для корреляции.

# app/core/logging.py
import structlog
import logging

def setup_logging(debug: bool = False):
    logging.basicConfig(
        format="%(message)s",
        level=logging.DEBUG if debug else logging.INFO,
    )
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.JSONRenderer(),
        ],
        logger_factory=structlog.PrintLoggerFactory(),
    )

logger = structlog.get_logger()

Middleware для логирования запросов

import uuid
import time
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
import structlog

logger = structlog.get_logger()

class RequestLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = str(uuid.uuid4())
        structlog.contextvars.clear_contextvars()
        structlog.contextvars.bind_contextvars(
            request_id=request_id,
            method=request.method,
            path=request.url.path,
        )

        start = time.perf_counter()
        response = await call_next(request)
        duration_ms = (time.perf_counter() - start) * 1000

        logger.info(
            "request_handled",
            status_code=response.status_code,
            duration_ms=round(duration_ms, 2),
        )
        response.headers["X-Request-ID"] = request_id
        return response

Метрики с prometheus-fastapi-instrumentator

from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator

app = FastAPI()

Instrumentator().instrument(app).expose(app, endpoint="/metrics")
# теперь /metrics отдаёт http_requests_total, http_request_duration_seconds и др.

Distributed Tracing с OpenTelemetry

# pip install opentelemetry-sdk opentelemetry-instrumentation-fastapi
#             opentelemetry-exporter-otlp
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor

def setup_tracing(app):
    provider = TracerProvider()
    exporter = OTLPSpanExporter(endpoint="http://jaeger:4317", insecure=True)
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)

    FastAPIInstrumentor.instrument_app(app)
    SQLAlchemyInstrumentor().instrument(engine=engine)  # трейсит SQL-запросы

Добавление кастомных span

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

@app.get("/orders/{order_id}")
async def get_order(order_id: int):
    with tracer.start_as_current_span("fetch_order") as span:
        span.set_attribute("order.id", order_id)
        order = await order_service.get(order_id)
        span.set_attribute("order.status", order.status)
    return order

Подводные камни

• Логирование тела запроса — request.body() можно прочитать только один раз; после чтения в middleware тело будет пустым для endpoint. Используйте request.body() с последующим созданием нового Request-объекта через receive.
• structlog contextvars и asyncio — contextvars изолированы per-coroutine; при запуске фоновых задач (asyncio.create_task) контекст не наследуется автоматически — нужно явно копировать через copy_context().
• Sampling rate трейсинга — трейсить 100% запросов в high-traffic сервисе дорого. Настраивайте TraceIdRatioBased(0.1) для выборочного сбора.
• Sensitive data в span attributes — OpenTelemetry по умолчанию включает URL с query params; маскируйте пароли и токены через кастомный SpanProcessor.
• prometheus-fastapi-instrumentator и /metrics без auth — endpoint /metrics должен быть закрыт от публичного доступа; добавьте IP-whitelist в nginx или token auth.
• Дублирование request_id — если nginx или API gateway уже добавляет X-Request-ID, читайте его из заголовков вместо генерации нового: request.headers.get("X-Request-ID", str(uuid.uuid4())).
• JSON-логи с большими payload — логирование тела ответа с большими JSON-массивами может замедлить сервис и засорить хранилище логов; обрезайте до разумного размера.

• Описывать logging and tracing только как термин и не показывать механизм на минимальном примере.
• Игнорировать ошибки, пустые данные, конкурентный доступ или границы транзакции.
• Не связывать поведение с официальным контрактом FastAPI и реальной эксплуатацией.

• Объясняет logging and tracing через последовательность действий, а не через набор ключевых слов.
• Приводит короткий кодовый пример или production-сценарий с ожидаемым поведением.
• Называет хотя бы один риск: производительность, безопасность, транзакции, память или сопровождение.
• Умеет обсудить отказ, наблюдаемость и rollback без изменения публичного контракта.