1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
|
import asyncio
from typing import Any, Dict, Optional
from fastapi import FastAPI, HTTPException
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Route
from starlette.testclient import TestClient
from prometheus_fastapi_instrumentator import Instrumentator
# ------------------------------------------------------------------------------
# Setup
def create_fastapi_app() -> FastAPI:
app = FastAPI()
@app.get("/")
def read_root():
return "Hello World!"
@app.get("/sleep")
async def sleep(seconds: float):
await asyncio.sleep(seconds)
return f"I have slept for {seconds}s"
@app.get("/always_error")
def read_always_error():
raise HTTPException(status_code=404, detail="Not really error")
@app.get("/ignore")
def read_ignore():
return "Should be ignored"
@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None):
return {"item_id": item_id, "q": q}
@app.get("/just_another_endpoint")
def read_just_another_endpoint():
return "Green is my pepper"
@app.post("/items")
def create_item(item: Dict[Any, Any]):
return None
return app
def create_starlette_app() -> Starlette:
async def homepage(request):
return PlainTextResponse("Homepage")
return Starlette(routes=[Route("/", endpoint=homepage)])
def reset_prometheus() -> None:
from prometheus_client import REGISTRY
# Unregister all collectors.
collectors = list(REGISTRY._collector_to_names.keys())
print(f"before unregister collectors={collectors}")
for collector in collectors:
REGISTRY.unregister(collector)
print(f"after unregister collectors={list(REGISTRY._collector_to_names.keys())}")
# Import default collectors.
from prometheus_client import gc_collector, platform_collector, process_collector
# Re-register default collectors.
process_collector.ProcessCollector()
platform_collector.PlatformCollector()
gc_collector.GCCollector()
# ------------------------------------------------------------------------------
# Tests
def test_expose_default_content_type():
reset_prometheus()
app = create_fastapi_app()
Instrumentator().instrument(app).expose(app)
client = TestClient(app)
response = client.get("/metrics")
print(response.headers.items())
assert (
"text/plain; version=0.0.4; charset=utf-8; charset=utf-8"
not in response.headers.values()
)
def test_fastapi_app_expose():
reset_prometheus()
app = create_fastapi_app()
Instrumentator().instrument(app).expose(app)
client = TestClient(app)
response = client.get("/metrics")
assert response.status_code == 200
def test_starlette_app_expose():
reset_prometheus()
app = create_starlette_app()
Instrumentator().instrument(app).expose(app)
client = TestClient(app)
response = client.get("/metrics")
assert response.status_code == 200
|
