feat(phase-b): metrics history ring buffer + sparkline charts

Add a 24h per-node metrics ring buffer to the aggregator (1440 samples at 1-min resolution via background poller), expose via GET /api/v2/node/:name/metrics-history, and wire up hot-tier usage and cache hit-rate sparklines in NodeCard using Recharts AreaChart/LineChart. Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
2026-07-23 16:36:45 +00:00
parent a3efdbfc04
commit 3fdf46d416
6 changed files with 625 additions and 33 deletions
@@ -13,12 +13,13 @@
 //! Design doc: `docs/dashboard-v2.md`.

 use axum::{
-    extract::{Path, State},
+    extract::{Path, Query, State},
    http::StatusCode,
    routing::{get, post},
    Json, Router,
 };
 use serde::{Deserialize, Serialize};
+use std::collections::{HashMap, VecDeque};
 use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::Duration;
@@ -33,6 +34,57 @@ use crate::cluster::transport::{NodeIdentity, QuicClient};
 use crate::config::{Config, PeerEntry, TokenEntry};
 use crate::sessions::{LeasedTag, Session, SessionStore};

+// ── metrics history ring buffer ──────────────────────────────────
+
+const HISTORY_MAX_SAMPLES: usize = 1440; // 24h at 1-min resolution
+
+/// One time-series sample snapshotted from a peer's DashboardStatus RPC.
+#[derive(Debug, Clone, Serialize)]
+pub struct MetricSample {
+    pub unix_ts: u64,
+    pub hot_used_bytes: u64,
+    pub hot_max_bytes: u64,
+    pub cache_hit_rate: f64,
+    pub cache_hits: u64,
+    pub cache_misses: u64,
+    pub has_chunk_hits: u64,
+    pub has_chunk_misses: u64,
+    pub fs_used_bytes: u64,
+    pub fs_total_bytes: u64,
+    pub fs_available_bytes: u64,
+}
+
+pub struct MetricsHistory {
+    samples: HashMap<String, VecDeque<MetricSample>>,
+}
+
+impl MetricsHistory {
+    fn new() -> Self {
+        Self { samples: HashMap::new() }
+    }
+
+    fn push(&mut self, node: &str, sample: MetricSample) {
+        let deque = self.samples.entry(node.to_string()).or_default();
+        deque.push_back(sample);
+        while deque.len() > HISTORY_MAX_SAMPLES {
+            deque.pop_front();
+        }
+    }
+
+    fn get_last(&self, node: &str, limit: usize) -> Vec<MetricSample> {
+        let limit = limit.min(HISTORY_MAX_SAMPLES);
+        self.samples
+            .get(node)
+            .map(|d| {
+                let skip = d.len().saturating_sub(limit);
+                d.iter().skip(skip).cloned().collect()
+            })
+            .unwrap_or_default()
+    }
+}
+
+// ─────────────────────────────────────────────────────────────────
+
 /// Aggregator runtime: one QuicClient, one peer list, one identity.
 ///
 /// The client is reused across every RPC (quinn holds one UDP
@@ -68,6 +120,9 @@ pub struct V2State {
    /// leases and are reaped by a background sweeper when their
    /// `expires_at_unix` passes without a `renew` or `commit`.
    pub sessions: SessionStore,
+    /// Ring buffer of per-node metric samples (Phase B). 1440 entries
+    /// = 24h at 1-min resolution. Written by `metrics_poller`.
+    pub history: Arc<tokio::sync::Mutex<MetricsHistory>>,
 }

 impl V2State {
@@ -103,6 +158,7 @@ impl V2State {
                .map(|a| a.tokens.clone())
                .unwrap_or_default(),
            sessions,
+            history: Arc::new(tokio::sync::Mutex::new(MetricsHistory::new())),
        })
    }

@@ -514,6 +570,7 @@ impl V2State {
            api_token: self.api_token.clone(),
            token_entries: self.token_entries.clone(),
            sessions: self.sessions.clone(),
+            history: self.history.clone(),
        }
    }
 }
