How to add middleware to the Fast API to create metrics to track time spent and requests made?

Question

I am adding Middleware to my Fast API app to create Prometheus metrics to get the Processing Time and Number of requests per route. Can someone tell me what I am missing?

Using Prometheus/client_python

This is my middleware.

import time

from fastapi import Request
from prometheus_client import Summary, make_asgi_app, CollectorRegistry
from starlette.middleware.base import BaseHTTPMiddleware

registry = CollectorRegistry()
metrics_app = make_asgi_app(registry=registry)
summary_metrics = {}


class MyMiddleware:

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

        route_names = [r.name for r in app.routes]
        if len(set(route_names)) != len(route_names):
            raise Exception("Route names must be unique")

        for route_name in route_names:
            if route_name is not None:
                summary_metrics[route_name] = Summary(route_name, f'{route_name} processing time', registry=registry)

    async def __call__(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        if request.scope.get('route'):
            summary_metrics.get(request.scope.get('route')).observe(process_time)
        return response

This is my app.py

app.include_router(some_valid_router)
app.mount("/metrics", metrics_app)
my_middleware = MyMiddleware(app)
app.add_middleware(BaseHTTPMiddleware, dispatch=my_middleware)


if __name__ == "__main__":
    LOGGER.info("Starting up the application")
    uvicorn_logger = (
        UVICORN_LOG_CONFIG if config["CUSTOM_UVICORN_LOGGING"].lower() == "y" else None
    )
    if uvicorn_logger:
        uvicorn.run(
            "app:app",
            host="0.0.0.0",
            port=int(config["PORT"]),
            reload=config["DEBUG"],
            log_config=uvicorn_logger,
        )
    else:
        uvicorn.run(
            "app:app", host="0.0.0.0", port=int(config["PORT"]), reload=config["DEBUG"]
        )

I am getting the ValueError: Duplicated timeseries in CollectorRegistry: error if I have the my_middleware = MyMiddleware(app) app.add_middleware(BaseHTTPMiddleware, dispatch=my_middleware) outside of __main__, and my metrics endpoint is blank if I move those two lines to the starting of __main__.

I can see the overall requests count and total time spent if I change my Middleware to below and move app.add_middleware to the outside of __main__.

import time

from fastapi import Request
from prometheus_client import Summary, make_asgi_app, CollectorRegistry
from starlette.middleware.base import BaseHTTPMiddleware

registry = CollectorRegistry()
metrics_app = make_asgi_app(registry=registry)
summary_metrics = {}
REQUEST_TIME = Summary('request_processing_seconds', 'Time spent processing request', registry=registry)


class MyMiddleware:

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

    async def __call__(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        REQUEST_TIME.observe(process_time)
        return response

Can someone tell me what I am doing wrong? This is my first interaction with the Fast API. Please let me know if my approach needs to be corrected or if there are better ways to get the Processing Time and Number of requests per route metrics.

Thanks!!

Answer 1

ValueError: Duplicated timeseries in CollectorRegistry  

means that at leat two metrics were added with same metric name.
You should add the metric to the registry only one time.

In my case, I declared the metric in the util directory. (like util/prometheus/metrics.py)
And then import it ( from util.prometheus import metrics) and set the label and values in every api.

check the below code.

# src/prometheus/metrics.py
from prometheus_client import  Gauge 


metric_name = "metrics_info"
metric_description = "metric test"
metric_labels = ["status_code","os","handler"]
metric_gauge = Gauge(
    name=metric_name,
    documentation = metric_description,
    labelnames = metric_labels
) 

# src/api/test.py
from ..prometheus import metrics

@router.get("/metrictest")
def test():
... do sth 
metrics.metric_gauge.labels(status_code="400",os="someos",handler="/api/test/metrictest").set(1)        
... do sth else