FastAPI: Logging, metrics & tracing

At 2 a.m. the alert says “orders API error rate 8%”. Can you tell which endpoint, since when, for which customers, and why? That depends entirely on what you built before the incident. The three signals are logs (what happened, in detail), metrics (how much and how fast, cheaply, over time) and traces (where the time went across services). One request id ties them together. This part builds each one from the parts, runs it, and shows what production libraries do the same way.

1. Structured logs with a request id

Logs are for machines first: one JSON object per line, with the same field names everywhere, so that Loki, Elasticsearch or CloudWatch can filter request_id="…" or route="/orders/{id}" AND status>=500. A ContextVar carries the request id to every log call without passing it around. It even crosses into the thread pool, because Starlette copies the context when it runs a sync endpoint.

JSON logs, request ids and an access log

import contextvars
import json
import logging
import re
import sys
import time
import uuid
from fastapi import FastAPI
from fastapi.testclient import TestClient

request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        entry = {"level": record.levelname, "logger": record.name, "msg": record.getMessage(),
                 "request_id": request_id.get()}
        entry.update(getattr(record, "fields", {}))                 # log.info(..., extra={"fields": {...}})
        if record.exc_info:
            entry["error"] = repr(record.exc_info[1])
        return json.dumps(entry)

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler], force=True)
logging.getLogger("httpx").setLevel(logging.WARNING)                # the test client's own request chatter
log = logging.getLogger("orders")
access = logging.getLogger("access")

SAFE_ID = re.compile(r"[A-Za-z0-9-]{8,64}")

class RequestContext:                                               # pure ASGI middleware (lesson 12)
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)
        incoming = dict(scope["headers"]).get(b"x-request-id", b"").decode()
        rid = incoming if SAFE_ID.fullmatch(incoming) else uuid.uuid4().hex[:12]   # never trust it blindly
        token = request_id.set(rid)
        status, start = 500, time.perf_counter()

        async def send_with_id(message):
            nonlocal status
            if message["type"] == "http.response.start":
                status = message["status"]
                message["headers"] = [*message["headers"], (b"x-request-id", rid.encode())]
            await send(message)
        try:
            await self.app(scope, receive, send_with_id)
        finally:
            route = getattr(scope.get("route"), "path", "unmatched")
            access.info("request", extra={"fields": {
                "method": scope["method"], "route": route, "status": status,
                "duration_ms": round((time.perf_counter() - start) * 1000, 1)}})
            request_id.reset(token)

app = FastAPI()
app.add_middleware(RequestContext)

def load_order(order_id: int) -> dict:                              # deep in the service layer
    log.info("loading order", extra={"fields": {"order_id": order_id}})
    return {"id": order_id}

@app.get("/orders/{order_id}")
async def read(order_id: int):
    return load_order(order_id)

@app.get("/report")
def report():                                                       # sync: runs in the thread pool
    log.info("building report")
    return {"ok": True}

@app.get("/boom")
async def boom():
    try:
        1 / 0
    except ZeroDivisionError:
        log.exception("report failed")
        raise

client = TestClient(app, raise_server_exceptions=False)
r = client.get("/orders/42", headers={"X-Request-ID": "checkout-7f3a9c21"})
print("response header:", r.headers["x-request-id"])
client.get("/report")
client.get("/boom")
client.get("/orders/1", headers={"X-Request-ID": "x\nFAKE LOG LINE"})

Output

{"level": "INFO", "logger": "orders", "msg": "loading order", "request_id": "checkout-7f3a9c21", "order_id": 42}
{"level": "INFO", "logger": "access", "msg": "request", "request_id": "checkout-7f3a9c21", "method": "GET", "route": "/orders/{order_id}", "status": 200, "duration_ms": 1.5}
response header: checkout-7f3a9c21
{"level": "INFO", "logger": "orders", "msg": "building report", "request_id": "48e9c009bee5"}
{"level": "INFO", "logger": "access", "msg": "request", "request_id": "48e9c009bee5", "method": "GET", "route": "/report", "status": 200, "duration_ms": 1.8}
{"level": "ERROR", "logger": "orders", "msg": "report failed", "request_id": "118d3fd82abb", "error": "ZeroDivisionError('division by zero')"}
{"level": "INFO", "logger": "access", "msg": "request", "request_id": "118d3fd82abb", "method": "GET", "route": "/boom", "status": 500, "duration_ms": 0.7}
{"level": "INFO", "logger": "orders", "msg": "loading order", "request_id": "f873d66cdbb5", "order_id": 1}
{"level": "INFO", "logger": "access", "msg": "request", "request_id": "f873d66cdbb5", "method": "GET", "route": "/orders/{order_id}", "status": 200, "duration_ms": 0.6}

2. Metrics: rate, errors, duration

