ASGI Middleware for Request Logging

Problem Statement

Standard web server access logs (like default Uvicorn or Gunicorn output) often fall short in cloud-native microservices environments. They typically output plain text strings that are difficult to parse, lack request context (such as unique correlation IDs), and do not report crucial performance metrics like exact downstream response latency or response body sizes. Furthermore, unhandled application errors often bypass standard access logs, leaving operational blind spots.

This script implements an ASGI Middleware for Structured Request Logging. It wraps any standard ASGI application (Starlette, FastAPI, Quartz, etc.) to capture detailed metrics for every HTTP interaction—including request processing times, status codes, client IPs, user agents, and generated or propagated correlation IDs (X-Request-ID). It formats these entries into structured JSON output ready for ingestion by log collectors like Datadog, ELK, or CloudWatch, while cleanly filtering out noisy noise endpoints like /healthz.

How It Works

ASGI (Asynchronous Server Gateway Interface) operates on a standard tuple-based signature: app(scope, receive, send). The middleware implements this protocol as an interceptor wrapped around the downstream application:

1. Request Interception: When an incoming HTTP request arrives (scope["type"] == "http"), the middleware extracts incoming headers (e.g., X-Request-ID, User-Agent), client address details, and the endpoint path.
2. Correlation ID Generation: If no request ID header is present in the request, the middleware generates a UUID4 correlation ID and attaches it to both the logging context and the outgoing response headers.
3. Response Interception: The middleware wraps the ASGI send callable with an internal collector function. As the downstream application emits http.response.start and http.response.body messages, the wrapper captures the HTTP status code and accumulates total bytes sent without altering the payload.
4. Execution Timing & Logging: Using time.perf_counter(), the middleware measures exact execution latency. Whether the application completes normally or raises an unhandled exception, a finally block guarantees that structured log data is serialized and emitted to sys.stdout.

Full Script

#!/usr/bin/env python3
"""
ASGI Middleware for Request Logging

A high-performance, structured JSON request logging middleware for ASGI applications.
Captures performance timing, status codes, client metadata, and correlation IDs.
"""

import json
import logging
import os
import sys
import time
import uuid
from typing import Any, Callable, Coroutine
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route
from starlette.types import ASGIApp, Receive, Scope, Send

# ==============================================================================
# Configuration
# ==============================================================================

LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()
LOG_FORMAT = os.environ.get("LOG_FORMAT", "json").lower()  # "json" or "text"
EXCLUDE_PATHS = set(
    path.strip()
    for path in os.environ.get("EXCLUDE_PATHS", "/healthz,/metrics").split(",")
    if path.strip()
)
REQUEST_ID_HEADER = os.environ.get("REQUEST_ID_HEADER", "X-Request-ID").lower()

# Configure root logger
logger = logging.getLogger("asgi.access")
logger.setLevel(LOG_LEVEL)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(handler)
logger.propagate = False


# ==============================================================================
# Middleware Implementation
# ==============================================================================

class RequestLoggingMiddleware:
    """ASGI Middleware that intercepts HTTP requests and emits structured log entries."""

    def __init__(
        self,
        app: ASGIApp,
        exclude_paths: set[str] | None = None,
        log_as_json: bool = True,
    ) -> None:
        self.app = app
        self.exclude_paths = exclude_paths if exclude_paths is not None else EXCLUDE_PATHS
        self.log_as_json = log_as_json

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        # Pass non-HTTP scopes (e.g., websocket, lifespan) directly to downstream app
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        path: str = scope.get("path", "")

        # Skip excluded endpoints (e.g., health checks) to reduce log noise
        if path in self.exclude_paths:
            await self.app(scope, receive, send)
            return

        start_time = time.perf_counter()
        headers: list[tuple[bytes, bytes]] = scope.get("headers", [])

        # Extract client IP address safely
        client: tuple[str, int] | None = scope.get("client")
        client_ip = client[0] if client else "unknown"

        # Helper function to parse header values from scope tuple list
        request_id = self._get_header_value(headers, REQUEST_ID_HEADER)
        if not request_id:
            request_id = uuid.uuid4().hex

        user_agent = self._get_header_value(headers, "user-agent") or "unknown"
        method: str = scope.get("method", "UNKNOWN")

        # Response details state (captured during wrapped send)
        status_code = 500  # Default to 500 if downstream crashes before responding
        response_bytes = 0

        async def wrapped_send(message: dict[str, Any]) -> None:
            nonlocal status_code, response_bytes

            if message["type"] == "http.response.start":
                status_code = message.get("status", 500)

                # Inject correlation ID into outgoing response headers
                response_headers = list(message.get("headers", []))
                header_key = REQUEST_ID_HEADER.encode("latin1")
                header_val = request_id.encode("latin1")
                response_headers.append((header_key, header_val))
                message["headers"] = response_headers

            elif message["type"] == "http.response.body":
                body = message.get("body", b"")
                response_bytes += len(body)

            await send(message)

        # Process the request and measure total elapsed execution time
        error_occurred = False
        try:
            await self.app(scope, receive, wrapped_send)
        except Exception:
            error_occurred = True
            raise
        finally:
            duration_ms = round((time.perf_counter() - start_time) * 1000, 2)
            self._log_request(
                method=method,
                path=path,
                status_code=status_code,
                duration_ms=duration_ms,
                client_ip=client_ip,
                user_agent=user_agent,
                request_id=request_id,
                response_bytes=response_bytes,
                error=error_occurred,
            )

    @staticmethod
    def _get_header_value(headers: list[tuple[bytes, bytes]], key: str) -> str:
        """Extract a UTF-8/Latin1 decoded header value from raw scope bytes."""
        target_key = key.lower().encode("latin1")
        for k, v in headers:
            if k.lower() == target_key:
                return v.decode("latin1")
        return ""

    def _log_request(
        self,
        method: str,
        path: str,
        status_code: int,
        duration_ms: float,
        client_ip: str,
        user_agent: str,
        request_id: str,
        response_bytes: int,
        error: bool,
    ) -> None:
        """Format and write the access log entry."""
        log_data = {
            "request_id": request_id,
            "method": method,
            "path": path,
            "status": status_code,
            "duration_ms": duration_ms,
            "client_ip": client_ip,
            "bytes_sent": response_bytes,
            "user_agent": user_agent,
        }

        # Select appropriate log level based on response status
        if status_code >= 500 or error:
            log_fn = logger.error
        elif status_code >= 400:
            log_fn = logger.warning
        else:
            log_fn = logger.info

        if self.log_as_json:
            log_fn(json.dumps(log_data))
        else:
            log_fn(
                f"[{request_id}] {method} {path} -> {status_code} "
                f"({duration_ms}ms, {response_bytes}B) - {client_ip}"
            )


