Userverse includes built-in support for Prometheus metrics, optional OpenTelemetry tracing, and structured HTTP request logging via LogMiddleware.
Prometheus metrics
A /metrics endpoint is registered automatically by create_app. It returns Prometheus-compatible metrics in the standard text format.
@app.get("/metrics")
def metrics():
return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
Unhandled exceptions are tracked with a dedicated counter:
UNHANDLED_EXCEPTIONS = Counter(
"unhandled_exceptions_total",
"Total number of unhandled exceptions",
["method", "endpoint"],
)
Scrape the endpoint from your Prometheus instance:
scrape_configs:
- job_name: userverse
static_configs:
- targets: ['localhost:8500']
metrics_path: /metrics
OpenTelemetry tracing
The setup_otel function in app/middleware/otel.py configures tracing with an OTLP HTTP exporter. It instruments FastAPI automatically using FastAPIInstrumentor.
def setup_otel(
app,
service_name: str = "fastapi-app",
collector_endpoint="http://otel-collector:4318/v1/traces",
):
resource = Resource(attributes={SERVICE_NAME: service_name})
tracer_provider = TracerProvider(resource=resource)
otlp_exporter = OTLPSpanExporter(endpoint=collector_endpoint)
span_processor = BatchSpanProcessor(otlp_exporter)
tracer_provider.add_span_processor(span_processor)
trace.set_tracer_provider(tracer_provider)
FastAPIInstrumentor.instrument_app(app)
RequestsInstrumentor().instrument()
OpenTelemetry tracing is commented out by default in app/main.py. To enable it, uncomment the setup_otel(app) call inside create_app.
# setup_otel(app) ← uncomment to enable tracing
app.add_middleware(LogMiddleware)
Request logging
LogMiddleware is applied to every request. It records the HTTP method, URL, client IP, user agent, response status code, and duration in milliseconds.
app/middleware/logging.py
class LogMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
start_time = time.time()
...
logger.info(
"Request handled",
extra={
"extra": {
"method": request.method,
"url": str(request.url),
"client": request.client.host if request.client else None,
"user_agent": request.headers.get("user-agent"),
"status_code": response.status_code,
"duration_ms": round(process_time, 2),
}
},
)
Failed requests are logged at the ERROR level with the exception message included.