Metrics are numbers aggregated in the process and scraped every 15 seconds or so. For every HTTP service, the RED metrics come first: request Rate, Error rate, and Duration as a histogram (so that p50, p95 and p99 can be computed). The number one mistake is a label with unbounded values, such as a raw URL path or user id. Each distinct label set is a separate time series, and too many of them take down the metrics server. This block implements Prometheus’ text format by hand to show what prometheus_client does:

RED metrics with safe labels

import bisect
import time
from collections import defaultdict
from fastapi import FastAPI, HTTPException
from fastapi.responses import PlainTextResponse
from fastapi.testclient import TestClient

BUCKETS = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, float("inf"))
requests_total: dict[tuple, int] = defaultdict(int)
duration_buckets: dict[tuple, list[int]] = defaultdict(lambda: [0] * len(BUCKETS))
duration_sum: dict[tuple, float] = defaultdict(float)
raw_paths_seen: set[str] = set()                        # what a naive "path" label would create

class Metrics:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http" or scope["path"] == "/metrics":
            return await self.app(scope, receive, send)
        status, start = 500, time.perf_counter()

        async def capture(message):
            nonlocal status
            if message["type"] == "http.response.start":
                status = message["status"]
            await send(message)
        try:
            await self.app(scope, receive, capture)
        finally:
            elapsed = time.perf_counter() - start
            route = getattr(scope.get("route"), "path", "unmatched")
            raw_paths_seen.add(scope["path"])
            requests_total[(scope["method"], route, str(status))] += 1
            key = (scope["method"], route)
            duration_buckets[key][bisect.bisect_left(BUCKETS, elapsed)] += 1
            duration_sum[key] += elapsed

app = FastAPI()
app.add_middleware(Metrics)

@app.get("/orders/{order_id}")
async def read(order_id: int):
    if order_id % 50 == 0:
        raise HTTPException(404)
    return {"id": order_id}

@app.get("/metrics", include_in_schema=False)
def metrics():
    lines = ["# TYPE http_requests_total counter"]
    lines += [f'http_requests_total{{method="{m}",route="{r}",status="{s}"}} {n}'
              for (m, r, s), n in sorted(requests_total.items())]
    lines.append("# TYPE http_request_duration_seconds histogram")
    for (m, r), counts in sorted(duration_buckets.items()):
        running = 0
        for le, n in zip(BUCKETS, counts):
            running += n
            le_text = "+Inf" if le == float("inf") else le
            lines.append(f'http_request_duration_seconds_bucket{{method="{m}",route="{r}",le="{le_text}"}} {running}')
        lines.append(f'http_request_duration_seconds_count{{method="{m}",route="{r}"}} {running}')
    return PlainTextResponse("\n".join(lines) + "\n")

client = TestClient(app)
for order_id in range(1, 1001):
    client.get(f"/orders/{order_id}")
for probe in ("/wp-login.php", "/.env", "/admin"):
    client.get(probe)

text = client.get("/metrics").text
for line in text.splitlines():
    if "bucket" not in line or 'le="0.005"' in line or 'le="+Inf"' in line:
        print(line)
print(f"\nseries with a route label: {len(requests_total)}   with a raw path label it would be: {len(raw_paths_seen)}")

Output

# TYPE http_requests_total counter
http_requests_total{method="GET",route="/orders/{order_id}",status="200"} 980
http_requests_total{method="GET",route="/orders/{order_id}",status="404"} 20
http_requests_total{method="GET",route="unmatched",status="404"} 3
# TYPE http_request_duration_seconds histogram
http_request_duration_seconds_bucket{method="GET",route="/orders/{order_id}",le="0.005"} 1000
http_request_duration_seconds_bucket{method="GET",route="/orders/{order_id}",le="+Inf"} 1000
http_request_duration_seconds_count{method="GET",route="/orders/{order_id}"} 1000
http_request_duration_seconds_bucket{method="GET",route="unmatched",le="0.005"} 3
http_request_duration_seconds_bucket{method="GET",route="unmatched",le="+Inf"} 3
http_request_duration_seconds_count{method="GET",route="unmatched"} 3

series with a route label: 3   with a raw path label it would be: 1003

3. Distributed tracing with OpenTelemetry

A trace is a tree of spans, each a timed operation. The traceparent header (W3C Trace Context) carries the trace id from service to service, so one trace can show the gateway, your API, the database and the payment service together. This block uses the real OpenTelemetry SDK. It continues a trace started by a caller, creates child spans for a query and an outgoing HTTP call, and forwards the trace to that downstream service.

OpenTelemetry SDK: server span, child spans, propagation

import asyncio
import httpx
from fastapi import FastAPI, Request
from fastapi.testclient import TestClient
from opentelemetry import propagate, trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.trace import SpanKind

exporter = InMemorySpanExporter()                    # production: an OTLP exporter to a collector
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("orders-api")

