feat: metriche gateway con push a VictoriaMetrics + dashboard Grafana
- raccolta in-memory: richieste per endpoint, latenza (sum/count), errori per status, search queries/hits, punti - push periodico (30s) a VM via /api/v1/import/prometheus (pattern energy engine, timestamp ms) - VM_PUSH_URL/VM_PUSH_INTERVAL/METRICS_ENABLED configurabili; default host.docker.internal:8428 - endpoint /v1/metrics (auth) per verifica manuale - dashboard Grafana versionata in monitoring/qmem-dashboard.json (provisioning: /home/enne2/domotics/grafana/dashboards/) - verificato sul server: /v1/metrics OK, 6 serie qmem_* in VM
This commit is contained in:
+108
@@ -52,6 +52,20 @@ API_KEYS: set[str] = {k.strip() for k in os.environ.get("API_KEYS", "").split(",
|
||||
RATE_LIMIT_PER_MIN = int(os.environ.get("RATE_LIMIT_PER_MIN", "120"))
|
||||
MAX_TEXT_LEN = int(os.environ.get("MAX_TEXT_LEN", "8000"))
|
||||
|
||||
# Metriche: push a VictoriaMetrics (stesso pattern dell'energy engine domotics)
|
||||
VM_PUSH_URL = os.environ.get("VM_PUSH_URL", "http://host.docker.internal:8428/api/v1/import/prometheus")
|
||||
VM_PUSH_INTERVAL = int(os.environ.get("VM_PUSH_INTERVAL", "30"))
|
||||
METRICS_ENABLED = os.environ.get("METRICS_ENABLED", "true").lower() == "true"
|
||||
|
||||
_metrics: dict[str, Any] = {
|
||||
"requests": Counter(),
|
||||
"duration_sum": Counter(),
|
||||
"duration_count": Counter(),
|
||||
"errors": Counter(),
|
||||
"search_queries": 0,
|
||||
"search_hits": 0,
|
||||
}
|
||||
|
||||
# Hybrid retrieval: sparse vector BM25 (Qdrant/bm25 via fastembed, modifier IDF)
|
||||
SPARSE_VECTOR_NAME = "bm25"
|
||||
try:
|
||||
@@ -107,6 +121,7 @@ async def lifespan(_app: FastAPI):
|
||||
_backfill_sparse()
|
||||
|
||||
cleanup_task = asyncio.create_task(_cleanup_loop())
|
||||
metrics_task = asyncio.create_task(_metrics_push_loop()) if METRICS_ENABLED else None
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
@@ -115,6 +130,12 @@ async def lifespan(_app: FastAPI):
|
||||
await cleanup_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
if metrics_task is not None:
|
||||
metrics_task.cancel()
|
||||
try:
|
||||
await metrics_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
global _http
|
||||
if _http is not None:
|
||||
await _http.aclose()
|
||||
@@ -136,6 +157,22 @@ async def request_id_middleware(request: Request, call_next):
|
||||
response.headers["X-Request-ID"] = rid
|
||||
return response
|
||||
|
||||
|
||||
@app.middleware("http")
|
||||
async def metrics_middleware(request: Request, call_next):
|
||||
"""Raccoglie conteggi, latenza ed errori per endpoint."""
|
||||
start = time.monotonic()
|
||||
response = await call_next(request)
|
||||
dur = time.monotonic() - start
|
||||
route = request.scope.get("route")
|
||||
endpoint = route.path if route else request.url.path
|
||||
_metrics["requests"][endpoint] += 1
|
||||
_metrics["duration_sum"][endpoint] += dur
|
||||
_metrics["duration_count"][endpoint] += 1
|
||||
if response.status_code >= 400:
|
||||
_metrics["errors"][(endpoint, response.status_code)] += 1
|
||||
return response
|
||||
|
||||
# Rate limit in-memory: {key: [timestamps]}
|
||||
_ratelimit: dict[str, list[float]] = {}
|
||||
|
||||
@@ -482,6 +519,8 @@ async def search_memories(body: SearchIn, key: str = Depends(require_auth)) -> d
|
||||
min_score=body.min_score,
|
||||
hits=len(results),
|
||||
)
|
||||
_metrics["search_queries"] += 1
|
||||
_metrics["search_hits"] += len(results)
|
||||
return {"results": results, "min_score": body.min_score, "total_hits": len(results)}
|
||||
|
||||
|
||||
@@ -586,6 +625,75 @@ async def status(request: Request) -> dict:
|
||||
}
|
||||
|
||||
|
||||
@app.get("/v1/metrics")
|
||||
async def metrics(key: str = Depends(require_auth)) -> dict:
|
||||
"""Riepilogo metriche in-memory (per verifica manuale; il push a VM è automatico)."""
|
||||
try:
|
||||
info = qdrant.get_collection(COLLECTION)
|
||||
points = info.points_count
|
||||
except Exception: # noqa: BLE001
|
||||
points = None
|
||||
return {
|
||||
"requests": dict(_metrics["requests"]),
|
||||
"avg_duration_ms": {
|
||||
e: round(_metrics["duration_sum"][e] / _metrics["duration_count"][e] * 1000, 2)
|
||||
for e in _metrics["duration_count"]
|
||||
},
|
||||
"errors": {f"{e}:{s}": c for (e, s), c in _metrics["errors"].items()},
|
||||
"search_queries": _metrics["search_queries"],
|
||||
"search_hits": _metrics["search_hits"],
|
||||
"points": points,
|
||||
}
|
||||
|
||||
|
||||
def _prometheus_lines() -> list[str]:
|
||||
"""Metriche in formato Prometheus text (senza timestamp, aggiunto dal push)."""
|
||||
lines: list[str] = []
|
||||
for endpoint, count in _metrics["requests"].items():
|
||||
lines.append(f'qmem_requests_total{{endpoint="{endpoint}"}} {count}')
|
||||
for endpoint, s in _metrics["duration_sum"].items():
|
||||
c = _metrics["duration_count"][endpoint]
|
||||
lines.append(f'qmem_request_duration_seconds_sum{{endpoint="{endpoint}"}} {s:.6f}')
|
||||
lines.append(f'qmem_request_duration_seconds_count{{endpoint="{endpoint}"}} {c}')
|
||||
for (endpoint, status), count in _metrics["errors"].items():
|
||||
lines.append(f'qmem_errors_total{{endpoint="{endpoint}",status="{status}"}} {count}')
|
||||
lines.append(f"qmem_search_queries_total {_metrics['search_queries']}")
|
||||
lines.append(f"qmem_search_hits_total {_metrics['search_hits']}")
|
||||
try:
|
||||
info = qdrant.get_collection(COLLECTION)
|
||||
lines.append(f"qmem_points {info.points_count}")
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
return lines
|
||||
|
||||
|
||||
async def _metrics_push_loop() -> None:
|
||||
"""Push periodico delle metriche a VictoriaMetrics (formato Prometheus + timestamp ms)."""
|
||||
while True:
|
||||
try:
|
||||
now_ms = int(time.time() * 1000)
|
||||
body = "\n".join(f"{l} {now_ms}" for l in _prometheus_lines()) + "\n"
|
||||
r = await _get_http().post(
|
||||
VM_PUSH_URL,
|
||||
content=body,
|
||||
headers={"Content-Type": "text/plain"},
|
||||
)
|
||||
if r.status_code >= 300:
|
||||
log.warning("metrics push: HTTP %s", r.status_code)
|
||||
except Exception as e: # noqa: BLE001
|
||||
log.warning("metrics push error: %s", e)
|
||||
await asyncio.sleep(VM_PUSH_INTERVAL)
|
||||
return {
|
||||
"status": "ok",
|
||||
"collection": COLLECTION,
|
||||
"points": info.points_count,
|
||||
"embedding_model": EMBED_MODEL,
|
||||
"embedding_dim": EMBED_DIM,
|
||||
"access": "shared",
|
||||
"api_keys": len(API_KEYS),
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Cleanup periodico: rimuove record scaduti (expires_at < now)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user