# ==============================================================================
# Sample Starlette Application & Test Runner
# ==============================================================================

async def homepage(request: Request) -> JSONResponse:
    """Sample endpoint returning a standard response."""
    return JSONResponse({"message": "Hello, World!"})


async def healthcheck(request: Request) -> JSONResponse:
    """Ignored endpoint according to EXCLUDE_PATHS default configuration."""
    return JSONResponse({"status": "healthy"})


async def trigger_error(request: Request) -> JSONResponse:
    """Sample endpoint triggering an uncaught exception for testing error handling."""
    raise RuntimeError("Simulated internal database crash!")


# Define routes
routes = [
    Route("/", homepage),
    Route("/healthz", healthcheck),
    Route("/error", trigger_error),
]

# Create Starlette app and wrap with middleware
app = Starlette(routes=routes)
app.add_middleware(
    RequestLoggingMiddleware,
    log_as_json=(LOG_FORMAT == "json"),
)

if __name__ == "__main__":
    import uvicorn

    print(f"Starting server on http://127.0.0.1:8000 (Log Format: {LOG_FORMAT})...")
    uvicorn.run(app, host="127.0.0.1", port=8000, log_config=None)

Configuration

The script uses environment variables to adjust logging parameters without changing code:

Usage Examples

1. Default Run (Structured JSON Logging)

Run the application using standard defaults:

python3 request_logging.py

In another terminal, send a request using curl:

curl -i http://127.0.0.1:8000/

Response output:

HTTP/1.1 200 OK
date: Sun, 18 May 2025 12:00:00 GMT
server: uvicorn
content-length: 27
content-type: application/json
x-request-id: 3b1a8d3e91124df standard header attached...

Stdout output generated by script:

{"request_id": "3b1a8d3e91124df", "method": "GET", "path": "/", "status": 200, "duration_ms": 1.42, "client_ip": "127.0.0.1", "bytes_sent": 27, "user_agent": "curl/7.81.0"}

2. Custom Configuration & Plain Text Mode

Run with plain text log formatting and custom excluded paths:

LOG_FORMAT=text EXCLUDE_PATHS="/healthz,/ping" python3 request_logging.py

Send a request passing an existing custom X-Request-ID:

curl -H "X-Request-ID: trace-abc-123" http://127.0.0.1:8000/

Stdout output generated by script:

[trace-abc-123] GET / -> 200 (0.89ms, 27B) - 127.0.0.1

3. Error Handling Test

Trigger an internal server failure:

curl -i http://127.0.0.1:8000/error

Stdout output generated by script:

{"request_id": "f82b7c622a104b2", "method": "GET", "path": "/error", "status": 500, "duration_ms": 2.15, "client_ip": "127.0.0.1", "bytes_sent": 0, "user_agent": "curl/7.81.0"}

Error Handling

• Uncaught Downstream Exceptions: The middleware executes the downstream request inside a try...finally block. If an unhandled exception occurs (e.g., database connection failure or uncaught runtime error), the exception is re-raised to allow ASGI server error handlers to operate, but the execution time and status code (500) are guaranteed to be logged.
• Missing or Invalid Request Headers: Raw headers from the ASGI scope are parsed using defensive string decoding (latin1), avoiding encoding crashes if clients send non-standard character sets in headers.
• Non-HTTP ASGI Scopes: WebSocket framing (websocket) and lifespan initialization (lifespan) events bypass timing wrappers safely to avoid breaking non-HTTP protocol implementations.

Extending Ideas

1. Header & Body Sanitization: Modify _log_request to redact sensitive headers (such as Authorization or Cookie) or query parameters (e.g., ?token=...) before emitting JSON logs.
2. Prometheus Metrics Generation: Extend wrapped_send to increment Prometheus counter vectors (http_requests_total) and observe latencies in a histogram (http_request_duration_seconds).
3. OpenTelemetry Integration: Extract incoming traceparent headers to attach OpenTelemetry trace context directly into the log entries for unified distributed tracing.
4. Dynamic Log Sampling: Add logic to log 100% of errors (status >= 400) while sampling successful GET requests at 10% volume in high-throughput environments.