class Tracing:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)
        headers = {k.decode(): v.decode() for k, v in scope["headers"]}
        parent = propagate.extract(headers)                             # continue the caller's trace
        with tracer.start_as_current_span(scope["method"], context=parent, kind=SpanKind.SERVER) as span:
            span.set_attribute("http.request.method", scope["method"])
            span.set_attribute("url.path", scope["path"])

            async def capture(message):
                if message["type"] == "http.response.start":
                    span.set_attribute("http.response.status_code", message["status"])
                await send(message)
            await self.app(scope, receive, capture)
            if route := getattr(scope.get("route"), "path", None):
                span.set_attribute("http.route", route)
                span.update_name(f"{scope['method']} {route}")          # low-cardinality span name

def payment_service(request: httpx.Request) -> httpx.Response:        # the downstream service
    return httpx.Response(200, json={"received_traceparent": request.headers.get("traceparent")})

app = FastAPI()
app.add_middleware(Tracing)

@app.get("/orders/{order_id}")
async def read(order_id: int, request: Request):
    with tracer.start_as_current_span("SELECT orders", kind=SpanKind.CLIENT) as span:
        span.set_attribute("db.system.name", "postgresql")
        await asyncio.sleep(0.02)                                      # pretend query
    with tracer.start_as_current_span("GET payments", kind=SpanKind.CLIENT):
        headers: dict[str, str] = {}
        propagate.inject(headers)                                      # adds traceparent for the next hop
        async with httpx.AsyncClient(transport=httpx.MockTransport(payment_service)) as http:
            downstream = (await http.get("http://payments/charges", headers=headers)).json()
    return {"id": order_id, **downstream}

caller_trace = "4bf92f3577b34da6a3ce929d0e0e4736"
body = TestClient(app).get("/orders/7", headers={"traceparent": f"00-{caller_trace}-00f067aa0ba902b7-01"}).json()

spans = exporter.get_finished_spans()
by_id = {s.context.span_id: s for s in spans}
def depth(s):
    return 0 if s.parent is None or s.parent.span_id not in by_id else 1 + depth(by_id[s.parent.span_id])
for s in sorted(spans, key=lambda s: s.start_time):
    ms = (s.end_time - s.start_time) / 1e6
    print(f"{'  ' * depth(s)}{s.name:<26} {s.kind.name:<6} {ms:6.1f} ms  trace={format(s.context.trace_id, '032x')[:8]}…")
server = next(s for s in spans if s.kind == SpanKind.SERVER)
print("\nserver span attributes:", dict(server.attributes))
print("same trace as the caller:", format(server.context.trace_id, "032x") == caller_trace)
print("downstream got traceparent:", body["received_traceparent"][:36] + "…")

Output

GET /orders/{order_id}     SERVER   33.4 ms  trace=4bf92f35…
  SELECT orders              CLIENT   29.5 ms  trace=4bf92f35…
  GET payments               CLIENT    1.7 ms  trace=4bf92f35…

server span attributes: {'http.request.method': 'GET', 'url.path': '/orders/7', 'http.response.status_code': 200, 'http.route': '/orders/{order_id}'}
same trace as the caller: True
downstream got traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-…

4. Liveness and readiness

Orchestrators ask two different questions, and mixing them up causes outages. Liveness: “is this process stuck, so restart it?” It must not check dependencies. If the database is down, restarting every API pod makes things worse. Readiness: “should this instance get traffic right now?” It checks critical dependencies with a short timeout, and it turns false during shutdown so the load balancer drains the pod first (part 23).

livez vs readyz

import asyncio
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient

state = {"db_up": True, "draining": False}

async def check_database() -> None:
    if not state["db_up"]:
        await asyncio.sleep(5)                                 # a hung connection, not a quick error

app = FastAPI()

@app.get("/livez", include_in_schema=False)
async def livez():
    return {"status": "ok"}                                    # the event loop answered: that is all

@app.get("/readyz", include_in_schema=False)
async def readyz():
    if state["draining"]:
        return JSONResponse({"status": "draining"}, status_code=503)
    try:
        await asyncio.wait_for(check_database(), timeout=0.5)  # probes must be fast
    except TimeoutError:
        return JSONResponse({"status": "unready", "database": "timeout"}, status_code=503)
    return {"status": "ready", "database": "ok"}

with TestClient(app) as c:
    for label, change in [("healthy", {}), ("database hangs", {"db_up": False}),
                          ("shutting down", {"db_up": True, "draining": True})]:
        state.update(change)
        live, ready = c.get("/livez"), c.get("/readyz")
        print(f"{label:<15} livez={live.status_code}  readyz={ready.status_code} {ready.json()}")

Output

healthy         livez=200  readyz=200 {'status': 'ready', 'database': 'ok'}
database hangs  livez=200  readyz=503 {'status': 'unready', 'database': 'timeout'}
shutting down   livez=200  readyz=503 {'status': 'draining'}

Recap