Как используется trace ID при отладке и мониторинге?

Использование Trace ID при отладке и мониторинге

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

1. Основная концепция Trace ID

Trace ID позволяет отследить полный путь одного запроса через все компоненты системы:

Клиент
  ↓ (Request с trace_id: abc123)
API Gateway
  ↓ (abc123 передаётся дальше)
User Service
  ↓ (abc123)
Database
  ↓
Cache Service
  ↓
Audit Service
  ↓ (все логируют abc123)
Клиент ← Response с trace_id: abc123

2. Реализация Trace ID в FastAPI/Django

FastAPI пример:

import uuid
from fastapi import FastAPI, Request
from fastapi.middleware.base import BaseHTTPMiddleware
import logging

app = FastAPI()

logger = logging.getLogger(__name__)

class TraceIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Получаем trace_id из заголовка или генерируем новый
        trace_id = request.headers.get('X-Trace-ID', str(uuid.uuid4()))
        
        # Сохраняем в контекст запроса
        request.state.trace_id = trace_id
        
        # Добавляем в логи
        response = await call_next(request)
        response.headers['X-Trace-ID'] = trace_id
        
        return response

app.add_middleware(TraceIDMiddleware)

@app.get('/users/{user_id}')
async def get_user(user_id: int, request: Request):
    trace_id = request.state.trace_id
    logger.info(f'[{trace_id}] Получение пользователя {user_id}')
    return {'user_id': user_id, 'trace_id': trace_id}

3. Использование Trace ID в сервисах

Передаём trace_id между вызовами микросервисов:

import httpx
from contextvars import ContextVar

trace_id_var: ContextVar[str] = ContextVar('trace_id', default=None)

class ServiceClient:
    def __init__(self):
        self.client = httpx.AsyncClient()
    
    async def call_service(self, service_url: str, endpoint: str):
        """Вызываем другой сервис с передачей trace_id."""
        trace_id = trace_id_var.get()
        
        headers = {}
        if trace_id:
            headers['X-Trace-ID'] = trace_id
        
        response = await self.client.get(
            f'{service_url}{endpoint}',
            headers=headers
        )
        return response.json()

# Использование
service_client = ServiceClient()

@app.get('/orders/{order_id}')
async def get_order(order_id: int, request: Request):
    trace_id = request.state.trace_id
    trace_id_var.set(trace_id)
    
    logger.info(f'[{trace_id}] Получение заказа {order_id}')
    
    # Вызываем другой сервис
    user_data = await service_client.call_service(
        'http://user-service:8000',
        f'/users/{order_id}'
    )
    
    return {'order_id': order_id, 'user_data': user_data}

4. Логирование с Trace ID

Добавляем trace_id во все логи автоматически:

import logging
from contextvars import ContextVar

trace_id_var: ContextVar[str] = ContextVar('trace_id', default='unknown')

class TraceIDFilter(logging.Filter):
    """Filter для добавления trace_id в логи."""
    def filter(self, record):
        record.trace_id = trace_id_var.get()
        return True

# Настройка логирования
logging.basicConfig(
    format='%(asctime)s - [%(trace_id)s] - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger(__name__)
logger.addFilter(TraceIDFilter())

# Использование
trace_id_var.set('req-12345')
logger.info('Обрабатываем запрос')  # [req-12345] Обрабатываем запрос
logger.error('Ошибка обработки')     # [req-12345] Ошибка обработки

5. Trace ID в асинхронном коде

Используем contextvars для асинхронных операций:

import asyncio
from contextvars import ContextVar

trace_id_var: ContextVar[str] = ContextVar('trace_id', default=None)

async def database_query(query: str):
    """Асинхронный запрос с trace_id."""
    trace_id = trace_id_var.get()
    logger.info(f'[{trace_id}] Выполнение запроса: {query}')
    await asyncio.sleep(0.1)  # Имитация запроса
    return {'result': 'success'}

async def process_request(trace_id: str):
    token = trace_id_var.set(trace_id)
    try:
        result1 = await database_query('SELECT * FROM users')
        result2 = await database_query('SELECT * FROM orders')
        return {'data': [result1, result2]}
    finally:
        trace_id_var.reset(token)

# Использование
await process_request('trace-abc123')

6. Интеграция с системой мониторинга

Передаём trace_id в OpenTelemetry или другую систему:

from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# Инициализация Jaeger
jaeger_exporter = JaegerExporter(
    agent_host_name='localhost',
    agent_port=6831,
)

trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)

tracer = trace.get_tracer(__name__)

@app.get('/users/{user_id}')
async def get_user(user_id: int, request: Request):
    trace_id = request.state.trace_id
    
    with tracer.start_as_current_span('get_user') as span:
        span.set_attribute('trace_id', trace_id)
        span.set_attribute('user_id', user_id)
        
        logger.info(f'[{trace_id}] Получение пользователя {user_id}')
        user = await fetch_user(user_id)
        
        return {'user': user, 'trace_id': trace_id}

7. Структурное логирование с Trace ID

Используем структурированные логи с bibliotekами типа structlog:

import structlog
from contextvars import ContextVar

trace_id_var: ContextVar[str] = ContextVar('trace_id', default=None)

structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt='iso'),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        structlog.processors.JSONRenderer()
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

logger = structlog.get_logger()

@app.get('/orders/{order_id}')
async def get_order(order_id: int, request: Request):
    trace_id = request.state.trace_id
    
    # Логируем с контекстом
    logger.info(
        'order_retrieved',
        trace_id=trace_id,
        order_id=order_id,
        timestamp=datetime.now().isoformat()
    )
    
    return {'order_id': order_id, 'trace_id': trace_id}

8. Анализ логов по Trace ID

Потом в системе мониторинга (ELK, CloudWatch, Datadog) ищем:

# Elasticsearch
GET /logs/_search
{
  "query": {
    "match": {
      "trace_id": "req-12345"
    }
  }
}

# CloudWatch Insights
fields @timestamp, @message, trace_id
| filter trace_id = "req-12345"
| stats count() as event_count

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

• Всегда генерируйте trace_id для входящих запросов
• Передавайте trace_id между всеми сервисами
• Включайте trace_id в ВСЕ логи автоматически
• Возвращайте trace_id в ответе клиенту
• Используйте contextvars для отслеживания в асинхронном коде
• Интегрируйте с системой мониторинга (Jaeger, DataDog, CloudWatch)
• Сохраняйте логи достаточно долго для анализа

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