3. Request Tracing con Correlation IDs

Descripción

Un correlation ID (también llamado request_id o trace_id) es el hilo conductor de todos los logs de un request. Sin él, tienes logs dispersos sin forma de correlacionar. Con él, puedes reconstruir la historia completa de cualquier request en segundos: qué llegó, qué procesó el guardrail, cuánto tardó el LLM, qué se devolvió. Esta cápsula implementa correlation IDs end-to-end con contextvars, middleware de FastAPI, y structlog.

El problema sin correlation IDs

Logs en producción sin correlation IDs:

10:30:01 INFO  {"event": "request_started", "endpoint": "/analyze"}
10:30:01 INFO  {"event": "request_started", "endpoint": "/analyze"}  ← Request 2, mezclado
10:30:02 WARNING {"event": "guardrail_activated", "type": "injection"}  ← ¿De cuál request?
10:30:02 INFO  {"event": "llm_request_completed", "tokens": 450}      ← ¿De cuál request?
10:30:03 ERROR {"event": "llm_request_failed", "error": "timeout"}    ← ¿De cuál request?
10:30:03 INFO  {"event": "request_completed", "status": 200}

Preguntas imposibles de responder:
→ ¿El guardrail de injection fue del request que falló con timeout o del que completó?
→ ¿Cuál de los dos requests del inicio fue el que completó?
→ ¿Cuál es el costo total del request que completó?

Los mismos logs CON correlation IDs:

10:30:01 INFO  {"event": "request_started", "request_id": "a1b2c3d4"}
10:30:01 INFO  {"event": "request_started", "request_id": "e5f6g7h8"}
10:30:02 WARNING {"event": "guardrail_activated", "request_id": "a1b2c3d4", "type": "injection"}
10:30:02 INFO  {"event": "llm_request_completed", "request_id": "e5f6g7h8", "tokens": 450}
10:30:03 ERROR {"event": "llm_request_failed", "request_id": "a1b2c3d4", "error": "timeout"}
10:30:03 INFO  {"event": "request_completed", "request_id": "e5f6g7h8", "status": 200}

Ahora sí puedo reconstruir:
→ request a1b2c3d4: detectó injection → falló con timeout (¿el timeout vino del intento de LLM para la detección?)
→ request e5f6g7h8: completó normalmente, 450 tokens

Generar correlation IDs

# src/tracing.py
import uuid
import contextvars
from typing import Optional

# ─── Almacenamiento del request_id por request (thread-safe, async-safe) ───

# contextvars es la forma correcta de almacenar estado por-request en async Python
# Es thread-safe Y async-safe: cada request tiene su propio "slot"
_request_id_var: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar(
    "request_id",
    default=None
)

def generate_request_id() -> str:
    """
    Genera un ID único de 8 caracteres hexadecimales.
    
    Por qué 8 chars en vez del UUID completo (32):
    - 8 chars hex = 4 billion combinaciones
    - Colisión esperada después de ~65,000 requests simultáneos
    - Para 1,000 req/día, probabilidad de colisión en 1 día es negligible
    - Mucho más legible en logs y en respuestas a usuarios
    
    Usar UUID completo si tienes millones de requests/segundo.
    """
    return uuid.uuid4().hex[:8]

def set_request_id(request_id: str) -> None:
    """Establecer el request_id para el contexto actual."""
    _request_id_var.set(request_id)

def get_request_id() -> Optional[str]:
    """Obtener el request_id del contexto actual."""
    return _request_id_var.get()

def get_or_create_request_id() -> str:
    """Obtener el request_id existente o crear uno nuevo."""
    current = _request_id_var.get()
    if current is None:
        current = generate_request_id()
        _request_id_var.set(current)
    return current

Middleware de FastAPI: inyectar request_id en cada request

# src/middleware.py
import time
import structlog
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.types import ASGIApp

from src.tracing import generate_request_id, set_request_id, get_request_id

log = structlog.get_logger()

class RequestTracingMiddleware(BaseHTTPMiddleware):
    """
    Middleware que:
    1. Extrae o genera un request_id
    2. Lo almacena en contextvars (accesible desde todo el código)
    3. Añade el request_id a la respuesta HTTP como header
    4. Logguea inicio y fin del request con métricas básicas
    """
    
    async def dispatch(self, request: Request, call_next) -> Response:
        # Extraer o generar request_id
        # El cliente puede enviar su propio X-Request-ID (para distributed tracing)
        request_id = (
            request.headers.get("X-Request-ID") or
            generate_request_id()
        )
        
        # Almacenar en contextvars para acceso global
        set_request_id(request_id)
        
        # También almacenar en structlog context para que aparezca en TODOS los logs
        structlog.contextvars.bind_contextvars(request_id=request_id)
        
        start_time = time.time()
        
        # Log de inicio del request
        log.info(
            "request_started",
            method=request.method,
            path=request.url.path,
            client_ip=request.client.host if request.client else "unknown"
        )
        
        try:
            # Procesar el request
            response = await call_next(request)
            
            duration_ms = (time.time() - start_time) * 1000
            
            # Log de fin del request
            log.info(
                "request_completed",
                status_code=response.status_code,
                duration_ms=round(duration_ms, 1),
                path=request.url.path
            )
            
            # Añadir request_id en la respuesta para que el cliente lo pueda reportar
            response.headers["X-Request-ID"] = request_id
            
            return response
        
        except Exception as e:
            duration_ms = (time.time() - start_time) * 1000
            
            log.error(
                "request_exception",
                error_type=type(e).__name__,
                error_message=str(e)[:200],
                duration_ms=round(duration_ms, 1)
            )
            raise
        
        finally:
            # Limpiar contexto de structlog al terminar el request
            structlog.contextvars.clear_contextvars()

