faucet_cli/sla/
spec.rs

1//! Config types for the top-level `sla:` block (#202).
2//!
3//! An SLA declares freshness and volume expectations for a pipeline. It is
4//! pipeline-level in v1 (no matrix-row override, like `resilience:`) and is
5//! evaluated after every **root** invocation by the executor — children fan
6//! out per parent record, so their volumes are not a stable series to baseline.
7
8use schemars::JsonSchema;
9use serde::{Deserialize, Serialize};
10
11/// Default number of successful runs required before volume anomaly detection
12/// starts firing (cold-start guard).
13pub const DEFAULT_MIN_HISTORY: u32 = 5;
14/// Default rolling-window size (successful runs kept in the volume baseline).
15pub const DEFAULT_WINDOW: u32 = 20;
16/// Default z-score threshold.
17pub const DEFAULT_ZSCORE_SENSITIVITY: f64 = 3.0;
18/// Default Tukey-fence IQR multiplier.
19pub const DEFAULT_IQR_SENSITIVITY: f64 = 1.5;
20
21/// Declared service-level agreement for a pipeline: freshness and volume
22/// expectations. Violations emit the
23/// `faucet_pipeline_sla_violations_total{pipeline,row,kind}` counter and a
24/// structured warning log; they never fail the run.
25#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
26#[serde(deny_unknown_fields)]
27pub struct SlaSpec {
28    /// Maximum seconds since the last *successful* run before the pipeline
29    /// counts as stale. Evaluated when a run fails (against the previous
30    /// success) and by `faucet doctor`. Requires a `state:` block to persist
31    /// the last-success timestamp.
32    #[serde(default, skip_serializing_if = "Option::is_none")]
33    pub max_staleness_secs: Option<u64>,
34
35    /// Static volume floor: a successful run that writes fewer records than
36    /// this violates the SLA (catches a source silently returning nothing).
37    /// Stateless — works without a `state:` block.
38    #[serde(default, skip_serializing_if = "Option::is_none")]
39    pub min_rows_per_run: Option<u64>,
40
41    /// Learned-baseline volume anomaly detection over recent successful runs.
42    /// Requires a `state:` block to persist the rolling baseline.
43    #[serde(default, skip_serializing_if = "Option::is_none")]
44    pub volume_anomaly: Option<VolumeAnomalySpec>,
45}
46
47/// How a run's record volume is flagged as anomalous against the rolling
48/// baseline of recent successful runs.
49#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
50#[serde(deny_unknown_fields)]
51pub struct VolumeAnomalySpec {
52    /// Detection method. Default `zscore`.
53    #[serde(default)]
54    pub method: AnomalyMethod,
55
56    /// Detection threshold. For `zscore`: the maximum |x − mean| / std
57    /// (default 3.0). For `iqr`: the Tukey fence multiplier — a volume outside
58    /// [Q1 − k·IQR, Q3 + k·IQR] is anomalous (default 1.5).
59    #[serde(default, skip_serializing_if = "Option::is_none")]
60    pub sensitivity: Option<f64>,
61
62    /// Minimum successful runs of history before detection starts (cold-start
63    /// guard). Default 5; must be at least 2.
64    #[serde(default = "default_min_history")]
65    pub min_history: u32,
66
67    /// Rolling-window size: how many recent successful-run volumes form the
68    /// baseline. Default 20; must be ≥ `min_history`.
69    #[serde(default = "default_window")]
70    pub window: u32,
71}
72
73fn default_min_history() -> u32 {
74    DEFAULT_MIN_HISTORY
75}
76
77fn default_window() -> u32 {
78    DEFAULT_WINDOW
79}
80
81/// Volume anomaly detection method.
82#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
83#[serde(rename_all = "snake_case")]
84pub enum AnomalyMethod {
85    /// Flag when |volume − mean| / std exceeds `sensitivity`.
86    #[default]
87    Zscore,
88    /// Flag when the volume falls outside the Tukey fences
89    /// [Q1 − k·IQR, Q3 + k·IQR] with k = `sensitivity`.
90    Iqr,
91}
92
93impl VolumeAnomalySpec {
94    /// The configured sensitivity, or the method's conventional default.
95    pub fn effective_sensitivity(&self) -> f64 {
96        self.sensitivity.unwrap_or(match self.method {
97            AnomalyMethod::Zscore => DEFAULT_ZSCORE_SENSITIVITY,
98            AnomalyMethod::Iqr => DEFAULT_IQR_SENSITIVITY,
99        })
100    }
101}
102
103impl SlaSpec {
104    /// Fail-fast validation, surfaced at config-load time by `expand`.
105    pub fn validate(&self) -> Result<(), String> {
106        if self.max_staleness_secs.is_none()
107            && self.min_rows_per_run.is_none()
108            && self.volume_anomaly.is_none()
109        {
110            return Err("declares no checks — set max_staleness_secs, \
111                 min_rows_per_run, or volume_anomaly"
112                .into());
113        }
114        if self.max_staleness_secs == Some(0) {
115            return Err("max_staleness_secs must be at least 1".into());
116        }
117        if self.min_rows_per_run == Some(0) {
118            return Err("min_rows_per_run must be at least 1 (omit the field to disable)".into());
119        }
120        if let Some(va) = &self.volume_anomaly {
121            if let Some(s) = va.sensitivity
122                && (!s.is_finite() || s <= 0.0)
123            {
124                return Err(format!(
125                    "volume_anomaly.sensitivity must be a finite number > 0, got {s}"
126                ));
127            }
128            if va.min_history < 2 {
129                return Err(format!(
130                    "volume_anomaly.min_history must be at least 2, got {}",
131                    va.min_history
132                ));
133            }
134            if va.window < va.min_history {
135                return Err(format!(
136                    "volume_anomaly.window ({}) must be >= min_history ({})",
137                    va.window, va.min_history
138                ));
139            }
140        }
141        Ok(())
142    }
143
144    /// Whether any configured check needs persisted history (a `state:` block).
145    pub fn needs_state(&self) -> bool {
146        self.max_staleness_secs.is_some() || self.volume_anomaly.is_some()
147    }
148
149    /// The rolling-window size to keep in the persisted baseline. Volumes are
150    /// tracked even when `volume_anomaly` is unset (as long as a state store
151    /// exists) so enabling anomaly detection later starts with warm history.
152    pub fn window(&self) -> usize {
153        self.volume_anomaly
154            .as_ref()
155            .map(|va| va.window as usize)
156            .unwrap_or(DEFAULT_WINDOW as usize)
157    }
158}
159
160#[cfg(test)]
161mod tests {
162    use super::*;
163
164    fn minimal() -> SlaSpec {
165        SlaSpec {
166            max_staleness_secs: Some(3600),
167            min_rows_per_run: None,
168            volume_anomaly: None,
169        }
170    }
171
172    #[test]
173    fn empty_spec_is_rejected() {
174        let s = SlaSpec {
175            max_staleness_secs: None,
176            min_rows_per_run: None,
177            volume_anomaly: None,
178        };
179        let err = s.validate().unwrap_err();
180        assert!(err.contains("declares no checks"), "{err}");
181    }
182
183    #[test]
184    fn zero_staleness_and_zero_min_rows_are_rejected() {
185        let mut s = minimal();
186        s.max_staleness_secs = Some(0);
187        assert!(s.validate().unwrap_err().contains("max_staleness_secs"));
188
189        let mut s = minimal();
190        s.min_rows_per_run = Some(0);
191        assert!(s.validate().unwrap_err().contains("min_rows_per_run"));
192    }
193
194    #[test]
195    fn bad_sensitivity_is_rejected() {
196        for bad in [0.0, -1.0, f64::NAN, f64::INFINITY] {
197            let s = SlaSpec {
198                max_staleness_secs: None,
199                min_rows_per_run: None,
200                volume_anomaly: Some(VolumeAnomalySpec {
201                    method: AnomalyMethod::Zscore,
202                    sensitivity: Some(bad),
203                    min_history: DEFAULT_MIN_HISTORY,
204                    window: DEFAULT_WINDOW,
205                }),
206            };
207            assert!(
208                s.validate().unwrap_err().contains("sensitivity"),
209                "sensitivity {bad} should be rejected"
210            );
211        }
212    }
213
214    #[test]
215    fn window_and_min_history_bounds() {
216        let s = SlaSpec {
217            max_staleness_secs: None,
218            min_rows_per_run: None,
219            volume_anomaly: Some(VolumeAnomalySpec {
220                method: AnomalyMethod::Iqr,
221                sensitivity: None,
222                min_history: 1,
223                window: DEFAULT_WINDOW,
224            }),
225        };
226        assert!(s.validate().unwrap_err().contains("min_history"));
227
228        let s = SlaSpec {
229            max_staleness_secs: None,
230            min_rows_per_run: None,
231            volume_anomaly: Some(VolumeAnomalySpec {
232                method: AnomalyMethod::Iqr,
233                sensitivity: None,
234                min_history: 10,
235                window: 5,
236            }),
237        };
238        assert!(s.validate().unwrap_err().contains("window"));
239    }
240
241    #[test]
242    fn effective_sensitivity_defaults_per_method() {
243        let z = VolumeAnomalySpec {
244            method: AnomalyMethod::Zscore,
245            sensitivity: None,
246            min_history: 5,
247            window: 20,
248        };
249        assert_eq!(z.effective_sensitivity(), DEFAULT_ZSCORE_SENSITIVITY);
250        let i = VolumeAnomalySpec {
251            method: AnomalyMethod::Iqr,
252            sensitivity: None,
253            min_history: 5,
254            window: 20,
255        };
256        assert_eq!(i.effective_sensitivity(), DEFAULT_IQR_SENSITIVITY);
257        let e = VolumeAnomalySpec {
258            sensitivity: Some(2.5),
259            ..z
260        };
261        assert_eq!(e.effective_sensitivity(), 2.5);
262    }
263
264    #[test]
265    fn needs_state_reflects_configured_checks() {
266        assert!(minimal().needs_state());
267        let rows_only = SlaSpec {
268            max_staleness_secs: None,
269            min_rows_per_run: Some(1),
270            volume_anomaly: None,
271        };
272        assert!(!rows_only.needs_state());
273        assert!(rows_only.validate().is_ok());
274    }
275
276    #[test]
277    fn deserializes_from_yaml_with_defaults() {
278        let s: SlaSpec =
279            serde_yaml::from_str("max_staleness_secs: 900\nvolume_anomaly:\n  method: iqr\n")
280                .unwrap();
281        assert_eq!(s.max_staleness_secs, Some(900));
282        let va = s.volume_anomaly.unwrap();
283        assert_eq!(va.method, AnomalyMethod::Iqr);
284        assert_eq!(va.min_history, DEFAULT_MIN_HISTORY);
285        assert_eq!(va.window, DEFAULT_WINDOW);
286        assert!(va.sensitivity.is_none());
287    }
288
289    #[test]
290    fn unknown_fields_are_rejected() {
291        let r: Result<SlaSpec, _> = serde_yaml::from_str("max_staleness: 900\n");
292        assert!(r.is_err());
293    }
294}
faucet_cli/sla/spec.rs

faucet_cli/sla/
spec.rs
