FastAPI添加Prometheus中间件实现路由级指标遇问题求助

使用依赖 #

prometheus/client_python

初始中间件代码 #

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

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"]
        )

问题详情 #

• 将my_middleware = MyMiddleware(app)和app.add_middleware(BaseHTTPMiddleware, dispatch=my_middleware)放在__main__外部时，触发ValueError: Duplicated timeseries in CollectorRegistry: 错误
• 将上述两行移到__main__开头时，/metrics端点返回内容为空
• 改用全局Summary实现的中间件（代码如下），放在__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

问题原因分析 #

1. 重复注册指标：当开启reload调试模式时，FastAPI会重复加载模块，导致MyMiddleware.__init__多次执行，重复创建同名的Summary指标，触发Prometheus注册表的重复时间序列校验错误。
2. 中间件初始化时机错误：将中间件初始化移到__main__内时，uvicorn启动的是app:app（即模块中的app实例），而__main__内的初始化操作仅作用于当前脚本的app实例，并未关联到uvicorn加载的实例，导致指标未被注册到有效注册表中，/metrics端点无数据。

解决方案：路由级指标的正确实现 #

推荐使用带标签的Summary（符合Prometheus最佳实践），无需提前遍历路由创建指标，通过标签动态区分不同路由，同时避免重复注册问题。

修正后的中间件代码 #

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

# 全局注册表与metrics应用
registry = CollectorRegistry()
metrics_app = make_asgi_app(registry=registry)

# 带路由标签的Summary，统一管理所有路由的处理时间指标
REQUEST_TIME = Summary(
    'request_processing_seconds',
    'Time spent processing request',
    ['route'],  # 用route标签区分不同路由
    registry=registry
)

class RouteMetricsMiddleware(BaseHTTPMiddleware):
    # 使用BaseHTTPMiddleware标准的dispatch方法，符合框架规范
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        
        # 获取当前路由名称，无匹配路由时用unknown_route
        route_name = request.scope.get('route').name if request.scope.get('route') else "unknown_route"
        # 给对应路由的标签指标记录处理时间
        REQUEST_TIME.labels(route=route_name).observe(process_time)
        
        return response

修正后的app.py代码 #

app.include_router(some_valid_router)
app.mount("/metrics", metrics_app)
# 直接添加中间件，框架自动处理实例化
app.add_middleware(RouteMetricsMiddleware)

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"]
        )

方案优势 #

• 符合Prometheus设计理念：通过标签区分路由，避免创建大量独立指标
• 彻底解决重复注册问题：仅初始化一个带标签的Summary，无论模块是否被重复加载
• 遵循FastAPI/Starlette中间件规范：使用BaseHTTPMiddleware的dispatch方法，避免自定义__call__带来的兼容性问题
• 自动适配路由变化：无需提前遍历路由，新增路由时自动生成对应标签的指标

备选方案：提前创建路由专属指标（不推荐） #

若坚持为每个路由创建独立指标，需在初始化时检查注册表是否已存在该指标，避免重复注册：

# 修改MyMiddleware.__init__方法
def __init__(self, app):
    self.app = app
    route_names = [r.name for r in app.routes if r.name is not None]
    if len(set(route_names)) != len(route_names):
        raise Exception("Route names must be unique")
    
    for route_name in route_names:
        # 检查注册表中是否已存在同名指标，不存在才创建
        metric_exists = any(metric.name == route_name for metric in registry.collect())
        if not metric_exists:
            summary_metrics[route_name] = Summary(route_name, f'{route_name} processing time', registry=registry)

内容的提问来源于stack exchange，提问作者user3595026