diff --git a/src/baseliner.rs b/src/baseliner.rs index b32ec47..e8a28da 100644 --- a/src/baseliner.rs +++ b/src/baseliner.rs @@ -40,6 +40,57 @@ fn ewma_factor(tick: f64, dur: f64) -> f64 { ((0.5_f64).ln() / (dur / tick)).exp() } +fn update_reflector_stats( + mut baseline: ReflectorStats, + mut recent: ReflectorStats, + down_time: f64, + up_time: f64, + receive_time: Instant, + start_time: Instant, + slow_factor: f64, + fast_factor: f64, +) -> (ReflectorStats, ReflectorStats, bool) { + if receive_time + .duration_since(baseline.last_receive_time_s) + .as_secs_f64() + > 30.0 + || receive_time + .duration_since(recent.last_receive_time_s) + .as_secs_f64() + > 30.0 + { + baseline.down_ewma = down_time; + baseline.up_ewma = up_time; + recent.down_ewma = down_time; + recent.up_ewma = up_time; + } + + baseline.last_receive_time_s = receive_time; + recent.last_receive_time_s = receive_time; + + if up_time > baseline.up_ewma + 5000.0 || down_time > baseline.down_ewma + 5000.0 { + baseline.last_receive_time_s = start_time; + recent.last_receive_time_s = start_time; + return (baseline, recent, true); + } + + baseline.down_ewma = baseline.down_ewma * slow_factor + (1.0 - slow_factor) * down_time; + baseline.up_ewma = baseline.up_ewma * slow_factor + (1.0 - slow_factor) * up_time; + + recent.down_ewma = recent.down_ewma * fast_factor + (1.0 - fast_factor) * down_time; + recent.up_ewma = recent.up_ewma * fast_factor + (1.0 - fast_factor) * up_time; + + if baseline.down_ewma > recent.down_ewma { + baseline.down_ewma = recent.down_ewma; + } + + if baseline.up_ewma > recent.up_ewma { + baseline.up_ewma = recent.up_ewma; + } + + (baseline, recent, false) +} + impl Baseliner { pub fn run(&self) -> anyhow::Result<()> { /* @@ -83,35 +134,21 @@ impl Baseliner { .entry(time_data.reflector) .or_insert(owd_recent_new); - if time_data - .last_receive_time_s - .duration_since(owd_baseline.last_receive_time_s) - .as_secs_f64() - > 30.0 - || time_data - .last_receive_time_s - .duration_since(owd_recent.last_receive_time_s) - .as_secs_f64() - > 30.0 - { - owd_baseline.down_ewma = time_data.down_time; - owd_baseline.up_ewma = time_data.up_time; - owd_baseline.last_receive_time_s = time_data.last_receive_time_s; - owd_recent.down_ewma = time_data.down_time; - owd_recent.up_ewma = time_data.up_time; - owd_recent.last_receive_time_s = time_data.last_receive_time_s; - } - - owd_baseline.last_receive_time_s = time_data.last_receive_time_s; - owd_recent.last_receive_time_s = time_data.last_receive_time_s; + let (updated_baseline, updated_recent, anomaly) = update_reflector_stats( + *owd_baseline, + *owd_recent, + time_data.down_time, + time_data.up_time, + time_data.last_receive_time_s, + self.start_time, + slow_factor, + fast_factor, + ); + *owd_baseline = updated_baseline; + *owd_recent = updated_recent; // if this reflection is more than 5 seconds higher than baseline... mark it no good and trigger a reselection - if time_data.up_time > owd_baseline.up_ewma + 5000.0 - || time_data.down_time > owd_baseline.down_ewma + 5000.0 - { - // mark the data as bad by setting the receive time to the time autorate was started - owd_baseline.last_receive_time_s = self.start_time; - owd_recent.last_receive_time_s = self.start_time; + if anomaly { info!( "Reflector {} has OWD > 5 seconds more than baseline, triggering reselection", time_data.reflector @@ -125,24 +162,6 @@ impl Baseliner { // The reselect channel is bounded to 1, // so we use try_send to avoid blocking if the channel is full let _ = self.reselect_trigger.try_send(true); - } else { - owd_baseline.down_ewma = owd_baseline.down_ewma * slow_factor - + (1.0 - slow_factor) * time_data.down_time; - owd_baseline.up_ewma = - owd_baseline.up_ewma * slow_factor + (1.0 - slow_factor) * time_data.up_time; - - owd_recent.down_ewma = - owd_recent.down_ewma * fast_factor + (1.0 - fast_factor) * time_data.down_time; - owd_recent.up_ewma = - owd_recent.up_ewma * fast_factor + (1.0 - fast_factor) * time_data.up_time; - - if owd_baseline.down_ewma > owd_recent.down_ewma { - owd_baseline.down_ewma = owd_recent.down_ewma; - } - - if owd_baseline.up_ewma > owd_recent.up_ewma { - owd_baseline.up_ewma = owd_recent.up_ewma; - } } self.baseline_metrics.send(Metric::Baseline { @@ -164,3 +183,159 @@ impl Baseliner { } } } + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + fn stats(down_ewma: f64, up_ewma: f64, last_receive_time_s: Instant) -> ReflectorStats { + ReflectorStats { + down_ewma, + up_ewma, + last_receive_time_s, + } + } + + fn assert_close(actual: f64, expected: f64) { + assert!( + (actual - expected).abs() < 1e-9, + "expected {expected}, got {actual}" + ); + } + + #[test] + fn updates_ewmas() { + let start = Instant::now(); + let receive_time = start + Duration::from_secs(1); + let (baseline, recent, anomaly) = update_reflector_stats( + stats(100.0, 100.0, start), + stats(100.0, 100.0, start), + 200.0, + 300.0, + receive_time, + start, + 0.9, + 0.5, + ); + + assert!(!anomaly); + assert_close(baseline.down_ewma, 110.0); + assert_close(baseline.up_ewma, 120.0); + assert_close(recent.down_ewma, 150.0); + assert_close(recent.up_ewma, 200.0); + assert_eq!(baseline.last_receive_time_s, receive_time); + assert_eq!(recent.last_receive_time_s, receive_time); + } + + #[test] + fn keeps_baseline_at_or_below_recent() { + let start = Instant::now(); + let (baseline, recent, anomaly) = update_reflector_stats( + stats(200.0, 200.0, start), + stats(100.0, 100.0, start), + 100.0, + 100.0, + start + Duration::from_secs(1), + start, + 0.9, + 0.5, + ); + + assert!(!anomaly); + assert_close(baseline.down_ewma, 100.0); + assert_close(baseline.up_ewma, 100.0); + assert_close(recent.down_ewma, 100.0); + assert_close(recent.up_ewma, 100.0); + } + + #[test] + fn resets_stats_after_a_long_gap() { + let start = Instant::now(); + let receive_time = start + Duration::from_secs(31); + let (baseline, recent, anomaly) = update_reflector_stats( + stats(100.0, 200.0, start), + stats(300.0, 400.0, start), + 500.0, + 600.0, + receive_time, + start, + 0.9, + 0.5, + ); + + assert!(!anomaly); + assert_close(baseline.down_ewma, 500.0); + assert_close(baseline.up_ewma, 600.0); + assert_close(recent.down_ewma, 500.0); + assert_close(recent.up_ewma, 600.0); + assert_eq!(baseline.last_receive_time_s, receive_time); + assert_eq!(recent.last_receive_time_s, receive_time); + } + + #[test] + fn marks_anomalies_for_reselection() { + let start = Instant::now(); + let (baseline, recent, anomaly) = update_reflector_stats( + stats(100.0, 200.0, start), + stats(300.0, 400.0, start), + 5_101.0, + 200.0, + start + Duration::from_secs(1), + start, + 0.9, + 0.5, + ); + + assert!(anomaly); + assert_close(baseline.down_ewma, 100.0); + assert_close(baseline.up_ewma, 200.0); + assert_close(recent.down_ewma, 300.0); + assert_close(recent.up_ewma, 400.0); + assert_eq!(baseline.last_receive_time_s, start); + assert_eq!(recent.last_receive_time_s, start); + } + + #[test] + fn does_not_mark_exact_anomaly_threshold() { + let start = Instant::now(); + let receive_time = start + Duration::from_secs(1); + let (baseline, recent, anomaly) = update_reflector_stats( + stats(100.0, 200.0, start), + stats(100.0, 200.0, start), + 5_100.0, + 5_200.0, + receive_time, + start, + 0.9, + 0.5, + ); + + assert!(!anomaly); + assert_eq!(baseline.last_receive_time_s, receive_time); + assert_eq!(recent.last_receive_time_s, receive_time); + } + + #[test] + fn resets_both_tracks_when_either_track_is_stale() { + let start = Instant::now(); + let recent_time = start + Duration::from_secs(20); + let receive_time = start + Duration::from_secs(31); + let (baseline, recent, anomaly) = update_reflector_stats( + stats(100.0, 200.0, start), + stats(300.0, 400.0, recent_time), + 500.0, + 600.0, + receive_time, + start, + 0.9, + 0.5, + ); + + assert!(!anomaly); + assert_close(baseline.down_ewma, 500.0); + assert_close(baseline.up_ewma, 600.0); + assert_close(recent.down_ewma, 500.0); + assert_close(recent.up_ewma, 600.0); + } +}