Как trace ID попадает в лог?

Как trace ID попадает в логи

Что такое trace ID

Trace ID (трассировочный идентификатор) — это уникальный идентификатор запроса, который проходит через всю систему. Он позволяет отследить жизненный цикл единого запроса, особенно в микросервисной архитектуре.

Механизм попадания trace ID в логи

1. Генерация и инициализация

Trace ID создаётся в начале обработки запроса и сохраняется в контексте:

import logging
import uuid
from contextvars import ContextVar

# Контекстная переменная для хранения trace ID
trace_id_var: ContextVar[str] = ContextVar('trace_id', default='')

class TraceIDMiddleware:
    async def __call__(self, request, call_next):
        # Генерируем или берём из заголовка
        trace_id = request.headers.get('X-Trace-ID', str(uuid.uuid4()))
        token = trace_id_var.set(trace_id)
        
        try:
            response = await call_next(request)
            # Добавляем trace ID в ответ
            response.headers['X-Trace-ID'] = trace_id
            return response
        finally:
            trace_id_var.reset(token)

2. Интеграция с логгером

Trace ID попадает в логи через filter или formatter:

class TraceIDFilter(logging.Filter):
    def filter(self, record):
        # Добавляем trace ID к каждой записи лога
        record.trace_id = trace_id_var.get()
        return True

# Конфигурация логирования
logger = logging.getLogger(__name__)
logger.addFilter(TraceIDFilter())

# Формат логов
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - [%(trace_id)s] - %(levelname)s - %(message)s'
)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)

# Использование
logger.info('Обработка запроса')  # [uuid-xxxx] будет добавлен автоматически

3. Пример с FastAPI

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

app = FastAPI()

class TraceMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        trace_id = request.headers.get('X-Trace-ID', str(uuid.uuid4()))
        token = trace_id_var.set(trace_id)
        
        try:
            response = await call_next(request)
            response.headers['X-Trace-ID'] = trace_id
            return response
        finally:
            trace_id_var.reset(token)

app.add_middleware(TraceMiddleware)

@app.get('/items/{item_id}')
async def get_item(item_id: int):
    logger.info(f'Getting item {item_id}')  # trace ID добавится автоматически
    return {'item_id': item_id}

4. Распространение в микросервисах

Когда вы делаете запрос к другому сервису, передавайте trace ID:

import httpx

async def call_other_service():
    trace_id = trace_id_var.get()
    
    async with httpx.AsyncClient() as client:
        response = await client.get(
            'http://other-service/api/data',
            headers={'X-Trace-ID': trace_id}
        )
    return response.json()

5. Структурированные логи

Для лучшей интеграции используйте структурированные логи (JSON):

import json

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'logger': record.name,
            'trace_id': trace_id_var.get(),
            'message': record.getMessage(),
        }
        return json.dumps(log_data)

# Вывод: {"timestamp": "...", "level": "INFO", "trace_id": "uuid-xxxx", ...}

Лучшие практики

• Используйте contextvars для безопасного хранения в асинхронном коде
• Передавайте trace ID между сервисами через стандартные заголовки (X-Trace-ID, X-Request-ID)
• Логируйте trace ID в каждой значимой операции
• Используйте структурированные логи для легкого парсинга и аналитики
• Кэшируйте в ELK/Datadog для центрального анализа

Trace ID — незаменимый инструмент для отладки в продакшене и анализа потока данных через распределённую систему.