如何使用aioprometheus和FastAPI向外部服务添加指标？

Question

我正在尝试在使用

aioprometheus

构建的应用程序中使用

FastAPI

将指标添加到外部服务。这是我想要实现的目标的简化示例。

假设我有一个包装应用程序类：

from aioprometheus import Registry, Counter, Histogram
from fastapi import FastAPI

class App:
    def __init__(self, ...):
        self.registry = Registry()
        self.counter = Counter(
            name="counts", doc="request counts"
        )
        self.latency = Histogram(
            name="latency",
            doc="request latency",
            buckets=[0.1, 0.5, 1, 1.5, 2]
        )

        self.app = FastAPI()
        self._metrics()

    def _metrics(self):
        # Counter metrics
        @self.app.middleware("http")
        async def counter_metrics(request, call_next):
            response = await call_next(request)
            self.counter.inc(
                {"path": str(request.url.path), "status": response.status_code}
            )
            return response

        # Latency metrics
        @self.app.middleware("http")
        async def latency_metrics(request, call_next):
            start = time.time()
            response = await call_next(request)
            total_duration = time.time() - start
            self.latency.observe(
                {"path": str(request.url.path)}, total_duration
            )
            return response
        
        @self.app.on_event("startup")
        async def startup():
            self.app.include_router(some_router(...))

        self.registry.register(self.counter)
        self.registry.register(self.latency)

基本上，我已经启动了

Registry

、

Counter

和

Histogram

。在

_metrics

中，我有

Counter

和

Histogram

具体逻辑，稍后添加到

Registry

中。当调用

some_router

中的端点时，这将发挥其魔力并捕获指标（这很好！我想保留它，以及拥有外部服务指标）。

但是，假设我从

some_router

调用外部服务：

from fastapi import APIRouter

def some_router():
    router = APIRouter()

    @router.get("/some_router")
    async def some_router():
        response = await external_service()

        return response

    return router

在这种情况下，我如何专门向

external_service

添加指标，即该特定外部服务的延迟？

Answer 1

选项1

根据 documentation，您需要使用通用 app.state 属性将指标附加到

app 实例

（另请参阅 Starlette 的

State

类的实现），以便可以轻松访问它们在路由处理程序中 - 因为指标通常是在与其使用位置不同的模块中创建的（如您的情况）。因此，在实例化指标后，您可以在

App

类中使用以下内容：

self.app.state.registry = registry
self.app.state.counter = counter
self.app.state.latency = latency

在路由器模块中，您可以使用

app

对象获取

Request

实例，如here和here所述，然后使用它来获取指标实例（如下所示），这将让您将指标添加到您的

external_service

:

from fastapi import Request
...
@router.get("/some_router")
async def some_router(request: Request):
    registry = request.app.state.registry
    counter = request.app.state.counter
    latency = request.app.state.latency

    response = await external_service()
    
    return response

选项2

最近，Starlette 推出了

Lifespan

处理程序，它具有

state

的概念，即一个字典，可用于在生命周期和请求之间共享对象。可以使用

state

在端点内访问

request.state

。请查看此答案以获取更多详细信息和示例。