Skip to content

python.missing_correlation_id

Observability Low

Detects HTTP handlers and service calls that don’t propagate correlation IDs, making distributed debugging difficult.

Why It Matters

Without correlation IDs:

Impossible to trace requests — Can’t follow a request across services
Debugging nightmares — Logs from different services can’t be correlated
Slow incident response — Finding related logs takes much longer
Missing context — Support tickets can’t be linked to specific requests

Example

# ❌ Before (no correlation ID)
@app.get("/users/{user_id}")
async def get_user(user_id: str):
    logger.info(f"Getting user {user_id}")
    user = await fetch_user(user_id)
    return user

# ✅ After (with correlation ID)
from fastapi import Header
import uuid

@app.get("/users/{user_id}")
async def get_user(
    user_id: str,
    x_correlation_id: str = Header(default_factory=lambda: str(uuid.uuid4()))
):
    logger.info(
        "Getting user",
        extra={"user_id": user_id, "correlation_id": x_correlation_id}
    )
    user = await fetch_user(user_id, correlation_id=x_correlation_id)
    return user

What Unfault Detects

HTTP handlers without correlation ID header handling
Outgoing HTTP calls without correlation ID forwarding
Log statements without correlation ID context
gRPC calls without metadata propagation

Auto-Fix

Unfault generates patches that add correlation ID middleware:

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
import contextvars
import uuid

correlation_id_var = contextvars.ContextVar('correlation_id', default=None)

class CorrelationIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        correlation_id = request.headers.get(
            'X-Correlation-ID',
            str(uuid.uuid4())
        )
        correlation_id_var.set(correlation_id)

        response = await call_next(request)
        response.headers['X-Correlation-ID'] = correlation_id
        return response

app.add_middleware(CorrelationIdMiddleware)

Structured Logging Integration

import structlog

def add_correlation_id(logger, method_name, event_dict):
    correlation_id = correlation_id_var.get()
    if correlation_id:
        event_dict['correlation_id'] = correlation_id
    return event_dict

structlog.configure(
    processors=[
        add_correlation_id,
        structlog.processors.JSONRenderer(),
    ]
)