Persist latency/error history + fix Overview GPU strip for remote nodes

src/berth/daemon/admin_runtime.py

@@ -11,6 +11,7 @@
 from berth.daemon.admin import get_backends, get_conn, router
 from berth.store import deployments as dep_store
 from berth.store import nodes as nodes_store
+from berth.store import request_metrics as _request_metrics
 from berth.store import usage_events as _usage_events
 
 
@@ -102,6 +103,35 @@ def usage_series(
     return {**base, "buckets": data}
 
 
+@router.get("/metrics/history")
+def metrics_history(
+    window_s: int = 86400,
+    bucket_s: int = 3600,
+    group_by: str | None = None,
+    summary: bool = False,
+    conn: sqlite3.Connection = Depends(get_conn),
+):
+    """Read-only latency/error history for the Overview dashboard. Per-bucket
+    latency percentiles + error rate (raw rows for short windows, hourly rollup
+    for long). group_by may be 'model' or 'route' (or omitted/'none')."""
+    if window_s <= 0 or bucket_s <= 0:
+        raise HTTPException(400, "window_s and bucket_s must be positive")
+    if window_s // bucket_s > 1024:
+        raise HTTPException(400, "too many buckets requested (cap is 1024)")
+    if group_by not in (None, "none", "model", "route"):
+        raise HTTPException(400, "group_by must be one of: model, route")
+    gb = None if group_by in (None, "none") else group_by
+    if summary:
+        data = _request_metrics.summary(conn, window_s=window_s, group_by=gb)
+        return {"window_s": window_s, "group_by": gb,
+                ("groups" if gb else "summary"): data}
+    data = _request_metrics.history(
+        conn, window_s=window_s, bucket_s=bucket_s, group_by=gb,
+    )
+    return {"window_s": window_s, "bucket_s": bucket_s, "group_by": gb,
+            ("groups" if gb else "buckets"): data}
+
+
 @router.get("/deployments/current/logs")
 def stream_current_logs(request: Request):
     conn: sqlite3.Connection = request.app.state.conn

src/berth/daemon/app.py

@@ -179,7 +179,10 @@ def build_apps(
     event_bus = EventBus()
     stream_tokens = StreamTokenStore()
     from berth.daemon.request_tracer import RequestTracer
-    request_tracer = RequestTracer()
+    from berth.store import request_metrics as _rm_store
+    request_tracer = RequestTracer(
+        on_finalize=lambda trace: _rm_store.record_from_trace(conn, trace),
+    )
     manager = LifecycleManager(
         conn=conn,
         docker_client=docker_client,
@@ -223,6 +226,11 @@ def build_apps(
     # aggregated into usage_aggregates and removed from usage_events.
     # Keeps the predictor's hot table bounded for long-running boxes.
     rollup_task = UsageRollupTask(conn=conn, config=predictor_cfg)
+    # Hourly rollup: request_metrics raw rows older than RAW_RETENTION_H get
+    # aggregated (with exact per-bucket percentiles) into request_metrics_hourly;
+    # hourly rows past retention_days are purged.
+    from berth.lifecycle.metrics_rollup_task import MetricsRollupTask
+    metrics_rollup_task = MetricsRollupTask(conn=conn, config=predictor_cfg)
 
     @asynccontextmanager
     async def lifespan(_app: FastAPI):
@@ -235,6 +243,7 @@ async def lifespan(_app: FastAPI):
         health_monitor.start()
         predictor_task.start()
         rollup_task.start()
+        metrics_rollup_task.start()
         import asyncio as _asyncio
 
         async def _local_metrics_tick() -> None:
@@ -302,6 +311,7 @@ async def _local_metrics_tick() -> None:
         except (Exception, _asyncio.CancelledError):
             pass
         await rollup_task.stop()
+        await metrics_rollup_task.stop()
         await predictor_task.stop()
         await health_monitor.stop()
         await reaper.stop()

src/berth/daemon/request_tracer.py

@@ -12,6 +12,7 @@
 import time
 import uuid
 from collections import deque
+from collections.abc import Callable
 from dataclasses import asdict, dataclass, field
 from typing import Any
 
@@ -63,9 +64,16 @@ class RequestTracer:
     lock needed. The deque's append/popleft are atomic.
     """
 
-    def __init__(self, capacity: int = _MAX_TRACES) -> None:
+    def __init__(
+        self,
+        capacity: int = _MAX_TRACES,
+        on_finalize: Callable[[RequestTrace], None] | None = None,
+    ) -> None:
         self._buffer: deque[RequestTrace] = deque(maxlen=capacity)
         self._subscribers: list[_Subscriber] = []
+        # Optional sink invoked once per finalized request (used to persist
+        # request_metrics). Failure-isolated: a sink error never breaks serving.
+        self._on_finalize = on_finalize
 
     def start(self, *, method: str, path: str) -> RequestTrace:
         trace = RequestTrace(
@@ -88,6 +96,12 @@ def finalize(self, trace: RequestTrace, **fields: Any) -> None:
         for k, v in fields.items():
             setattr(trace, k, v)
         self._publish(trace, "completed")
+        if self._on_finalize is not None:
+            try:
+                self._on_finalize(trace)
+            except Exception:
+                import logging
+                logging.getLogger(__name__).exception("tracer on_finalize failed")
 
     def snapshot(self) -> list[dict[str, Any]]:
         return [t.to_dict() for t in self._buffer]

src/berth/lifecycle/metrics_rollup_task.py

@@ -0,0 +1,59 @@
+"""Daemon task that rolls aged request_metrics into request_metrics_hourly and
+purges old hourly rows. Mirrors UsageRollupTask."""
+from __future__ import annotations
+
+import asyncio
+import logging
+import sqlite3
+
+from berth.lifecycle.predictor import PredictorConfig
+from berth.store import request_metrics_rollup as rmr
+from berth.store.request_metrics import RAW_RETENTION_H
+
+log = logging.getLogger(__name__)
+
+
+class MetricsRollupTask:
+    def __init__(
+        self,
+        *,
+        conn: sqlite3.Connection,
+        config: PredictorConfig | None = None,
+        tick_s: float = 3600.0,
+    ):
+        self._conn = conn
+        self._config = config or PredictorConfig()
+        self._tick_s = tick_s
+        self._task: asyncio.Task | None = None
+        self._stop_event = asyncio.Event()
+
+    async def tick_once(self) -> dict:
+        res = rmr.rollup_aged_raw(self._conn, older_than_h=RAW_RETENTION_H)
+        res["hourly_purged"] = rmr.purge_hourly_older_than(
+            self._conn, days=self._config.retention_days,
+        )
+        return res
+
+    async def run(self) -> None:
+        while not self._stop_event.is_set():
+            try:
+                r = await self.tick_once()
+                if r["raw_deleted"] or r["hourly_purged"]:
+                    log.info(
+                        "metrics rollup: %d raw rolled, %d hourly purged",
+                        r["raw_deleted"], r["hourly_purged"],
+                    )
+            except Exception:
+                log.exception("metrics rollup tick failed")
+            try:
+                await asyncio.wait_for(self._stop_event.wait(), timeout=self._tick_s)
+            except TimeoutError:
+                pass
+
+    def start(self) -> None:
+        self._task = asyncio.create_task(self.run())
+
+    async def stop(self) -> None:
+        self._stop_event.set()
+        if self._task:
+            await self._task

src/berth/store/migrations/017_request_metrics.sql

@@ -0,0 +1,38 @@
+-- Per-request outcome + latency for the Overview dashboard's historical
+-- latency/error views. Distinct from usage_events (the predictor's
+-- served-request log): this captures ALL finalized requests, including
+-- pre-dispatch failures (bad model, auth, no-ready-service, adapter errors).
+
+CREATE TABLE IF NOT EXISTS request_metrics (
+    id              INTEGER PRIMARY KEY,
+    ts              TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    model_name      TEXT,
+    route_name      TEXT,
+    deployment_id   INTEGER,
+    backend         TEXT,
+    status_code     INTEGER,
+    is_error        INTEGER NOT NULL DEFAULT 0,
+    dispatched      INTEGER NOT NULL DEFAULT 0,
+    latency_ms      INTEGER,
+    ttft_ms         INTEGER
+);
+CREATE INDEX IF NOT EXISTS idx_rm_ts ON request_metrics(ts);
+CREATE INDEX IF NOT EXISTS idx_rm_model_ts ON request_metrics(model_name, ts);
+
+-- Hourly rollup with exact per-bucket percentiles (computed from raw rows at
+-- rollup time). model_name NULL = the all-models aggregate row for that hour.
+CREATE TABLE IF NOT EXISTS request_metrics_hourly (
+    id                INTEGER PRIMARY KEY,
+    bucket_start      TIMESTAMP NOT NULL,
+    model_name        TEXT,
+    count             INTEGER NOT NULL,
+    error_count       INTEGER NOT NULL,
+    dispatched_count  INTEGER NOT NULL,
+    latency_p50_ms    INTEGER,
+    latency_p95_ms    INTEGER,
+    ttft_p50_ms       INTEGER,
+    ttft_p95_ms       INTEGER
+);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_rmh_bucket_model
+    ON request_metrics_hourly(bucket_start, COALESCE(model_name, ''));
+CREATE INDEX IF NOT EXISTS idx_rmh_bucket ON request_metrics_hourly(bucket_start);