A customer says their payment failed on Tuesday afternoon. Your API, a background worker and two third-party services were involved. Without a shared identifier, you’re matching timestamps across five sets of logs. With a correlation ID, you search for one string.
What a correlation ID is
A correlation ID is a unique value attached to a request when it enters your system. It is written into every log line produced while handling that request and passed on to every service the request calls. Many teams carry it in an X-Request-ID header. If your load balancer or API gateway already sets one, reuse it rather than creating a second ID.
The example
The FastAPI app below accepts or creates an ID for each request, adds it to every log line and forwards it to a partner API. It uses only the standard library, FastAPI and httpx.
import logging
import re
import uuid
from contextvars import ContextVar
import httpx
from fastapi import FastAPI, Request
request_id: ContextVar[str] = ContextVar("request_id", default="-")
VALID_ID = re.compile(r"^[A-Za-z0-9-]{1,64}$")
class RequestIdFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
record.request_id = request_id.get()
return True
handler = logging.StreamHandler()
handler.addFilter(RequestIdFilter())
handler.setFormatter(
logging.Formatter("%(asctime)s %(levelname)s [%(request_id)s] %(name)s: %(message)s")
)
logging.basicConfig(level=logging.INFO, handlers=[handler])
logger = logging.getLogger("orders")
app = FastAPI()
@app.middleware("http")
async def add_request_id(request: Request, call_next):
incoming = request.headers.get("X-Request-ID", "")
rid = incoming if VALID_ID.match(incoming) else str(uuid.uuid4())
token = request_id.set(rid)
try:
response = await call_next(request)
finally:
request_id.reset(token)
response.headers["X-Request-ID"] = rid
return response
async def forward_request_id(request: httpx.Request) -> None:
request.headers["X-Request-ID"] = request_id.get()
partner = httpx.AsyncClient(
base_url="https://partner.example.com",
timeout=10,
event_hooks={"request": [forward_request_id]},
)
@app.post("/orders")
async def create_order(order: dict) -> dict:
logger.info("Creating order")
response = await partner.post("/orders", json=order)
response.raise_for_status()
logger.info("Partner accepted order")
return {"status": "created"}Keep the ID for the life of the request
Python’s contextvars hold a separate value for each request, even in async code where many requests share one thread. The middleware sets the ID on the way in, resets it on the way out and adds it to the response, so clients and support staff can quote it back to you.
Put it in every log line
The logging filter copies the ID onto each log record so the formatter can print it. Existing logger.info calls don’t need to change, which makes this easy to add to a codebase that already logs well.
Pass it on
The httpx event hook adds the header to every request made with that client. If the service you call logs the same ID, you can follow a request across the boundary and give a partner something precise when you raise an issue with them.
Do the same for background work. Put the ID in the message you send to your queue, and set it in the worker before processing, so the job’s logs line up with the request that started it.
Don’t trust the header blindly
An incoming ID comes from outside your system. The example accepts only short values made of letters, digits and hyphens, and generates a fresh one otherwise. That stops anyone writing line breaks or very long strings into your logs.
What it gives you
- Every log line for one customer’s request, in a single search.
- A way to follow a failure across your API, workers and partners.
- A precise reference to share when a third party needs to help.
We tested the example with a mocked partner API, including a request with an invalid ID. It is a small change that pays for itself the first time something goes wrong in production, which is where our production support work usually starts.