FastAPI Tracing Middleware
Introduction
Tracing is a critical aspect of monitoring in modern web applications, especially in microservices architectures. It allows developers to track the flow of requests through their application, measure performance, and identify bottlenecks. In FastAPI, we can implement tracing through middleware components that intercept requests and responses.
In this tutorial, you'll learn:
- What tracing is and why it's important
- How to implement basic tracing middleware in FastAPI
- How to use OpenTelemetry for advanced tracing
- How to visualize and analyze tracing data
What is Tracing?
Tracing follows the journey of a request as it travels through different parts of your application. It measures how long each component takes to process the request, creating a timeline of events known as a "trace." Each trace consists of multiple "spans," which represent operations within your application.
Basic Tracing Middleware
Let's start by implementing a simple tracing middleware to log request processing times:
import time
from fastapi import FastAPI, Request
from fastapi.middleware.base import BaseHTTPMiddleware
from typing import Callable
import logging
logger = logging.getLogger(__name__)
class TracingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next: Callable):
# Record the start time
start_time = time.time()
# Get the route path for logging
route_path = request.url.path
method = request.method
# Process the request
response = await call_next(request)
# Calculate processing time
process_time = time.time() - start_time
# Log the information
logger.info(f"{method} {route_path} processed in {process_time:.4f} seconds")
# Add timing header to response
response.headers["X-Process-Time"] = str(process_time)
return response
# Create FastAPI app instance
app = FastAPI()
# Add the tracing middleware
app.add_middleware(TracingMiddleware)
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.get("/slow")
async def slow_route():
# Simulate slow operation
time.sleep(1)
return {"message": "This was a slow request"}
In this example:
- We created a custom middleware class that inherits from
BaseHTTPMiddleware - We implemented the
dispatchmethod to intercept requests and responses - We measured and logged the processing time for each request
- We added the processing time as a custom header in the response
When you send requests to this API, you'll see log entries like:
INFO:__main__:GET / processed in 0.0023 seconds
INFO:__main__:GET /slow processed in 1.0037 seconds
Advanced Tracing with OpenTelemetry
For production applications, you'll want a more robust tracing solution. OpenTelemetry is an industry standard for tracing that works with many visualization and analysis tools.
Let's implement tracing with OpenTelemetry:
from fastapi import FastAPI
import time
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
# Set up OpenTelemetry
trace.set_tracer_provider(TracerProvider())
# Configure exporter to send data to Jaeger
otlp_exporter = OTLPSpanExporter(endpoint="http://localhost:4317")
span_processor = BatchSpanProcessor(otlp_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)
# Create a tracer
tracer = trace.get_tracer(__name__)
# Create FastAPI app
app = FastAPI()
# Instrument FastAPI application
FastAPIInstrumentor.instrument_app(app)
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.get("/users/{user_id}")
async def get_user(user_id: int):
# Create a custom span for database operation
with tracer.start_as_current_span("database_query") as span:
# Add attributes to the span
span.set_attribute("user.id", user_id)
# Simulate database query
time.sleep(0.1)
# Simulate another operation
with tracer.start_as_current_span("data_processing"):
time.sleep(0.05)
return {"user_id": user_id, "name": "Test User"}
To use this example, you'll need to install the required packages:
pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp opentelemetry-instrumentation-fastapi
You'll also need a backend to collect and visualize traces. Jaeger is a popular option:
-
Run Jaeger with Docker:
bashdocker run -d --name jaeger \
-e COLLECTOR_OTLP_ENABLED=true \
-p 16686:16686 \
-p 4317:4317 \
jaegertracing/all-in-one:latest -
Start your FastAPI application
-
Visit http://localhost:16686 to view traces
Custom Context Propagation
In microservices architectures, you often need to propagate context (like trace IDs) across service boundaries. Here's how to add custom context:
from fastapi import FastAPI, Request, Depends
from fastapi.middleware.base import BaseHTTPMiddleware
import uuid
from contextvars import ContextVar
# Create a context variable for request ID
request_id_ctx_var = ContextVar("request_id", default=None)
def get_request_id():
return request_id_ctx_var.get()
class RequestTracingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# Generate or extract request ID
request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4())
# Store in context variable
request_id_ctx_var.set(request_id)
# Process request
response = await call_next(request)
# Add request ID to response headers
response.headers["X-Request-ID"] = request_id
return response
app = FastAPI()
app.add_middleware(RequestTracingMiddleware)
@app.get("/endpoint")
async def endpoint(request_id: str = Depends(get_request_id)):
# The request_id is available throughout the request lifecycle
return {"request_id": request_id, "message": "Processing request"}
This middleware ensures that a request ID is available throughout the request processing, and can be included in logs, traces, and outgoing requests to other services.
Real-World Example: Tracing with Performance Monitoring
Let's build a more comprehensive example that combines tracing with performance monitoring:
import time
from fastapi import FastAPI, Request, HTTPException, status
from fastapi.middleware.base import BaseHTTPMiddleware
from pydantic import BaseModel
import logging
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import statistics
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RequestMetrics(BaseModel):
path: str
method: str
status_code: int
duration: float
timestamp: datetime
class MetricsStorage:
def __init__(self, window_size: int = 100):
self.metrics: List[RequestMetrics] = []
self.window_size = window_size
self.path_stats: Dict[str, List[float]] = {}
def add_metric(self, metric: RequestMetrics):
# Add new metric
self.metrics.append(metric)
# Update path statistics
path_key = f"{metric.method} {metric.path}"
if path_key not in self.path_stats:
self.path_stats[path_key] = []
self.path_stats[path_key].append(metric.duration)
# Trim if exceeding window size
if len(self.metrics) > self.window_size:
oldest = self.metrics.pop(0)
oldest_key = f"{oldest.method} {oldest.path}"
if oldest_key in self.path_stats and self.path_stats[oldest_key]:
self.path_stats[oldest_key].pop(0)
def get_stats(self):
stats = {}
for path, durations in self.path_stats.items():
if durations:
stats[path] = {
"count": len(durations),
"avg_ms": statistics.mean(durations) * 1000,
"p95_ms": statistics.quantiles(durations, n=20)[18] * 1000 if len(durations) >= 20 else None,
"min_ms": min(durations) * 1000,
"max_ms": max(durations) * 1000
}
return stats
def get_slow_requests(self, threshold: float = 1.0):
return [m for m in self.metrics if m.duration >= threshold]
class TracingMiddleware(BaseHTTPMiddleware):
def __init__(self, app, metrics_storage: MetricsStorage):
super().__init__(app)
self.metrics_storage = metrics_storage
async def dispatch(self, request: Request, call_next):
# Start timer
start_time = time.time()
# Process request
try:
response = await call_next(request)
status_code = response.status_code
except Exception as e:
# Log the exception
logger.exception(f"Error processing request: {str(e)}")
status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
raise
finally:
# Calculate duration
duration = time.time() - start_time
# Store metrics
metric = RequestMetrics(
path=request.url.path,
method=request.method,
status_code=status_code,
duration=duration,
timestamp=datetime.now()
)
self.metrics_storage.add_metric(metric)
# Log slow requests
if duration > 1.0:
logger.warning(f"Slow request: {request.method} {request.url.path} took {duration:.4f}s")
# Add timing header
response.headers["X-Process-Time"] = f"{duration:.6f}"
return response
# Create FastAPI app
app = FastAPI()
# Create metrics storage
metrics_storage = MetricsStorage()
# Add middleware
app.add_middleware(TracingMiddleware, metrics_storage=metrics_storage)
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.get("/slow")
async def slow_route():
# Simulate slow operation
time.sleep(1.2)
return {"message": "This was a slow request"}
@app.get("/error")
async def error_route():
# Simulate an error
raise HTTPException(status_code=500, detail="Internal error")
@app.get("/metrics")
async def get_metrics():
return {
"stats": metrics_storage.get_stats(),
"slow_requests": metrics_storage.get_slow_requests()
}
In this example:
- We created a
MetricsStorageclass to store and analyze request metrics - Our middleware captures request duration, path, and status code
- We log warnings for slow requests
- We expose a
/metricsendpoint to view performance statistics
To test this example:
- Make several requests to the different endpoints
- Check the
/metricsendpoint to see the statistics - Notice how slow requests are automatically identified
Visualizing Traces
For production applications, you'll want to visualize traces. Here are some popular options:
- Jaeger: Open-source, end-to-end distributed tracing
- Zipkin: Another popular open-source tracing system
- Datadog APM: Commercial solution with extensive features
- New Relic: Commercial APM with distributed tracing support
Most of these tools can be integrated with OpenTelemetry, providing a standardized way to collect and export traces.
Summary
In this tutorial, you've learned:
- How to implement basic tracing middleware in FastAPI
- How to integrate with OpenTelemetry for distributed tracing
- How to propagate context across service boundaries
- How to build advanced metrics collection and analysis
- Options for visualizing traces in production environments
Tracing middleware is an essential tool for monitoring and debugging FastAPI applications, especially in production environments. By implementing proper tracing, you can identify performance bottlenecks, track request flows, and ensure your application is running efficiently.
Additional Resources
- OpenTelemetry Documentation
- Jaeger Tracing
- FastAPI Documentation on Middleware
- Distributed Tracing in Microservices
Exercises
- Extend the basic tracing middleware to include request body size and response time breakdown
- Implement a middleware that exports trace data to a CSV file
- Create a middleware that detects and alerts on unusual response times (outliers)
- Set up a complete tracing system with Jaeger and OpenTelemetry
- Implement context propagation between two FastAPI services using HTTP headers
Happy coding!
If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)