# Registrar en la app FastAPI:
# app.add_middleware(RequestTracingMiddleware)

structlog context: request_id en todos los logs sin pasarlo explícitamente

# Cómo funciona structlog.contextvars:

# En el middleware (inicio del request):
structlog.contextvars.bind_contextvars(request_id="a1b2c3d4")

# En cualquier lugar del código durante ese request:
log = structlog.get_logger()
log.info("llm_called", model="gpt-4o-mini")
# → {"event": "llm_called", "model": "gpt-4o-mini", "request_id": "a1b2c3d4"}
# El request_id se añade automáticamente porque está en el contexto

# En otro lugar del código, sin pasarlo:
log.warning("guardrail_activated", type="injection")
# → {"event": "guardrail_activated", "type": "injection", "request_id": "a1b2c3d4"}

# Esto funciona porque structlog.contextvars usa contextvars internamente,
# que es thread-safe y async-safe

# Configuración necesaria en structlog:
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,  # ← Este processor lee el contexto
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ]
)

Propagación a servicios externos: distributed tracing

# Si tu app llama a otros servicios internos:

import httpx
import structlog
from src.tracing import get_request_id

log = structlog.get_logger()

async def call_external_service(endpoint: str, data: dict) -> dict:
    """
    Llamar a un servicio externo pasando el request_id como header.
    Esto permite correlacionar logs entre múltiples servicios.
    """
    request_id = get_request_id()
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            endpoint,
            json=data,
            headers={
                "X-Request-ID": request_id,       # Propagar el ID
                "X-Correlation-ID": request_id,   # Alias común
            }
        )
    
    log.info(
        "external_service_called",
        service_endpoint=endpoint,
        response_status=response.status_code
        # request_id se añade automáticamente desde el contexto
    )
    
    return response.json()

# El servicio externo recibe el X-Request-ID y lo usa en SUS logs.
# Ahora puedes correlacionar logs entre servicio A y servicio B.

Usar request_id para soporte al usuario

# src/app/main.py

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from src.tracing import get_request_id

app = FastAPI()

class ErrorResponse(BaseModel):
    error: str
    request_id: str  # El usuario puede reportar esto para soporte

@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
    request_id = get_request_id() or "unknown"
    
    return JSONResponse(
        status_code=500,
        content={
            "error": "Internal server error",
            "request_id": request_id,
            "message": "Si el error persiste, contáctanos con el request_id"
        }
    )

# El flujo de soporte con correlation IDs:

Usuario: "Tuve un error hace 10 minutos"
Support: "¿Tienes el request_id del error?"
Usuario: "Sí, es a1b2c3d4"

Support engineer busca en logs:
$ jq 'select(.request_id == "a1b2c3d4")' logs.json | jq -s 'sort_by(.timestamp)'

Resultado:
{"event": "request_started", "request_id": "a1b2c3d4", "timestamp": "10:30:01", "path": "/analyze"}
{"event": "llm_request_started", "request_id": "a1b2c3d4", "timestamp": "10:30:01"}
{"event": "llm_request_failed", "request_id": "a1b2c3d4", "timestamp": "10:30:06", "error": "RateLimitError"}
{"event": "request_exception", "request_id": "a1b2c3d4", "timestamp": "10:30:06", "duration_ms": 5234}

Diagnóstico en 30 segundos: el usuario fue afectado por un rate limit del API.

Configuración completa de structlog con tracing

# src/logging_config.py
import logging
import os
import structlog

def configure_logging():
    """
    Configuración completa de structlog para producción y desarrollo.
    
    Producción: JSON logs (parseable por máquinas)
    Desarrollo: Console output legible por humanos
    """
    
    # Configurar el logger estándar de Python para interceptar logs de librerías
    logging.basicConfig(
        format="%(message)s",
        stream=None,
        level=logging.INFO
    )
    
    # Procesadores compartidos entre dev y prod
    shared_processors = [
        structlog.contextvars.merge_contextvars,  # Leer request_id del contexto
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.stdlib.add_logger_name,
        structlog.processors.StackInfoRenderer(),
    ]
    
    is_production = os.getenv("ENVIRONMENT", "development") == "production"
    
    if is_production:
        processors = shared_processors + [
            structlog.processors.format_exc_info,  # Excepciones en JSON
            structlog.processors.JSONRenderer()
        ]
    else:
        # Desarrollo: colores y formato legible
        processors = shared_processors + [
            structlog.dev.ConsoleRenderer(colors=True)
        ]
    
    structlog.configure(
        processors=processors,
        wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(),
        cache_logger_on_first_use=True,
    )

