Каков механизм добавления сквозного trace ID в логи в распределенной системе

В распределенных системах (микросервисах) сквозной trace ID (идентификатор трассировки) используется для отслеживания полного пути запроса через несколько сервисов. Его добавление в логи происходит в три этапа: генерация, проброс и инъекция.

1. Генерация и получение

trace ID генерируется на входе в систему (например, в API Gateway или первом сервисе), если он отсутствует во входящем запросе. Если запрос уже содержит trace ID (например, в HTTP-заголовке X-Request-ID или traceparent), он извлекается оттуда.

2. Проброс через контекст

После получения trace ID его необходимо сделать доступным на протяжении всего жизненного цикла обработки запроса внутри сервиса. Для этого используются контекстно-локальные хранилища:

• contextvars в Python: Идеально подходит для асинхронных приложений (asyncio), так как корректно работает с конкурентными задачами.
• threading.local: Используется в многопоточных синхронных приложениях.

Пример с contextvars и Middleware в FastAPI:

import logging
from contextvars import ContextVar
from uuid import uuid4
from fastapi import Request

# Создаем контекстную переменную
trace_id_var: ContextVar[str] = ContextVar('trace_id', default=None)

async def add_trace_id_middleware(request: Request, call_next):
    # Получаем или генерируем trace_id
    trace_id = request.headers.get("X-Request-ID", str(uuid4()))

    # Устанавливаем значение в контекст
    token = trace_id_var.set(trace_id)

    response = await call_next(request)

    # Сбрасываем контекст после завершения запроса
    trace_id_var.reset(token)

    return response

3. Инъекция в логи

На этапе логирования trace ID извлекается из контекста и добавляется в каждую запись лога. Это делается с помощью кастомных фильтров или форматтеров.

Пример фильтра для logging:

class TraceIdFilter(logging.Filter):
    def filter(self, record):
        # Извлекаем trace_id из contextvars и добавляем в запись лога
        record.trace_id = trace_id_var.get()
        return True

# Настройка логгера
logger = logging.getLogger(__name__)
logger.addFilter(TraceIdFilter())

# Настройка форматтера для вывода trace_id
formatter = logging.Formatter("%(asctime)s - [%(levelname)s] - [trace_id=%(trace_id)s] - %(message)s")
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)

# При вызове logger.info() trace_id будет добавлен автоматически
logger.warning("User authentication failed")

Результат в логах:

2023-10-27 15:30:00,123 - [WARNING] - [trace_id=a1b2c3d4-e5f6-7890-1234-567890abcdef] - User authentication failed

Инструменты для observability, такие как OpenTelemetry, автоматизируют все эти шаги, включая передачу trace ID в дочерние HTTP-запросы.