faucet_cli/sla/
eval.rs

1//! Pure SLA evaluation: staleness, static volume floor, and learned-baseline
2//! volume anomaly detection (z-score / Tukey IQR fences). No I/O — the
3//! orchestration in `sla::evaluate_post_run` owns state loading/persisting.
4
5use super::spec::{AnomalyMethod, SlaSpec, VolumeAnomalySpec};
6use super::state::SlaState;
7use std::fmt;
8
9/// One detected SLA violation.
10#[derive(Debug, Clone, PartialEq)]
11pub enum SlaViolation {
12    /// No successful run within `max_staleness_secs`.
13    Staleness { since_secs: u64, max_secs: u64 },
14    /// A successful run wrote fewer records than `min_rows_per_run`.
15    MinRows { rows: u64, min: u64 },
16    /// A successful run's volume is anomalous against the rolling baseline.
17    Volume { rows: u64, detail: String },
18}
19
20impl SlaViolation {
21    /// Stable metric-label value (`kind` on
22    /// `faucet_pipeline_sla_violations_total`).
23    pub fn kind(&self) -> &'static str {
24        match self {
25            SlaViolation::Staleness { .. } => "staleness",
26            SlaViolation::MinRows { .. } => "min_rows",
27            SlaViolation::Volume { .. } => "volume",
28        }
29    }
30}
31
32impl fmt::Display for SlaViolation {
33    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
34        match self {
35            SlaViolation::Staleness {
36                since_secs,
37                max_secs,
38            } => write!(
39                f,
40                "pipeline is stale: last success {since_secs}s ago exceeds max_staleness_secs {max_secs}"
41            ),
42            SlaViolation::MinRows { rows, min } => write!(
43                f,
44                "run wrote {rows} record(s), below min_rows_per_run {min}"
45            ),
46            SlaViolation::Volume { rows, detail } => {
47                write!(f, "run volume {rows} is anomalous: {detail}")
48            }
49        }
50    }
51}
52
53/// Checks that apply to a **successful** run: the static floor and the
54/// learned-baseline anomaly, both against the *prior* baseline (before this
55/// run's volume is folded in).
56pub fn evaluate_success(spec: &SlaSpec, prior: &SlaState, rows: u64) -> Vec<SlaViolation> {
57    let mut out = Vec::new();
58    if let Some(min) = spec.min_rows_per_run
59        && rows < min
60    {
61        out.push(SlaViolation::MinRows { rows, min });
62    }
63    if let Some(va) = &spec.volume_anomaly
64        && prior.volumes.len() >= va.min_history as usize
65        && let Some(detail) = detect_anomaly(&prior.volumes, rows, va)
66    {
67        out.push(SlaViolation::Volume { rows, detail });
68    }
69    out
70}
71
72/// Checks that apply to a **failed** run: staleness of the last success. A
73/// pipeline with no recorded success yet cannot be measured (cold start).
74pub fn evaluate_failure(spec: &SlaSpec, prior: &SlaState, now_unix: i64) -> Vec<SlaViolation> {
75    match (spec.max_staleness_secs, prior.last_success_unix) {
76        (Some(max_secs), Some(last)) => {
77            let since_secs = now_unix.saturating_sub(last).max(0) as u64;
78            if since_secs > max_secs {
79                vec![SlaViolation::Staleness {
80                    since_secs,
81                    max_secs,
82                }]
83            } else {
84                Vec::new()
85            }
86        }
87        _ => Vec::new(),
88    }
89}
90
91/// Run the configured detector; `Some(detail)` when `rows` is anomalous
92/// against `baseline`. Callers guarantee `baseline.len() >= min_history >= 2`.
93pub fn detect_anomaly(baseline: &[u64], rows: u64, va: &VolumeAnomalySpec) -> Option<String> {
94    let sensitivity = va.effective_sensitivity();
95    match va.method {
96        AnomalyMethod::Zscore => zscore_anomaly(baseline, rows, sensitivity),
97        AnomalyMethod::Iqr => iqr_anomaly(baseline, rows, sensitivity),
98    }
99}
100
101fn zscore_anomaly(baseline: &[u64], rows: u64, sensitivity: f64) -> Option<String> {
102    let n = baseline.len() as f64;
103    let mean = baseline.iter().map(|&v| v as f64).sum::<f64>() / n;
104    let var = baseline
105        .iter()
106        .map(|&v| {
107            let d = v as f64 - mean;
108            d * d
109        })
110        .sum::<f64>()
111        / n;
112    let std = var.sqrt();
113    let x = rows as f64;
114    if std == 0.0 {
115        // Constant baseline: any deviation is a regime change.
116        if x != mean {
117            return Some(format!(
118                "deviates from a constant baseline of {mean:.0} records/run"
119            ));
120        }
121        return None;
122    }
123    let z = (x - mean).abs() / std;
124    if z > sensitivity {
125        return Some(format!(
126            "|z| {z:.2} exceeds {sensitivity} (baseline mean {mean:.1}, std {std:.1}, n {})",
127            baseline.len()
128        ));
129    }
130    None
131}
132
133fn iqr_anomaly(baseline: &[u64], rows: u64, sensitivity: f64) -> Option<String> {
134    let mut sorted = baseline.to_vec();
135    sorted.sort_unstable();
136    let q1 = quantile(&sorted, 0.25);
137    let q3 = quantile(&sorted, 0.75);
138    let iqr = q3 - q1;
139    let lower = q1 - sensitivity * iqr;
140    let upper = q3 + sensitivity * iqr;
141    let x = rows as f64;
142    if x < lower || x > upper {
143        return Some(format!(
144            "outside [{lower:.1}, {upper:.1}] (q1 {q1:.1}, q3 {q3:.1}, fence {sensitivity}×IQR, n {})",
145            baseline.len()
146        ));
147    }
148    None
149}
150
151/// Linear-interpolation quantile (R type-7) over an ascending slice.
152/// Callers guarantee `sorted` is non-empty.
153fn quantile(sorted: &[u64], q: f64) -> f64 {
154    let n = sorted.len();
155    if n == 1 {
156        return sorted[0] as f64;
157    }
158    let pos = q * (n - 1) as f64;
159    let lo = pos.floor() as usize;
160    let hi = pos.ceil() as usize;
161    let frac = pos - lo as f64;
162    sorted[lo] as f64 + (sorted[hi] as f64 - sorted[lo] as f64) * frac
163}
164
165#[cfg(test)]
166mod tests {
167    use super::super::spec::{DEFAULT_MIN_HISTORY, DEFAULT_WINDOW};
168    use super::*;
169
170    fn spec(
171        staleness: Option<u64>,
172        min_rows: Option<u64>,
173        va: Option<VolumeAnomalySpec>,
174    ) -> SlaSpec {
175        SlaSpec {
176            max_staleness_secs: staleness,
177            min_rows_per_run: min_rows,
178            volume_anomaly: va,
179        }
180    }
181
182    fn va(method: AnomalyMethod, sensitivity: Option<f64>) -> VolumeAnomalySpec {
183        VolumeAnomalySpec {
184            method,
185            sensitivity,
186            min_history: DEFAULT_MIN_HISTORY,
187            window: DEFAULT_WINDOW,
188        }
189    }
190
191    fn state_with(volumes: &[u64], last_success: Option<i64>) -> SlaState {
192        SlaState {
193            last_success_unix: last_success,
194            volumes: volumes.to_vec(),
195        }
196    }
197
198    #[test]
199    fn min_rows_fires_below_floor_only() {
200        let s = spec(None, Some(10), None);
201        let prior = SlaState::default();
202        let v = evaluate_success(&s, &prior, 3);
203        assert_eq!(v.len(), 1);
204        assert_eq!(v[0].kind(), "min_rows");
205        assert!(v[0].to_string().contains("below min_rows_per_run 10"));
206        assert!(evaluate_success(&s, &prior, 10).is_empty());
207    }
208
209    #[test]
210    fn volume_anomaly_waits_for_min_history() {
211        let s = spec(None, None, Some(va(AnomalyMethod::Zscore, None)));
212        // 4 samples < min_history 5 → no evaluation even for a wild outlier.
213        let prior = state_with(&[100, 100, 100, 100], None);
214        assert!(evaluate_success(&s, &prior, 0).is_empty());
215        // 5 samples → the same outlier fires.
216        let prior = state_with(&[100, 100, 100, 100, 100], None);
217        let v = evaluate_success(&s, &prior, 0);
218        assert_eq!(v.len(), 1);
219        assert_eq!(v[0].kind(), "volume");
220    }
221
222    #[test]
223    fn zscore_flags_injected_drop_and_passes_normal() {
224        let baseline = [100, 105, 95, 102, 98, 101, 99, 103];
225        let cfg = va(AnomalyMethod::Zscore, None);
226        assert!(detect_anomaly(&baseline, 0, &cfg).is_some(), "drop to zero");
227        assert!(detect_anomaly(&baseline, 500, &cfg).is_some(), "spike");
228        assert!(detect_anomaly(&baseline, 101, &cfg).is_none(), "normal");
229    }
230
231    #[test]
232    fn zscore_constant_baseline_flags_any_deviation() {
233        let baseline = [50, 50, 50, 50, 50];
234        let cfg = va(AnomalyMethod::Zscore, None);
235        let detail = detect_anomaly(&baseline, 49, &cfg).expect("deviation from constant");
236        assert!(detail.contains("constant baseline"), "{detail}");
237        assert!(detect_anomaly(&baseline, 50, &cfg).is_none());
238    }
239
240    #[test]
241    fn zscore_sensitivity_widens_the_pass_band() {
242        let baseline = [100, 110, 90, 105, 95];
243        // 120 is ~2.7σ here: anomalous at sensitivity 1, normal at 3 (default).
244        assert!(detect_anomaly(&baseline, 120, &va(AnomalyMethod::Zscore, Some(1.0))).is_some());
245        assert!(detect_anomaly(&baseline, 120, &va(AnomalyMethod::Zscore, None)).is_none());
246    }
247
248    #[test]
249    fn iqr_flags_outliers_outside_fences() {
250        let baseline = [100, 102, 98, 101, 99, 103, 97, 100];
251        let cfg = va(AnomalyMethod::Iqr, None);
252        assert!(detect_anomaly(&baseline, 0, &cfg).is_some(), "drop");
253        assert!(detect_anomaly(&baseline, 1000, &cfg).is_some(), "spike");
254        assert!(detect_anomaly(&baseline, 100, &cfg).is_none(), "median");
255    }
256
257    #[test]
258    fn iqr_zero_spread_flags_any_outside_value() {
259        let baseline = [70, 70, 70, 70, 70];
260        let cfg = va(AnomalyMethod::Iqr, None);
261        assert!(detect_anomaly(&baseline, 71, &cfg).is_some());
262        assert!(detect_anomaly(&baseline, 70, &cfg).is_none());
263    }
264
265    #[test]
266    fn quantile_interpolates() {
267        let sorted = [10, 20, 30, 40];
268        assert_eq!(quantile(&sorted, 0.0), 10.0);
269        assert_eq!(quantile(&sorted, 1.0), 40.0);
270        assert_eq!(quantile(&sorted, 0.5), 25.0);
271        assert_eq!(quantile(&sorted, 0.25), 17.5);
272        assert_eq!(quantile(&[42], 0.75), 42.0);
273    }
274
275    #[test]
276    fn staleness_fires_only_past_threshold_with_history() {
277        let s = spec(Some(3600), None, None);
278        // Fresh enough.
279        let prior = state_with(&[], Some(10_000));
280        assert!(evaluate_failure(&s, &prior, 10_000 + 3600).is_empty());
281        // Stale.
282        let v = evaluate_failure(&s, &prior, 10_000 + 3601);
283        assert_eq!(v.len(), 1);
284        assert_eq!(v[0].kind(), "staleness");
285        assert!(v[0].to_string().contains("3601s ago"));
286        // No success ever recorded → unmeasurable, no violation.
287        assert!(evaluate_failure(&s, &SlaState::default(), 999_999).is_empty());
288        // No staleness configured → nothing.
289        let s = spec(None, Some(1), None);
290        assert!(evaluate_failure(&s, &prior, 999_999).is_empty());
291    }
292
293    #[test]
294    fn staleness_tolerates_clock_skew() {
295        // A last-success timestamp in the future must not underflow or fire.
296        let s = spec(Some(60), None, None);
297        let prior = state_with(&[], Some(2_000));
298        assert!(evaluate_failure(&s, &prior, 1_000).is_empty());
299    }
300
301    #[test]
302    fn success_checks_combine() {
303        let s = spec(Some(3600), Some(50), Some(va(AnomalyMethod::Zscore, None)));
304        let prior = state_with(&[100, 101, 99, 100, 100], Some(0));
305        let v = evaluate_success(&s, &prior, 10);
306        let kinds: Vec<_> = v.iter().map(|x| x.kind()).collect();
307        assert_eq!(kinds, vec!["min_rows", "volume"]);
308    }
309}
faucet_cli/sla/eval.rs

faucet_cli/sla/
eval.rs
