diff --git a/Cargo.lock b/Cargo.lock index ed9b3c9..deb3c1b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -285,7 +285,7 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" [[package]] name = "cortiq-gateway" -version = "0.2.13" +version = "0.2.14" dependencies = [ "anyhow", "arc-swap", diff --git a/Cargo.toml b/Cargo.toml index 3ab2c49..f947605 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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" diff --git a/src/admin/mod.rs b/src/admin/mod.rs index cddf1ee..955d2fc 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -165,7 +165,7 @@ pub fn api_routes(admin_token: String) -> Router { "/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, Query(q): Query 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) -> 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) -> ApiResult { diff --git a/src/stats.rs b/src/stats.rs index d8dc75a..615f533 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -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, + #[serde(skip)] + pub by_account: HashMap, + #[serde(skip)] + pub by_tier: HashMap, + #[serde(skip)] + pub by_label: HashMap, } #[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, - by_account: HashMap, - by_tier: HashMap, - by_label: HashMap, + /// Per-minute buckets — the single source for the windowed admin snapshot. buckets: VecDeque, recent: VecDeque, } @@ -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 = 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 = 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 = 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 { 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 { diff --git a/web/js/api.js b/web/js/api.js index 2d2d14a..5c9768c 100644 --- a/web/js/api.js +++ b/web/js/api.js @@ -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 diff --git a/web/js/i18n/en.js b/web/js/i18n/en.js index e5b7731..c188d6c 100644 --- a/web/js/i18n/en.js +++ b/web/js/i18n/en.js @@ -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", diff --git a/web/js/i18n/ru.js b/web/js/i18n/ru.js index 08091a7..58464f3 100644 --- a/web/js/i18n/ru.js +++ b/web/js/i18n/ru.js @@ -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ч", diff --git a/web/js/views/dashboard.js b/web/js/views/dashboard.js index f04c04b..76d4a00 100644 --- a/web/js/views/dashboard.js +++ b/web/js/views/dashboard.js @@ -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"))) ) ),