Tests de request tracing

# tests/unit/test_tracing.py
import pytest
import json
import io
import structlog
from src.tracing import set_request_id, get_request_id, generate_request_id

def test_request_id_appears_in_all_logs():
    """El request_id del contexto debe aparecer en todos los logs."""
    output = io.StringIO()
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.JSONRenderer()
        ],
        logger_factory=structlog.PrintLoggerFactory(file=output)
    )
    
    structlog.contextvars.bind_contextvars(request_id="test-req-01")
    log = structlog.get_logger()
    
    log.info("event_1", field="a")
    log.info("event_2", field="b")
    log.warning("event_3", field="c")
    
    structlog.contextvars.clear_contextvars()
    
    lines = [json.loads(line) for line in output.getvalue().strip().split("\n") if line]
    
    assert len(lines) == 3
    for line in lines:
        assert line["request_id"] == "test-req-01", \
            f"request_id missing from log: {line}"

def test_different_requests_have_different_ids():
    """Dos requests distintos no deben tener el mismo ID."""
    ids = {generate_request_id() for _ in range(1000)}
    # 1000 IDs, todos distintos
    assert len(ids) == 1000

def test_contextvars_isolation():
    """
    Los contextvars de un 'request' no deben contaminar otro.
    Simula dos requests procesados secuencialmente.
    """
    # Request 1
    structlog.contextvars.bind_contextvars(request_id="req-001")
    assert structlog.contextvars.get_contextvars()["request_id"] == "req-001"
    
    # Fin del request 1, limpiar
    structlog.contextvars.clear_contextvars()
    
    # Request 2
    structlog.contextvars.bind_contextvars(request_id="req-002")
    assert structlog.contextvars.get_contextvars()["request_id"] == "req-002"
    
    structlog.contextvars.clear_contextvars()

Ejercicios

Ejercicio 1: Implementar el middleware desde cero

Sin mirar el ejemplo, escribe un middleware FastAPI que:

1. Extrae el X-Request-ID del header si existe, o genera uno nuevo
2. Lo añade al contexto de structlog
3. Lo añade al response header X-Request-ID
4. Logguea inicio y fin del request con duración

from starlette.middleware.base import BaseHTTPMiddleware
import structlog, time, uuid

log = structlog.get_logger()

class TracingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        request_id = request.headers.get("X-Request-ID") or uuid.uuid4().hex[:8]
        structlog.contextvars.bind_contextvars(request_id=request_id)
        
        start = time.time()
        log.info("request_started", path=request.url.path)
        
        try:
            response = await call_next(request)
            log.info("request_done", status=response.status_code,
                     duration_ms=round((time.time()-start)*1000, 1))
            response.headers["X-Request-ID"] = request_id
            return response
        finally:
            structlog.contextvars.clear_contextvars()

Ejercicio 2: Debugging con request_id

Dado este log output, reconstruye la historia del request b9c3e7a1:

{"event": "request_started", "request_id": "b9c3e7a1", "path": "/analyze", "timestamp": "T10:00:00"}
{"event": "guardrail_activated", "request_id": "b9c3e7a1", "type": "pii_redaction", "timestamp": "T10:00:00.1"}
{"event": "llm_request_completed", "request_id": "b9c3e7a1", "tokens": 180, "duration_ms": 1200, "timestamp": "T10:00:01.3"}
{"event": "validation_failed", "request_id": "b9c3e7a1", "error": "invalid JSON", "timestamp": "T10:00:01.3"}
{"event": "fallback_used", "request_id": "b9c3e7a1", "timestamp": "T10:00:01.3"}
{"event": "request_completed", "request_id": "b9c3e7a1", "status": 200, "timestamp": "T10:00:01.4"}

Ejercicio 3: Isolated context para tests async

¿Por qué en tests async es importante limpiar el contexto de structlog entre tests?

@pytest.fixture(autouse=True)
def clear_log_context():
    structlog.contextvars.clear_contextvars()
    yield
    structlog.contextvars.clear_contextvars()

Resumen

• Correlation ID (request_id) es el hilo conductor de todos los logs de un request
• contextvars almacena el request_id de forma thread-safe y async-safe
• structlog.contextvars.bind_contextvars hace que el ID aparezca en todos los logs del request sin pasarlo explícitamente
• El middleware de FastAPI es el lugar correcto para generar/extraer y almacenar el request_id
• El cliente recibe el request_id como header X-Request-ID para poder reportarlo en soporte
• Distributed tracing: propagar el request_id a servicios externos con headers

Recursos adicionales

1. structlog contextvars — Documentación oficial
2. Python contextvars — Módulo estándar
3. FastAPI Middleware — Middleware en FastAPI
4. W3C Trace Context — Estándar para propagación de trace IDs
5. OpenTelemetry Python — Para distributed tracing completo