feat(dashboard): windowed stats per range + clear-logs button

- Stats snapshot now computes the KPI totals AND the breakdown from the in-window per-minute buckets (previously they used all-time accumulators, so switching 1h/24h/7d changed only the chart, not the numbers). Buckets now carry failovers + per-group breakdown; Prometheus keeps all-time counters. - New DELETE /admin/api/stats clears all in-memory stats and truncates the JSONL log; the dashboard gets a "Clear logs" button (with confirm).
2026-07-09 17:07:05 +03:00
parent 0d1283c3fa
commit b55e72789f
8 changed files with 129 additions and 49 deletions
@@ -285,7 +285,7 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"

 [[package]]
 name = "cortiq-gateway"
-version = "0.2.13"
+version = "0.2.14"
 dependencies = [
 "anyhow",
 "arc-swap",
@@ -1,6 +1,6 @@
 [package]
 name = "cortiq-gateway"
-version = "0.2.13"
+version = "0.2.14"
 edition = "2021"
 rust-version = "1.75"
 license = "Apache-2.0"
@@ -165,7 +165,7 @@ pub fn api_routes(admin_token: String) -> Router<SharedState> {
            "/admin/api/secrets",
            get(list_secrets).put(set_secret).delete(clear_secret),
        )
-        .route("/admin/api/stats", get(get_stats))
+        .route("/admin/api/stats", get(get_stats).delete(clear_stats))
        .route("/admin/api/shadow", get(get_shadow))
        .route("/admin/api/hf/search", get(hf_search))
        .route("/admin/api/import", get(list_imports).post(start_import))
@@ -1197,6 +1197,13 @@ async fn get_stats(State(state): State<SharedState>, Query(q): Query<StatsQuery>
    ok(snap)
 }

+/// Clear all recorded stats/logs (in-memory aggregates, series, recent requests)
+/// and truncate the JSONL log — the dashboard "Clear logs" button.
+async fn clear_stats(State(state): State<SharedState>) -> ApiResult {
+    state.stats.clear();
+    ok(json!({ "ok": true }))
+}
+
 /// Self-warming shadow loop: per-task-type promotion state, Wilson-LB of the
 /// judged pass-rate, and whether the local model is serving it yet.
 async fn get_shadow(State(state): State<SharedState>) -> ApiResult {
@@ -83,9 +83,23 @@ impl Totals {
        self.cost_usd += r.cost_usd;
        self.latency_ms_sum += r.latency_ms;
    }
+    /// Accumulate another Totals into this one (for windowed aggregation).
+    fn merge(&mut self, o: &Totals) {
+        self.requests += o.requests;
+        self.ok += o.ok;
+        self.errors += o.errors;
+        self.failovers += o.failovers;
+        self.prompt_tokens += o.prompt_tokens;
+        self.completion_tokens += o.completion_tokens;
+        self.cost_usd += o.cost_usd;
+        self.latency_ms_sum += o.latency_ms_sum;
+    }
 }

-/// Per-minute bucket for time-series charts.
+/// Per-minute bucket. Beyond the chart fields it carries the per-group breakdown
+/// and failovers so the admin snapshot can be computed for ANY time window by
+/// summing the in-window buckets (the internal fields are not serialized into
+/// the `series` payload).
 #[derive(Clone, Debug, Default, Serialize)]
 pub struct Bucket {
    pub minute: u64, // unix seconds, rounded down to the minute
@@ -95,15 +109,25 @@ pub struct Bucket {
    pub completion_tokens: u64,
    pub cost_usd: f64,
    pub latency_ms_sum: u64,
+    #[serde(skip)]
+    pub failovers: u64,
+    #[serde(skip)]
+    pub by_model: HashMap<String, Totals>,
+    #[serde(skip)]
+    pub by_account: HashMap<String, Totals>,
+    #[serde(skip)]
+    pub by_tier: HashMap<String, Totals>,
+    #[serde(skip)]
+    pub by_label: HashMap<String, Totals>,
 }

 #[derive(Default)]
 struct Inner {
+    /// All-time totals + per-model (used by the Prometheus counters, which must
+    /// be monotonic — the windowed admin snapshot is derived from the buckets).
    total: Totals,
    by_model: HashMap<String, Totals>,
-    by_account: HashMap<String, Totals>,
-    by_tier: HashMap<String, Totals>,
-    by_label: HashMap<String, Totals>,
+    /// Per-minute buckets — the single source for the windowed admin snapshot.
    buckets: VecDeque<Bucket>,
    recent: VecDeque<RequestRecord>,
 }
@@ -166,31 +190,14 @@ impl Stats {
    }

    fn apply_aggregates(inner: &mut Inner, rec: &RequestRecord, retention_secs: u64) {
+        // all-time counters for Prometheus; the windowed snapshot is derived
+        // from the buckets below.
        inner.total.apply(rec);
        inner
            .by_model
            .entry(rec.model_id.clone())
            .or_default()
            .apply(rec);
-        inner
-            .by_account
-            .entry(if rec.account.is_empty() {
-                "anonymous".to_string()
-            } else {
-                rec.account.clone()
-            })
-            .or_default()
-            .apply(rec);
-        inner
-            .by_tier
-            .entry(rec.tier.clone())
-            .or_default()
-            .apply(rec);
-        inner
-            .by_label
-            .entry(rec.task_label.clone())
-            .or_default()
-            .apply(rec);

        let minute = rec.ts - (rec.ts % 60);
        match inner.buckets.back_mut() {
@@ -250,13 +257,31 @@ impl Stats {
            .filter(|b| b.minute >= cutoff)
            .collect();

-        let breakdown_src = match groupby {
-            "account" => &inner.by_account,
-            "tier" => &inner.by_tier,
-            "label" => &inner.by_label,
-            _ => &inner.by_model,
-        };
-        let mut breakdown: Vec<serde_json::Value> = breakdown_src
+        // Windowed totals + breakdown = sum of the in-window buckets, so the KPIs
+        // and the breakdown recompute whenever the range changes.
+        let mut win = Totals::default();
+        let mut group: HashMap<String, Totals> = HashMap::new();
+        for b in &series {
+            win.requests += b.requests;
+            win.errors += b.errors;
+            win.ok += b.requests.saturating_sub(b.errors);
+            win.failovers += b.failovers;
+            win.prompt_tokens += b.prompt_tokens;
+            win.completion_tokens += b.completion_tokens;
+            win.cost_usd += b.cost_usd;
+            win.latency_ms_sum += b.latency_ms_sum;
+            let src = match groupby {
+                "account" => &b.by_account,
+                "tier" => &b.by_tier,
+                "label" => &b.by_label,
+                _ => &b.by_model,
+            };
+            for (k, t) in src {
+                group.entry(k.clone()).or_default().merge(t);
+            }
+        }
+
+        let mut breakdown: Vec<serde_json::Value> = group
            .iter()
            .map(|(k, t)| {
                serde_json::json!({
@@ -280,17 +305,17 @@ impl Stats {

        serde_json::json!({
            "totals": {
-                "requests": inner.total.requests,
-                "ok": inner.total.ok,
-                "errors": inner.total.errors,
-                "failovers": inner.total.failovers,
-                "prompt_tokens": inner.total.prompt_tokens,
-                "completion_tokens": inner.total.completion_tokens,
-                "total_tokens": inner.total.prompt_tokens + inner.total.completion_tokens,
-                "cost_usd": inner.total.cost_usd,
-                "avg_latency_ms": avg(inner.total.latency_ms_sum, inner.total.requests),
-                "success_rate": if inner.total.requests > 0 {
-                    inner.total.ok as f64 / inner.total.requests as f64
+                "requests": win.requests,
+                "ok": win.ok,
+                "errors": win.errors,
+                "failovers": win.failovers,
+                "prompt_tokens": win.prompt_tokens,
+                "completion_tokens": win.completion_tokens,
+                "total_tokens": win.prompt_tokens + win.completion_tokens,
+                "cost_usd": win.cost_usd,
+                "avg_latency_ms": avg(win.latency_ms_sum, win.requests),
+                "success_rate": if win.requests > 0 {
+                    win.ok as f64 / win.requests as f64
                } else { 0.0 },
            },
            "groupby": groupby,
@@ -299,6 +324,18 @@ impl Stats {
        })
    }

+    /// Reset all in-memory stats and truncate the JSONL log — the "clear logs"
+    /// admin action.
+    pub fn clear(&self) {
+        {
+            let mut inner = self.inner.lock().unwrap();
+            *inner = Inner::default();
+        }
+        if let Some(path) = &self.file {
+            let _ = std::fs::write(path, "");
+        }
+    }
+
    /// Recent requests (newest first), with pagination.
    pub fn recent(&self, limit: usize, offset: usize) -> Vec<RequestRecord> {
        let inner = self.inner.lock().unwrap();
@@ -358,10 +395,25 @@ fn fill_bucket(b: &mut Bucket, r: &RequestRecord) {
    if r.outcome != "ok" {
        b.errors += 1;
    }
+    if r.failover {
+        b.failovers += 1;
+    }
    b.prompt_tokens += r.prompt_tokens as u64;
    b.completion_tokens += r.completion_tokens as u64;
    b.cost_usd += r.cost_usd;
    b.latency_ms_sum += r.latency_ms;
+    // per-group breakdown, so the snapshot can be windowed by summing buckets
+    b.by_model.entry(r.model_id.clone()).or_default().apply(r);
+    b.by_account
+        .entry(if r.account.is_empty() {
+            "anonymous".to_string()
+        } else {
+            r.account.clone()
+        })
+        .or_default()
+        .apply(r);
+    b.by_tier.entry(r.tier.clone()).or_default().apply(r);
+    b.by_label.entry(r.task_label.clone()).or_default().apply(r);
 }

 fn avg(sum: u64, n: u64) -> u64 {
@@ -87,6 +87,7 @@ export const api = {

  // stats
  stats: (range, groupby) => j("GET", `/stats?range=${range}&groupby=${groupby}`),
+  clearStats: () => j("DELETE", "/stats"),
  requests: (limit = 50, offset = 0) => j("GET", `/requests?limit=${limit}&offset=${offset}`),

  // playground
@@ -59,6 +59,9 @@ export default {
  // dashboard
  "dash.title": "Dashboard",
  "dash.subtitle": "Live traffic, cost and routing analytics for your model pool.",
+  "dash.clear": "Clear logs",
+  "dash.clearConfirm": "Clear all recorded stats and logs? This cannot be undone.",
+  "dash.cleared": "Logs cleared",
  "dash.range": "Range",
  "dash.range.1h": "1h",
  "dash.range.24h": "24h",
@@ -58,6 +58,9 @@ export default {
  // dashboard
  "dash.title": "Дашборд",
  "dash.subtitle": "Трафик в реальном времени, расходы и аналитика роутинга по пулу моделей.",
+  "dash.clear": "Очистить логи",
+  "dash.clearConfirm": "Очистить всю статистику и логи? Отменить нельзя.",
+  "dash.cleared": "Логи очищены",
  "dash.range": "Период",
  "dash.range.1h": "1ч",
  "dash.range.24h": "24ч",
@@ -1,5 +1,5 @@
 // Dashboard — KPIs, time-series charts, breakdown, recent requests, health.
-import { h, mount, money, num, ms, pct, timeAgo } from "../ui.js";
+import { h, mount, money, num, ms, pct, timeAgo, toast, confirmDialog } from "../ui.js";
 import { t } from "../i18n.js";
 import { api } from "../api.js";
 import { lineChart, barList } from "../charts.js";
@@ -53,10 +53,24 @@ export async function renderDashboard() {
          "div",
          { class: "flex" },
          h("div", { class: "grow" }, h("h2", {}, t("dash.title")), h("p", {}, t("dash.subtitle"))),
-          seg(RANGES, range, (r) => {
-            range = r;
-            load();
-          }, (r) => t("dash.range." + r))
+          h("div", { class: "flex", style: "gap:8px;align-items:center" },
+            seg(RANGES, range, (r) => {
+              range = r;
+              load();
+            }, (r) => t("dash.range." + r)),
+            h("button", {
+              class: "btn sm danger",
+              onClick: async () => {
+                if (!(await confirmDialog(t("dash.clearConfirm")))) return;
+                try {
+                  await api.clearStats();
+                  toast(t("dash.cleared"), "good");
+                  await load();
+                } catch (e) {
+                  toast(String(e.message || e), "bad");
+                }
+              },
+            }, t("dash.clear")))
        )
      ),