@@ -1289,6 +1346,64 @@ async fn reap_expired(state: Arc<V2State>, sess: Session) {
 ///
 /// Also spawns the background TTL sweeper (Phase 9 S1). The task is
 /// detached — its lifetime is the process lifetime.
+// ── metrics poller (Phase B) ─────────────────────────────────────
+
+/// Background task: poll every peer every 60s, append a `MetricSample`
+/// to the ring buffer. Runs indefinitely — dropped only on daemon exit.
+async fn metrics_poller(state: Arc<V2State>) {
+    let mut interval = tokio::time::interval(Duration::from_secs(60));
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    loop {
+        interval.tick().await;
+        for peer in &state.peers {
+            let peer_name = peer.name.clone();
+            match state.fetch_node(peer).await {
+                Ok(r) => {
+                    let now = std::time::SystemTime::now()
+                        .duration_since(std::time::UNIX_EPOCH)
+                        .unwrap_or_default()
+                        .as_secs();
+                    let sample = MetricSample {
+                        unix_ts: now,
+                        hot_used_bytes: r.hot.as_ref().map(|h| h.used_bytes).unwrap_or(0),
+                        hot_max_bytes: r.hot.as_ref().map(|h| h.max_bytes).unwrap_or(0),
+                        cache_hit_rate: r.cache.as_ref().map(|c| c.hit_rate).unwrap_or(0.0),
+                        cache_hits: r.cache.as_ref().map(|c| c.hits).unwrap_or(0),
+                        cache_misses: r.cache.as_ref().map(|c| c.misses).unwrap_or(0),
+                        has_chunk_hits: r.cache.as_ref().map(|c| c.has_chunk_hits).unwrap_or(0),
+                        has_chunk_misses: r.cache.as_ref().map(|c| c.has_chunk_misses).unwrap_or(0),
+                        fs_used_bytes: r.filesystem.as_ref().map(|f| f.used_bytes).unwrap_or(0),
+                        fs_total_bytes: r.filesystem.as_ref().map(|f| f.total_bytes).unwrap_or(0),
+                        fs_available_bytes: r.filesystem.as_ref().map(|f| f.available_bytes).unwrap_or(0),
+                    };
+                    let mut hist = state.history.lock().await;
+                    hist.push(&peer_name, sample);
+                }
+                Err(e) => {
+                    tracing::debug!(peer = %peer_name, error = %e, "metrics poll skipped");
+                }
+            }
+        }
+    }
+}
+
+#[derive(Deserialize)]
+struct MetricsHistoryQuery {
+    limit: Option<usize>,
+}
+
+async fn handle_metrics_history(
+    Path(name): Path<String>,
+    Query(q): Query<MetricsHistoryQuery>,
+    State(s): State<Arc<V2State>>,
+) -> Json<Vec<MetricSample>> {
+    let limit = q.limit.unwrap_or(60).min(1440);
+    let hist = s.history.lock().await;
+    Json(hist.get_last(&name, limit))
+}
+
+// ─────────────────────────────────────────────────────────────────
+
 pub fn build(state: Arc<V2State>) -> Router {
    // Spawn the sweeper. 15s tick is a reasonable balance: quick
    // enough that a mid-wizard-close cleanup feels prompt, slow
@@ -1305,9 +1420,12 @@ pub fn build(state: Arc<V2State>) -> Router {
            },
        );
    }
+    // Spawn the metrics ring-buffer poller (Phase B).
+    tokio::spawn(metrics_poller(state.clone()));
    Router::new()
        .route("/api/v2/fleet", get(handle_fleet))
        .route("/api/v2/node/:name/status", get(handle_node_status))
+        .route("/api/v2/node/:name/metrics-history", get(handle_metrics_history))
        .route("/api/v2/storage/blobs", get(handle_blobs))
        .route("/api/v2/storage/tags", get(handle_tags))
        .route("/api/v2/storage/refs", get(handle_refs))