feat(phase-c): drift-adaptive anomaly detection + smarter eviction scoring

serve_v2: Welford online stats per node (hot_pct, hit_rate, fs_pct,
chunk_miss_rate) updated on every poller tick. Anomaly score = sum of z²;
warn at 9, alert at 16. Exposed via GET /api/v2/anomalies. Score embedded
in MetricSample so sparklines can also surface it.

hot: Replace pure-LRU gc_by_space with size×age eviction scoring.
Candidates are ranked by ln(size) × ln(age_secs); pinned projects immune.
Evicts the most space with the least cost rather than just the oldest entry.

dashboard: Fleet anomaly banner in CommandCenter (red/amber) + per-node
"anomaly"/"drift" badge on NodeCard. Anomalies polled every 10s alongside
the fleet endpoint.

Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
This commit is contained in:
Omar Sobh
2026-07-23 17:10:09 +00:00
co-authored by Claude Sonnet 4.6
parent 3fdf46d416
commit 70ad863006
5 changed files with 228 additions and 16 deletions
+27 -10
View File
@@ -55,25 +55,42 @@ pub fn gc_stale_targets(manifest: &Manifest, stale_hours: u64) -> Result<Vec<Str
Ok(evicted) Ok(evicted)
} }
/// Eviction priority score: larger and older = higher = evict first.
/// log-scaled so a 10GB project doesn't completely dominate a 1GB project
/// and a week-old entry doesn't mask a 3-day-old one of similar value.
fn eviction_score(size_bytes: u64, secs_since_active: u64) -> f64 {
let size_factor = (size_bytes as f64 + 1.0).ln();
let age_factor = (secs_since_active as f64 + 1.0).ln();
size_factor * age_factor
}
pub fn gc_by_space(manifest: &mut Manifest, max_gb: f64) -> Result<Vec<String>> { pub fn gc_by_space(manifest: &mut Manifest, max_gb: f64) -> Result<Vec<String>> {
let now = Utc::now();
let mut evicted = Vec::new(); let mut evicted = Vec::new();
loop { loop {
let used = total_used_gb(manifest)?; let used = total_used_gb(manifest)?;
if used <= max_gb { if used <= max_gb {
break; break;
} }
// LRU eviction skips pinned projects — they may NEVER be evicted // Score-based eviction: prefer removing large, stale projects over
// for space pressure. The trade-off: if every non-pinned project // arbitrarily picking the oldest-active one. Pinned projects are
// is gone and we're still over `max_gb`, we stop and log; better // never evicted regardless of score — operator intent is absolute.
// to over-allocate hot than to violate operator intent. let best = manifest
let lru_name = manifest
.projects .projects
.iter() .iter()
.filter(|p| !p.pinned) .filter(|p| !p.pinned && p.hot_target_path.exists())
.filter(|p| p.hot_target_path.exists()) .map(|p| {
.min_by_key(|p| p.last_active) let size = target_size_bytes(&p.hot_target_path).unwrap_or(0);
.map(|p| p.name.clone()); let age_secs = p
match lru_name { .last_active
.map(|t| (now - t).num_seconds().max(0) as u64)
.unwrap_or(0);
(p.name.clone(), eviction_score(size, age_secs))
})
.max_by(|(_, a), (_, b)| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal))
.map(|(name, _)| name);
match best {
None => { None => {
tracing::warn!( tracing::warn!(
used_gb = used, used_gb = used,
+139 -3
View File
@@ -34,9 +34,99 @@ use crate::cluster::transport::{NodeIdentity, QuicClient};
use crate::config::{Config, PeerEntry, TokenEntry}; use crate::config::{Config, PeerEntry, TokenEntry};
use crate::sessions::{LeasedTag, Session, SessionStore}; use crate::sessions::{LeasedTag, Session, SessionStore};
// ── metrics history ring buffer ────────────────────────────────── // ── metrics history ring buffer + drift-adaptive anomaly detection ──
const HISTORY_MAX_SAMPLES: usize = 1440; // 24h at 1-min resolution const HISTORY_MAX_SAMPLES: usize = 1440; // 24h at 1-min resolution
/// Require at least this many samples before reporting an anomaly score.
/// Protects against false positives on a freshly-started daemon.
const ANOMALY_MIN_SAMPLES: u64 = 10;
/// Sum-of-z² above this value triggers a "warn" level anomaly.
/// Interpretation: average ~1.5σ deviation across 4 tracked metrics.
const ANOMALY_WARN_THRESHOLD: f64 = 9.0;
/// Sum-of-z² above this value triggers "alert".
const ANOMALY_ALERT_THRESHOLD: f64 = 16.0;
/// Welford's online algorithm for a running mean + variance.
/// Numerically stable, O(1) per update.
#[derive(Default, Clone)]
struct Welford {
n: u64,
mean: f64,
m2: f64,
}
impl Welford {
fn update(&mut self, x: f64) {
self.n += 1;
let delta = x - self.mean;
self.mean += delta / self.n as f64;
self.m2 += delta * (x - self.mean);
}
fn stddev(&self) -> f64 {
if self.n < 2 { return 1.0; }
(self.m2 / (self.n - 1) as f64).sqrt().max(1e-9)
}
/// Squared z-score for a new observation (non-destructive).
fn z_sq(&self, x: f64) -> f64 {
if self.n < ANOMALY_MIN_SAMPLES { return 0.0; }
let z = (x - self.mean) / self.stddev();
z * z
}
}
/// Per-node rolling stats for drift-adaptive anomaly detection.
#[derive(Default, Clone)]
struct NodeAnomalyStats {
hot_pct: Welford,
hit_rate: Welford,
fs_pct: Welford,
chunk_miss_rate: Welford,
/// Most recent anomaly score (sum of z²); 0 until ANOMALY_MIN_SAMPLES.
pub last_score: f64,
/// Total samples seen (mirrors hot_pct.n for convenience).
pub n: u64,
}
impl NodeAnomalyStats {
fn update(&mut self, s: &MetricSample) {
let hot_pct = if s.hot_max_bytes > 0 {
s.hot_used_bytes as f64 / s.hot_max_bytes as f64
} else {
0.0
};
let fs_pct = if s.fs_total_bytes > 0 {
s.fs_used_bytes as f64 / s.fs_total_bytes as f64
} else {
0.0
};
let total_chunks = s.has_chunk_hits + s.has_chunk_misses;
let chunk_miss_rate = if total_chunks > 0 {
s.has_chunk_misses as f64 / total_chunks as f64
} else {
0.0
};
self.last_score = self.hot_pct.z_sq(hot_pct)
+ self.hit_rate.z_sq(s.cache_hit_rate)
+ self.fs_pct.z_sq(fs_pct)
+ self.chunk_miss_rate.z_sq(chunk_miss_rate);
self.hot_pct.update(hot_pct);
self.hit_rate.update(s.cache_hit_rate);
self.fs_pct.update(fs_pct);
self.chunk_miss_rate.update(chunk_miss_rate);
self.n = self.hot_pct.n;
}
fn level(&self) -> &'static str {
if self.n < ANOMALY_MIN_SAMPLES { return "ok"; }
if self.last_score >= ANOMALY_ALERT_THRESHOLD { "alert" }
else if self.last_score >= ANOMALY_WARN_THRESHOLD { "warn" }
else { "ok" }
}
}
/// One time-series sample snapshotted from a peer's DashboardStatus RPC. /// One time-series sample snapshotted from a peer's DashboardStatus RPC.
#[derive(Debug, Clone, Serialize)] #[derive(Debug, Clone, Serialize)]
@@ -52,18 +142,41 @@ pub struct MetricSample {
pub fs_used_bytes: u64, pub fs_used_bytes: u64,
pub fs_total_bytes: u64, pub fs_total_bytes: u64,
pub fs_available_bytes: u64, pub fs_available_bytes: u64,
/// Drift-adaptive anomaly score (sum of z² across 4 metrics).
/// `None` until the node has collected ANOMALY_MIN_SAMPLES.
#[serde(skip_serializing_if = "Option::is_none")]
pub anomaly_score: Option<f64>,
}
#[derive(Serialize)]
pub struct AnomalyStatus {
pub node: String,
pub score: f64,
pub level: &'static str,
pub samples_used: u64,
} }
pub struct MetricsHistory { pub struct MetricsHistory {
samples: HashMap<String, VecDeque<MetricSample>>, samples: HashMap<String, VecDeque<MetricSample>>,
anomaly: HashMap<String, NodeAnomalyStats>,
} }
impl MetricsHistory { impl MetricsHistory {
fn new() -> Self { fn new() -> Self {
Self { samples: HashMap::new() } Self {
samples: HashMap::new(),
anomaly: HashMap::new(),
}
}
fn push(&mut self, node: &str, mut sample: MetricSample) {
// Update Welford stats and embed anomaly score in the sample.
let stats = self.anomaly.entry(node.to_string()).or_default();
stats.update(&sample);
if stats.n >= ANOMALY_MIN_SAMPLES {
sample.anomaly_score = Some(stats.last_score);
} }
fn push(&mut self, node: &str, sample: MetricSample) {
let deque = self.samples.entry(node.to_string()).or_default(); let deque = self.samples.entry(node.to_string()).or_default();
deque.push_back(sample); deque.push_back(sample);
while deque.len() > HISTORY_MAX_SAMPLES { while deque.len() > HISTORY_MAX_SAMPLES {
@@ -81,6 +194,21 @@ impl MetricsHistory {
}) })
.unwrap_or_default() .unwrap_or_default()
} }
pub fn anomaly_statuses(&self) -> Vec<AnomalyStatus> {
let mut out: Vec<AnomalyStatus> = self
.anomaly
.iter()
.map(|(node, s)| AnomalyStatus {
node: node.clone(),
score: s.last_score,
level: s.level(),
samples_used: s.n,
})
.collect();
out.sort_by(|a, b| a.node.cmp(&b.node));
out
}
} }
// ───────────────────────────────────────────────────────────────── // ─────────────────────────────────────────────────────────────────
@@ -1375,6 +1503,8 @@ async fn metrics_poller(state: Arc<V2State>) {
fs_used_bytes: r.filesystem.as_ref().map(|f| f.used_bytes).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_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), fs_available_bytes: r.filesystem.as_ref().map(|f| f.available_bytes).unwrap_or(0),
// Filled in by MetricsHistory::push after Welford update.
anomaly_score: None,
}; };
let mut hist = state.history.lock().await; let mut hist = state.history.lock().await;
hist.push(&peer_name, sample); hist.push(&peer_name, sample);
@@ -1402,6 +1532,11 @@ async fn handle_metrics_history(
Json(hist.get_last(&name, limit)) Json(hist.get_last(&name, limit))
} }
async fn handle_anomalies(State(s): State<Arc<V2State>>) -> Json<Vec<AnomalyStatus>> {
let hist = s.history.lock().await;
Json(hist.anomaly_statuses())
}
// ───────────────────────────────────────────────────────────────── // ─────────────────────────────────────────────────────────────────
pub fn build(state: Arc<V2State>) -> Router { pub fn build(state: Arc<V2State>) -> Router {
@@ -1426,6 +1561,7 @@ pub fn build(state: Arc<V2State>) -> Router {
.route("/api/v2/fleet", get(handle_fleet)) .route("/api/v2/fleet", get(handle_fleet))
.route("/api/v2/node/:name/status", get(handle_node_status)) .route("/api/v2/node/:name/status", get(handle_node_status))
.route("/api/v2/node/:name/metrics-history", get(handle_metrics_history)) .route("/api/v2/node/:name/metrics-history", get(handle_metrics_history))
.route("/api/v2/anomalies", get(handle_anomalies))
.route("/api/v2/storage/blobs", get(handle_blobs)) .route("/api/v2/storage/blobs", get(handle_blobs))
.route("/api/v2/storage/tags", get(handle_tags)) .route("/api/v2/storage/tags", get(handle_tags))
.route("/api/v2/storage/refs", get(handle_refs)) .route("/api/v2/storage/refs", get(handle_refs))
+12 -1
View File
@@ -5,12 +5,13 @@ import { NodeHistorySparklines } from './NodeHistorySparklines';
interface Props { interface Props {
node: NodeStatusV2; node: NodeStatusV2;
anomalyLevel?: 'ok' | 'warn' | 'alert';
} }
/// Human-oriented node card for the FleetHealth landing. /// Human-oriented node card for the FleetHealth landing.
/// Shows: overall health traffic-light, storage bars, mount state, /// Shows: overall health traffic-light, storage bars, mount state,
/// cache hit rate, next scheduled job. No hex, no primitives. /// cache hit rate, next scheduled job. No hex, no primitives.
export function NodeCard({ node }: Props) { export function NodeCard({ node, anomalyLevel }: Props) {
const health = healthOf(node); const health = healthOf(node);
const border = { const border = {
ok: 'border-emerald-700 hover:border-emerald-500', ok: 'border-emerald-700 hover:border-emerald-500',
@@ -41,6 +42,16 @@ export function NodeCard({ node }: Props) {
<span className="text-lg font-semibold text-slate-100"> <span className="text-lg font-semibold text-slate-100">
{node.node_name} {node.node_name}
</span> </span>
{anomalyLevel === 'alert' && (
<span className="text-xs font-mono px-1.5 py-0.5 rounded bg-red-900/60 text-red-300 border border-red-800">
anomaly
</span>
)}
{anomalyLevel === 'warn' && (
<span className="text-xs font-mono px-1.5 py-0.5 rounded bg-amber-900/60 text-amber-300 border border-amber-800">
drift
</span>
)}
</div> </div>
<span className="text-xs text-slate-500 font-mono"> <span className="text-xs text-slate-500 font-mono">
{node.zone || '—'} {node.zone || '—'}
+10
View File
@@ -155,6 +155,15 @@ export interface MetricSample {
fs_used_bytes: number; fs_used_bytes: number;
fs_total_bytes: number; fs_total_bytes: number;
fs_available_bytes: number; fs_available_bytes: number;
/** Sum of z² across 4 drift-adapted metrics. Absent until 10 samples collected. */
anomaly_score?: number;
}
export interface AnomalyStatus {
node: string;
score: number;
level: 'ok' | 'warn' | 'alert';
samples_used: number;
} }
export const api = { export const api = {
@@ -163,6 +172,7 @@ export const api = {
nodeStatus: (name: string) => get<NodeStatusV2>(`/v2/node/${name}/status`), nodeStatus: (name: string) => get<NodeStatusV2>(`/v2/node/${name}/status`),
metricsHistory: (name: string, limit = 60) => metricsHistory: (name: string, limit = 60) =>
get<MetricSample[]>(`/v2/node/${name}/metrics-history?limit=${limit}`), get<MetricSample[]>(`/v2/node/${name}/metrics-history?limit=${limit}`),
anomalies: () => get<AnomalyStatus[]>('/v2/anomalies'),
blobs: (limit = 200, offset = 0) => blobs: (limit = 200, offset = 0) =>
get<BlobSummary[]>(`/v2/storage/blobs?limit=${limit}&offset=${offset}`), get<BlobSummary[]>(`/v2/storage/blobs?limit=${limit}&offset=${offset}`),
tags: (prefix = '') => tags: (prefix = '') =>
+40 -2
View File
@@ -1,5 +1,5 @@
import { useEffect, useState } from 'react'; import { useEffect, useState } from 'react';
import { api, FleetSnapshot, fmtBytes, fmtAge } from '../lib/api'; import { api, FleetSnapshot, AnomalyStatus, fmtBytes, fmtAge } from '../lib/api';
import { NodeCard } from '../components/NodeCard'; import { NodeCard } from '../components/NodeCard';
import { ProjectsPanel } from '../components/ProjectsPanel'; import { ProjectsPanel } from '../components/ProjectsPanel';
@@ -7,6 +7,7 @@ import { ProjectsPanel } from '../components/ProjectsPanel';
// Polls the aggregator's /api/v2/fleet every 10 s. // Polls the aggregator's /api/v2/fleet every 10 s.
export function CommandCenter() { export function CommandCenter() {
const [fleet, setFleet] = useState<FleetSnapshot | null>(null); const [fleet, setFleet] = useState<FleetSnapshot | null>(null);
const [anomalies, setAnomalies] = useState<AnomalyStatus[]>([]);
const [err, setErr] = useState<string | null>(null); const [err, setErr] = useState<string | null>(null);
const [tick, setTick] = useState(0); const [tick, setTick] = useState(0);
@@ -18,6 +19,7 @@ export function CommandCenter() {
setErr(null); setErr(null);
}) })
.catch((e) => setErr(String(e))); .catch((e) => setErr(String(e)));
api.anomalies().then(setAnomalies).catch(() => {});
}, [tick]); }, [tick]);
useEffect(() => { useEffect(() => {
@@ -39,6 +41,13 @@ export function CommandCenter() {
) )
: null; : null;
// Build a node→level lookup for NodeCard props.
const anomalyMap = Object.fromEntries(
anomalies.map((a) => [a.node, a.level] as const)
);
const alertNodes = anomalies.filter((a) => a.level === 'alert');
const warnNodes = anomalies.filter((a) => a.level === 'warn');
return ( return (
<div className="space-y-6"> <div className="space-y-6">
<div> <div>
@@ -65,6 +74,31 @@ export function CommandCenter() {
</div> </div>
)} )}
{/* Anomaly banners (only shown once the ring buffer has 10+ samples) */}
{alertNodes.length > 0 && (
<div className="rounded border border-red-800 bg-red-950/40 px-4 py-3 text-sm flex items-start gap-3">
<span className="text-red-400 font-bold mt-0.5">●</span>
<div>
<span className="text-red-300 font-semibold">Metric anomaly detected — </span>
<span className="text-red-200">
{alertNodes.map((a) => a.node).join(', ')} deviating &gt;2σ from baseline
{alertNodes.length === 1 && ` (score ${alertNodes[0].score.toFixed(1)})`}
</span>
</div>
</div>
)}
{alertNodes.length === 0 && warnNodes.length > 0 && (
<div className="rounded border border-amber-800 bg-amber-950/40 px-4 py-3 text-sm flex items-start gap-3">
<span className="text-amber-400 font-bold mt-0.5">●</span>
<div>
<span className="text-amber-300 font-semibold">Metric drift — </span>
<span className="text-amber-200">
{warnNodes.map((a) => a.node).join(', ')} showing unusual patterns
</span>
</div>
</div>
)}
{totals && totals.diskTotal > 0 && ( {totals && totals.diskTotal > 0 && (
<div className="rounded-lg border border-slate-800 bg-slate-900/60 p-4 flex items-center justify-between"> <div className="rounded-lg border border-slate-800 bg-slate-900/60 p-4 flex items-center justify-between">
<div> <div>
@@ -89,7 +123,11 @@ export function CommandCenter() {
<h2 className="text-lg font-semibold text-slate-100 mb-3">Nodes</h2> <h2 className="text-lg font-semibold text-slate-100 mb-3">Nodes</h2>
<div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-3 gap-4"> <div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-3 gap-4">
{fleet?.nodes.map((n) => ( {fleet?.nodes.map((n) => (
<NodeCard key={n.node_name} node={n} /> <NodeCard
key={n.node_name}
node={n}
anomalyLevel={anomalyMap[n.node_name]}
/>
))} ))}
{!fleet && {!fleet &&
[1, 2, 3].map((i) => ( [1, 2, 3].map((i) => (