3. Request Tracing with Correlation IDs

Description

A correlation ID (also called request_id or trace_id) is the common thread of all the logs of a request. Without it, you have scattered logs with no way to correlate them. With it, you can reconstruct the complete story of any request in seconds: what arrived, what the guardrail processed, how long the LLM took, what was returned. This capsule implements correlation IDs end-to-end with contextvars, FastAPI middleware, and structlog.

The problem without correlation IDs

Logs in production without correlation IDs:

10:30:01 INFO  {"event": "request_started", "endpoint": "/analyze"}
10:30:01 INFO  {"event": "request_started", "endpoint": "/analyze"}  ← Request 2, mixed in
10:30:02 WARNING {"event": "guardrail_activated", "type": "injection"}  ← From which request?
10:30:02 INFO  {"event": "llm_request_completed", "tokens": 450}      ← From which request?
10:30:03 ERROR {"event": "llm_request_failed", "error": "timeout"}    ← From which request?
10:30:03 INFO  {"event": "request_completed", "status": 200}

Questions impossible to answer:
→ Was the injection guardrail from the request that failed with a timeout or the one that completed?
→ Which of the two requests at the start was the one that completed?
→ What is the total cost of the request that completed?

The same logs WITH 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}

Now I can reconstruct:
→ request a1b2c3d4: detected injection → failed with a timeout (did the timeout come from the LLM attempt for the detection?)
→ request e5f6g7h8: completed normally, 450 tokens

Generating correlation IDs

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

# ─── Per-request request_id storage (thread-safe, async-safe) ───

# contextvars is the correct way to store per-request state in async Python
# It's thread-safe AND async-safe: each request has its own "slot"
_request_id_var: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar(
    "request_id",
    default=None
)

def generate_request_id() -> str:
    """
    Generates a unique 8-character hexadecimal ID.
    
    Why 8 chars instead of the full UUID (32):
    - 8 hex chars = 4 billion combinations
    - Expected collision after ~65,000 simultaneous requests
    - For 1,000 req/day, the collision probability in 1 day is negligible
    - Much more readable in logs and in responses to users
    
    Use the full UUID if you have millions of requests/second.
    """
    return uuid.uuid4().hex[:8]

def set_request_id(request_id: str) -> None:
    """Set the request_id for the current context."""
    _request_id_var.set(request_id)

def get_request_id() -> Optional[str]:
    """Get the request_id from the current context."""
    return _request_id_var.get()

def get_or_create_request_id() -> str:
    """Get the existing request_id or create a new one."""
    current = _request_id_var.get()
    if current is None:
        current = generate_request_id()
        _request_id_var.set(current)
    return current

