Как организовать структурированное логирование (JSON-логи) с correlation id/request id, чтобы можно было сквозно проследить один запрос через несколько сервисов и слоёв приложения?

Короткий ответ

На входе генерируют или подхватывают request id, кладут его в контекст логирования (contextvars) и передают дальше в заголовках при межсервисных вызовах; каждая запись лога формируется в JSON с этим id, что позволяет в системе агрегации отфильтровать все записи одного запроса по всем сервисам.

Как это работает подробнее

middleware на входе проверяет заголовок или генерирует UUID, сохраняет в contextvars.ContextVar, доступный во всех корутинах обработки запроса; фильтр логирования добавляет значение из контекста в каждую запись.

Пример кода

example.pyКопировать

import contextvars, logging, uuid
request_id_var = contextvars.ContextVar("request_id", default="-")
class RequestIdFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id_var.get()
        return True
async def request_id_middleware(request, call_next):
    rid = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    request_id_var.set(rid)
    response = await call_next(request)
    response.headers["X-Request-ID"] = rid
    return response

Что использовать на практике

• Рекомендация. JSON-формат даёт возможность индексировать поля (request_id, user_id, duration_ms) и строить дашборды/алерты по ним; при вызове другого сервиса request id передают в заголовке, чтобы его подхватил downstream-сервис.

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

• Риск. генерировать request id на каждом сервисе независимо, не передавая его дальше; использовать threading.local в асинхронном коде вместо contextvars — один поток обслуживает множество конкурентных задач, и threading.local не изолирует их корректно.