我正在尝试在使用
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
添加指标,即该特定外部服务的延迟?
根据 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
最近,Starlette 推出了
Lifespan
处理程序,它具有 state
的概念,即一个字典,可用于在生命周期和请求之间共享对象。可以使用 state
在端点内访问 request.state
。请查看此答案以获取更多详细信息和示例。