45 lines
1.2 KiB
Python
45 lines
1.2 KiB
Python
import time
|
|
|
|
from fastapi import FastAPI, HTTPException, Request, Response
|
|
from prometheus_client import CONTENT_TYPE_LATEST, Counter, Histogram, generate_latest
|
|
|
|
app = FastAPI(title="DevOps Test Service")
|
|
|
|
REQUESTS = Counter(
|
|
"http_requests_total", "Total HTTP requests", ["method", "path", "status"]
|
|
)
|
|
LATENCY = Histogram(
|
|
"http_request_duration_seconds", "Request latency in seconds", ["path"]
|
|
)
|
|
|
|
|
|
@app.middleware("http")
|
|
async def track_metrics(request: Request, call_next):
|
|
start = time.perf_counter()
|
|
response = await call_next(request)
|
|
path = request.url.path
|
|
REQUESTS.labels(request.method, path, response.status_code).inc()
|
|
LATENCY.labels(path).observe(time.perf_counter() - start)
|
|
return response
|
|
|
|
|
|
@app.get("/")
|
|
def index():
|
|
return {"message": "Hello from the DevOps test service"}
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/error")
|
|
def error():
|
|
# Used to generate 5xx responses and trigger the high-error-rate alert
|
|
raise HTTPException(status_code=500, detail="simulated failure")
|
|
|
|
|
|
@app.get("/metrics")
|
|
def metrics():
|
|
return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
|