FastAPI middleware: inject request_id into every 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 that:
    1. Extracts or generates a request_id
    2. Stores it in contextvars (accessible from all the code)
    3. Adds the request_id to the HTTP response as a header
    4. Logs the start and end of the request with basic metrics
    """
    
    async def dispatch(self, request: Request, call_next) -> Response:
        # Extract or generate the request_id
        # The client can send its own X-Request-ID (for distributed tracing)
        request_id = (
            request.headers.get("X-Request-ID") or
            generate_request_id()
        )
        
        # Store in contextvars for global access
        set_request_id(request_id)
        
        # Also store in the structlog context so it appears in ALL logs
        structlog.contextvars.bind_contextvars(request_id=request_id)
        
        start_time = time.time()
        
        # Request start log
        log.info(
            "request_started",
            method=request.method,
            path=request.url.path,
            client_ip=request.client.host if request.client else "unknown"
        )
        
        try:
            # Process the request
            response = await call_next(request)
            
            duration_ms = (time.time() - start_time) * 1000
            
            # Request end log
            log.info(
                "request_completed",
                status_code=response.status_code,
                duration_ms=round(duration_ms, 1),
                path=request.url.path
            )
            
            # Add the request_id to the response so the client can report it
            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:
            # Clean up the structlog context when the request ends
            structlog.contextvars.clear_contextvars()

# Register in the FastAPI app:
# app.add_middleware(RequestTracingMiddleware)

structlog context: request_id in all logs without passing it explicitly

# How structlog.contextvars works:

# In the middleware (start of the request):
structlog.contextvars.bind_contextvars(request_id="a1b2c3d4")

# Anywhere in the code during that request:
log = structlog.get_logger()
log.info("llm_called", model="gpt-4o-mini")
# → {"event": "llm_called", "model": "gpt-4o-mini", "request_id": "a1b2c3d4"}
# The request_id is added automatically because it's in the context

# Elsewhere in the code, without passing it:
log.warning("guardrail_activated", type="injection")
# → {"event": "guardrail_activated", "type": "injection", "request_id": "a1b2c3d4"}

# This works because structlog.contextvars uses contextvars internally,
# which is thread-safe and async-safe

# Required configuration in structlog:
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,  # ← This processor reads the context
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ]
)

Propagation to external services: distributed tracing

# If your app calls other internal services:

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:
    """
    Call an external service passing the request_id as a header.
    This lets you correlate logs across multiple services.
    """
    request_id = get_request_id()
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            endpoint,
            json=data,
            headers={
                "X-Request-ID": request_id,       # Propagate the ID
                "X-Correlation-ID": request_id,   # Common alias
            }
        )
    
    log.info(
        "external_service_called",
        service_endpoint=endpoint,
        response_status=response.status_code
        # request_id is added automatically from the context
    )
    
    return response.json()

# The external service receives the X-Request-ID and uses it in ITS logs.
# Now you can correlate logs between service A and service B.

Using request_id for user support

# 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  # The user can report this for support

@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": "If the error persists, contact us with the request_id"
        }
    )

# The support flow with correlation IDs:

User: "I had an error 10 minutes ago"
Support: "Do you have the error's request_id?"
User: "Yes, it's a1b2c3d4"

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

Result:
{"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}

Diagnosis in 30 seconds: the user was affected by an API rate limit.

Complete structlog configuration with tracing

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

def configure_logging():
    """
    Complete structlog configuration for production and development.
    
    Production: JSON logs (machine-parseable)
    Development: Human-readable console output
    """
    
    # Configure Python's standard logger to intercept library logs
    logging.basicConfig(
        format="%(message)s",
        stream=None,
        level=logging.INFO
    )
    
    # Processors shared between dev and prod
    shared_processors = [
        structlog.contextvars.merge_contextvars,  # Read request_id from the context
        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,  # Exceptions in JSON
            structlog.processors.JSONRenderer()
        ]
    else:
        # Development: colors and readable format
        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,
    )

Request tracing tests

# 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():
    """The context's request_id must appear in all 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():
    """Two different requests must not have the same ID."""
    ids = {generate_request_id() for _ in range(1000)}
    # 1000 IDs, all different
    assert len(ids) == 1000

def test_contextvars_isolation():
    """
    One 'request's' contextvars must not contaminate another.
    Simulates two requests processed sequentially.
    """
    # Request 1
    structlog.contextvars.bind_contextvars(request_id="req-001")
    assert structlog.contextvars.get_contextvars()["request_id"] == "req-001"
    
    # End of request 1, clean up
    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()

Exercises

Exercise 1: Implement the middleware from scratch

Without looking at the example, write a FastAPI middleware that:

1. Extracts the X-Request-ID from the header if it exists, or generates a new one
2. Adds it to the structlog context
3. Adds it to the response header X-Request-ID
4. Logs the start and end of the request with duration

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()

Exercise 2: Debugging with request_id

Given this log output, reconstruct the story of 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"}

Exercise 3: Isolated context for async tests

Why is it important in async tests to clear the structlog context between tests?

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

Summary

• Correlation ID (request_id) is the common thread of all the logs of a request
• contextvars stores the request_id in a thread-safe and async-safe way
• structlog.contextvars.bind_contextvars makes the ID appear in all the request's logs without passing it explicitly
• The FastAPI middleware is the right place to generate/extract and store the request_id
• The client receives the request_id as the X-Request-ID header so it can report it in support
• Distributed tracing: propagate the request_id to external services with headers

Additional resources

1. structlog contextvars — Official documentation
2. Python contextvars — Standard module
3. FastAPI Middleware — Middleware in FastAPI
4. W3C Trace Context — Standard for propagating trace IDs
5. OpenTelemetry Python — For complete distributed tracing