use crate::Config; use crate::SHUTDOWN; use crate::metrics::{Metric, MetricsSender}; use crate::pinger::PingReply; use crate::util::ArcMutex; use crate::util::MutexExt; use log::{debug, info}; use std::collections::HashMap; use std::net::IpAddr; use std::sync::atomic::Ordering; use std::sync::mpsc::{Receiver, Sender}; use std::time::Instant; #[derive(Copy, Clone)] pub struct ReflectorStats { pub down_ewma: f64, pub up_ewma: f64, pub last_receive_time_s: Instant, } pub struct Baseliner { pub config: Config, pub owd_baseline: ArcMutex>, pub owd_recent: ArcMutex>, pub reselect_trigger: Sender, pub start_time: Instant, pub stats_rx: Receiver, pub baseline_metrics: MetricsSender, pub event_metrics: MetricsSender, } fn ewma_factor(tick: f64, dur: f64) -> f64 { ((0.5_f64).ln() / (dur / tick)).exp() } impl Baseliner { pub fn run(&self) -> anyhow::Result<()> { /* * 135 seconds to decay to 50% for the slow factor and * 0.4 seconds to decay to 50% for the fast factor. * The fast one can be adjusted to tune, try anything from 0.01 to 3.0 to get more or less sensitivity * with more sensitivity we respond faster to bloat, but are at risk from triggering due to lag spikes that * aren't bloat related, with less sensitivity (bigger numbers) we smooth through quick spikes * but take longer to respond to real bufferbloat */ let slow_factor = ewma_factor(self.config.tick_interval, 135.0); let fast_factor = ewma_factor(self.config.tick_interval, 0.4); loop { if SHUTDOWN.load(Ordering::Relaxed) { info!("Baseliner shutting down"); return Ok(()); } let time_data = self.stats_rx.recv()?; let mut owd_baseline_map = self.owd_baseline.lock_anyhow()?; let mut owd_recent_map = self.owd_recent.lock_anyhow()?; let owd_baseline_new = ReflectorStats { down_ewma: time_data.down_time, up_ewma: time_data.up_time, last_receive_time_s: time_data.last_receive_time_s, }; let owd_recent_new = ReflectorStats { down_ewma: time_data.down_time, up_ewma: time_data.up_time, last_receive_time_s: time_data.last_receive_time_s, }; let owd_baseline = owd_baseline_map .entry(time_data.reflector) .or_insert(owd_baseline_new); let owd_recent = owd_recent_map .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; // 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; info!( "Reflector {} has OWD > 5 seconds more than baseline, triggering reselection", time_data.reflector ); self.event_metrics.send(Metric::Event { name: "reselection", reason: "anomaly", reflector: Some(time_data.reflector), tags: &[], }); // If reselection is disabled this would trigger an error // so just ignore the result let _ = self.reselect_trigger.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 { reflector: time_data.reflector, baseline_up_ewma: owd_baseline.up_ewma, baseline_down_ewma: owd_baseline.down_ewma, recent_up_ewma: owd_recent.up_ewma, recent_down_ewma: owd_recent.down_ewma, }); debug!( "Reflector {} up baseline = {} down baseline = {}", time_data.reflector, owd_baseline.up_ewma, owd_baseline.down_ewma ); debug!( "Reflector {} up recent = {} down recent = {}", time_data.reflector, owd_recent.up_ewma, owd_recent.down_ewma ); } } }