diff --git a/gateway/main.py b/gateway/main.py index c67c80b..1a71263 100644 --- a/gateway/main.py +++ b/gateway/main.py @@ -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) # --------------------------------------------------------------------------- diff --git a/monitoring/qmem-dashboard.json b/monitoring/qmem-dashboard.json new file mode 100644 index 0000000..4d8154b --- /dev/null +++ b/monitoring/qmem-dashboard.json @@ -0,0 +1,153 @@ +{ + "title": "Qmem Memory Gateway", + "uid": "qmem-gateway", + "tags": ["qmem", "memory", "ai"], + "timezone": "browser", + "schemaVersion": 39, + "version": 1, + "editable": true, + "refresh": "30s", + "time": { "from": "now-6h", "to": "now" }, + "templating": { "list": [] }, + "panels": [ + { + "type": "stat", + "title": "Punti in memoria", + "gridPos": { "h": 4, "w": 4, "x": 0, "y": 0 }, + "datasource": { "type": "prometheus", "uid": "victoriametrics" }, + "targets": [{ "expr": "qmem_points", "refId": "A" }], + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "thresholds": { "mode": "absolute", "steps": [ { "color": "green", "value": null } ] } + }, + "overrides": [] + }, + "options": { "colorMode": "value", "graphMode": "area", "reduceOptions": { "calcs": ["lastNotNull"] } } + }, + { + "type": "stat", + "title": "Query di ricerca (ultima ora)", + "gridPos": { "h": 4, "w": 4, "x": 4, "y": 0 }, + "datasource": { "type": "prometheus", "uid": "victoriametrics" }, + "targets": [{ "expr": "sum(increase(qmem_search_queries_total[1h]))", "refId": "A" }], + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "thresholds": { "mode": "absolute", "steps": [ { "color": "blue", "value": null } ] } + }, + "overrides": [] + }, + "options": { "colorMode": "value", "graphMode": "area", "reduceOptions": { "calcs": ["lastNotNull"] } } + }, + { + "type": "stat", + "title": "Risultati restituiti (ultima ora)", + "gridPos": { "h": 4, "w": 4, "x": 8, "y": 0 }, + "datasource": { "type": "prometheus", "uid": "victoriametrics" }, + "targets": [{ "expr": "sum(increase(qmem_search_hits_total[1h]))", "refId": "A" }], + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "thresholds": { "mode": "absolute", "steps": [ { "color": "purple", "value": null } ] } + }, + "overrides": [] + }, + "options": { "colorMode": "value", "graphMode": "area", "reduceOptions": { "calcs": ["lastNotNull"] } } + }, + { + "type": "stat", + "title": "Errori (ultima ora)", + "gridPos": { "h": 4, "w": 4, "x": 12, "y": 0 }, + "datasource": { "type": "prometheus", "uid": "victoriametrics" }, + "targets": [{ "expr": "sum(increase(qmem_errors_total[1h]))", "refId": "A" }], + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "thresholds": { "mode": "absolute", "steps": [ { "color": "green", "value": null }, { "color": "orange", "value": 1 }, { "color": "red", "value": 5 } ] } + }, + "overrides": [] + }, + "options": { "colorMode": "value", "graphMode": "area", "reduceOptions": { "calcs": ["lastNotNull"] } } + }, + { + "type": "bargauge", + "title": "Richieste per endpoint", + "gridPos": { "h": 6, "w": 8, "x": 0, "y": 4 }, + "datasource": { "type": "prometheus", "uid": "victoriametrics" }, + "targets": [{ "expr": "sum(qmem_requests_total) by (endpoint)", "refId": "A" }], + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "thresholds": { "mode": "absolute", "steps": [ { "color": "green", "value": null } ] } + }, + "overrides": [] + }, + "options": { "orientation": "horizontal", "displayMode": "gradient", "reduceOptions": { "calcs": ["lastNotNull"] } } + }, + { + "type": "bargauge", + "title": "Latenza media per endpoint (ms)", + "gridPos": { "h": 6, "w": 8, "x": 8, "y": 4 }, + "datasource": { "type": "prometheus", "uid": "victoriametrics" }, + "targets": [ + { + "expr": "sum(qmem_request_duration_seconds_sum) by (endpoint) / sum(qmem_request_duration_seconds_count) by (endpoint) * 1000", + "refId": "A" + } + ], + "fieldConfig": { + "defaults": { + "unit": "ms", + "color": { "mode": "palette-classic" }, + "thresholds": { "mode": "absolute", "steps": [ { "color": "green", "value": null } ] } + }, + "overrides": [] + }, + "options": { "orientation": "horizontal", "displayMode": "gradient", "reduceOptions": { "calcs": ["lastNotNull"] } } + }, + { + "type": "timeseries", + "title": "Latenza media search (ms)", + "gridPos": { "h": 6, "w": 8, "x": 16, "y": 4 }, + "datasource": { "type": "prometheus", "uid": "victoriametrics" }, + "targets": [ + { + "expr": "sum(rate(qmem_request_duration_seconds_sum{endpoint=\"/v1/memories:search\"}[5m])) / sum(rate(qmem_request_duration_seconds_count{endpoint=\"/v1/memories:search\"}[5m])) * 1000", + "refId": "A" + } + ], + "fieldConfig": { + "defaults": { "unit": "ms", "color": { "mode": "palette-classic" } }, + "overrides": [] + }, + "options": { "legend": { "displayMode": "list", "placement": "bottom" }, "tooltip": { "mode": "multi" } } + }, + { + "type": "timeseries", + "title": "Richieste al minuto", + "gridPos": { "h": 6, "w": 12, "x": 0, "y": 10 }, + "datasource": { "type": "prometheus", "uid": "victoriametrics" }, + "targets": [ + { "expr": "sum(rate(qmem_requests_total[5m])) by (endpoint)", "refId": "A" } + ], + "fieldConfig": { + "defaults": { "color": { "mode": "palette-classic" } }, + "overrides": [] + }, + "options": { "legend": { "displayMode": "table", "placement": "bottom" }, "tooltip": { "mode": "multi" } } + }, + { + "type": "timeseries", + "title": "Punti in memoria nel tempo", + "gridPos": { "h": 6, "w": 12, "x": 12, "y": 10 }, + "datasource": { "type": "prometheus", "uid": "victoriametrics" }, + "targets": [{ "expr": "qmem_points", "refId": "A" }], + "fieldConfig": { + "defaults": { "color": { "mode": "palette-classic" } }, + "overrides": [] + }, + "options": { "legend": { "displayMode": "list", "placement": "bottom" }, "tooltip": { "mode": "multi" } } + } + ] +}