From a00270521ac085d27a33bdc8b4732a32640f0f0e Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 13 Oct 2026 15:11:31 +0700 Subject: [PATCH] feat(measurement): integrate IRTT with full configuration and per-stream tracking Extend reflector configuration to support IRTT-specific parameters (ports, HMAC) and track per-stream measurement state, enabling dynamic selection and authentication for IRTT probes. --- src/baseliner.rs | 26 ++++---- src/controller.rs | 122 +++++++++++++++++++++++------------- src/main.rs | 126 ++++++++++++++++++++++++++++++-------- src/measurement/irtt.rs | 21 +++++-- src/ratecontroller.rs | 38 ++++++++---- src/reflector_selector.rs | 64 ++++++++++++++----- src/settings/mod.rs | 86 +++++++++++++++++++++++--- 7 files changed, 366 insertions(+), 117 deletions(-) diff --git a/src/baseliner.rs b/src/baseliner.rs index 4908c43..c80e5a5 100644 --- a/src/baseliner.rs +++ b/src/baseliner.rs @@ -6,13 +6,12 @@ // SPDX-License-Identifier: MPL-2.0 use crate::SHUTDOWN; -use crate::measurement::model::{MeasurementEvent, MeasurementObservation}; +use crate::measurement::model::{MeasurementEvent, MeasurementObservation, MeasurementStream}; use crate::metrics::{Metric, MetricsSender}; use crate::settings::Settings; use flume::{Receiver, Sender}; use log::{debug, info}; use std::collections::HashMap; -use std::net::IpAddr; use std::sync::atomic::Ordering; use std::time::Duration; use std::time::Instant; @@ -72,7 +71,7 @@ impl ReflectorState { #[derive(Clone, Debug)] pub struct ControlSnapshot { pub generated_at: Instant, - pub reflectors: HashMap, + pub reflectors: HashMap, } pub struct Baseliner { @@ -121,7 +120,7 @@ fn update_reflector_state( } fn process_reply( - reflectors: &mut HashMap, + reflectors: &mut HashMap, reply: &MeasurementObservation, start_time: Instant, slow_factor: f64, @@ -129,7 +128,7 @@ fn process_reply( ) -> (ReflectorState, bool) { let sample = EwmaStats::from_reply(reply); let state = reflectors - .entry(reply.stream.peer) + .entry(reply.stream) .or_insert_with(|| ReflectorState::new(sample, reply.observed_at)); let anomaly = update_reflector_state( state, @@ -144,7 +143,7 @@ fn process_reply( } fn control_snapshot( - reflectors: &HashMap, + reflectors: &HashMap, generated_at: Instant, ) -> ControlSnapshot { ControlSnapshot { @@ -155,7 +154,7 @@ fn control_snapshot( impl Baseliner { pub fn run(&self) -> anyhow::Result<()> { - let mut reflectors = HashMap::::new(); + let mut reflectors = HashMap::::new(); let slow_factor = ewma_factor( self.settings.advanced_settings.tick_interval, @@ -249,6 +248,7 @@ mod tests { MeasurementSource, MeasurementStream, MeasurementStreamId, OneWayClock, OneWayLatency, SignedDuration, }; + use std::net::IpAddr; use std::time::Duration; fn state( @@ -434,7 +434,8 @@ mod tests { let received_at = start + Duration::from_secs(1); let reflector = "192.0.2.1".parse().unwrap(); let mut reflectors = HashMap::new(); - reflectors.insert(reflector, state(100.0, 100.0, 100.0, 100.0, start)); + let stream = reply(reflector, 0.0, 0.0, start).stream; + reflectors.insert(stream, state(100.0, 100.0, 100.0, 100.0, start)); let (updated, anomaly) = process_reply( &mut reflectors, @@ -450,7 +451,7 @@ mod tests { assert_close(updated.recent.down, 150.0); assert_close(updated.recent.up, 200.0); assert_eq!(updated.last_receive_at, received_at); - assert_close(reflectors[&reflector].recent.up, 200.0); + assert_close(reflectors[&stream].recent.up, 200.0); } #[test] @@ -469,14 +470,15 @@ mod tests { 0.5, ); let snapshot = control_snapshot(&reflectors, generated_at); + let stream = reply(reflector, 0.0, 0.0, received_at).stream; assert!(!anomaly); assert_eq!(snapshot.generated_at, generated_at); assert_close( - snapshot.reflectors[&reflector].baseline.down, + snapshot.reflectors[&stream].baseline.down, updated.baseline.down, ); - assert_close(snapshot.reflectors[&reflector].recent.up, updated.recent.up); - assert_eq!(snapshot.reflectors[&reflector].last_receive_at, received_at); + assert_close(snapshot.reflectors[&stream].recent.up, updated.recent.up); + assert_eq!(snapshot.reflectors[&stream].last_receive_at, received_at); } } diff --git a/src/controller.rs b/src/controller.rs index 11408a8..188f4ef 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -1,100 +1,140 @@ -use std::{collections::HashMap, net::IpAddr, time::Duration}; +use std::{ + collections::HashMap, + net::SocketAddr, + sync::{Arc, RwLock}, + time::Duration, +}; use flume::{Receiver, SendError, Sender}; use thiserror::Error; -use crate::measurement::{ - icmp::IcmpBinding, - model::{MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule}, +use crate::{ + measurement::{ + icmp::IcmpBinding, + irtt::IrttBinding, + model::{MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule}, + }, + settings::ReflectorDefinition, }; #[derive(Debug, Error)] pub enum ControllerError { #[error("ICMP channel error")] IcmpChannelError(#[from] SendError>), + #[error("IRTT channel error")] + IrttChannelError(#[from] SendError>), } pub struct Controller { - reflector_rx: Receiver>, + reflector_rx: Receiver>, icmp_binding_tx: Sender>, - + irtt_binding_tx: Sender>, + reflectors: HashMap, + active_streams: Arc>>, + endpoint_streams: Arc>>, source: MeasurementSource, period: Duration, - next_stream_id: u64, - stream_ids: HashMap<(IpAddr, MeasurementSource), MeasurementStreamId>, + stream_ids: HashMap<(SocketAddr, MeasurementSource), MeasurementStreamId>, } impl Controller { pub fn new( - reflector_rx: Receiver>, + reflector_rx: Receiver>, icmp_binding_tx: Sender>, + irtt_binding_tx: Sender>, + reflectors: HashMap, + active_streams: Arc>>, + endpoint_streams: Arc>>, + source: MeasurementSource, ) -> Self { Self { reflector_rx, icmp_binding_tx, - source: MeasurementSource::IcmpTimestamp, + irtt_binding_tx, + reflectors, + active_streams, + endpoint_streams, + source, period: Duration::from_millis(500), next_stream_id: 0, stream_ids: HashMap::new(), } } - pub async fn run(mut self, initial_reflectors: Vec) -> Result<(), ControllerError> { + pub async fn run(mut self, initial_reflectors: Vec) -> Result<(), ControllerError> { self.update_reflectors(initial_reflectors).await?; - while let Ok(reflectors) = self.reflector_rx.recv_async().await { self.update_reflectors(reflectors).await?; } - Ok(()) } - fn stream_for(&mut self, peer: IpAddr) -> MeasurementStream { - let key = (peer, self.source); - + fn stream_for(&mut self, endpoint: SocketAddr) -> MeasurementStream { + let key = (endpoint, self.source); let id = *self.stream_ids.entry(key).or_insert_with(|| { let id = MeasurementStreamId(self.next_stream_id); self.next_stream_id += 1; id }); - MeasurementStream { id, - peer, + peer: endpoint.ip(), source: self.source, } } - async fn update_reflectors(&mut self, reflectors: Vec) -> Result<(), ControllerError> { - let count = reflectors.len(); - - let bindings = reflectors - .into_iter() - .enumerate() - .map(|(index, peer)| { - let stream = self.stream_for(peer); - - IcmpBinding::new( - stream, - ProbeSchedule { - period: self.period, - offset: evenly_spaced_offset(self.period, index, count), - }, + async fn update_reflectors( + &mut self, + endpoints: Vec, + ) -> Result<(), ControllerError> { + let count = endpoints.len(); + let mut icmp = Vec::new(); + let mut irtt = Vec::new(); + let mut streams = Vec::with_capacity(count); + let mut endpoint_streams = HashMap::new(); + for (index, endpoint) in endpoints.into_iter().enumerate() { + let Some((port, hmac)) = self.reflectors.get(&endpoint).map(|definition| { + ( + definition.irtt_port.unwrap_or(2112), + definition.irtt_hmac.clone(), ) - }) - .collect(); - - self.icmp_binding_tx - .send_async(bindings) - .await - .map_err(|e| ControllerError::IcmpChannelError(e)) + }) else { + continue; + }; + let stream = self.stream_for(endpoint); + let schedule = ProbeSchedule { + period: self.period, + offset: evenly_spaced_offset(self.period, index, count), + }; + match self.source { + MeasurementSource::IcmpEcho | MeasurementSource::IcmpTimestamp => { + icmp.push(IcmpBinding::new(stream, schedule)) + } + MeasurementSource::Irtt => { + irtt.push(IrttBinding::new(stream, schedule, port, hmac)) + } + _ => unreachable!("unsupported reflector measurement source"), + } + streams.push(stream); + endpoint_streams.insert(endpoint, stream); + } + *self + .active_streams + .write() + .expect("active stream lock poisoned") = streams; + *self + .endpoint_streams + .write() + .expect("endpoint stream lock poisoned") = endpoint_streams; + self.icmp_binding_tx.send_async(icmp).await?; + self.irtt_binding_tx.send_async(irtt).await?; + Ok(()) } } fn evenly_spaced_offset(period: Duration, index: usize, count: usize) -> Duration { let nanos = period.as_nanos() * index as u128 / count as u128; - Duration::new( (nanos / 1_000_000_000) as u64, (nanos % 1_000_000_000) as u32, diff --git a/src/main.rs b/src/main.rs index 6d9bd1d..5977206 100644 --- a/src/main.rs +++ b/src/main.rs @@ -23,16 +23,17 @@ use crate::baseliner::Baseliner; use crate::controller::Controller; use crate::measurement::icmp::IcmpEngine; use crate::measurement::irtt::IrttEngine; +use crate::measurement::model::{MeasurementSource, MeasurementStream}; use crate::metrics::{Metric, Metrics, MetricsSender}; use crate::platform::{TrafficControlBackend, traffic_control_backend, warn_platform_limitations}; use crate::ratecontroller::Ratecontroller; use crate::reflector_selector::ReflectorSelector; -use crate::settings::Settings; +use crate::settings::{ReflectorDefinition, ReflectorProtocol, Settings}; use crate::util::RwLockExt; use ::log::{info, warn}; use flume::RecvTimeoutError; -use std::net::IpAddr; -use std::str::FromStr; +use std::collections::HashMap; +use std::net::{IpAddr, SocketAddr, ToSocketAddrs}; use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use std::sync::{Arc, RwLock}; use std::thread; @@ -40,8 +41,11 @@ use std::thread::{JoinHandle, sleep}; use std::time::{Duration, Instant}; struct ReflectorSetup { - peers: Arc>>, - pool: Vec, + peers: Arc>>, + active_streams: Arc>>, + endpoint_streams: Arc>>, + pool: Vec, + reflectors: HashMap, reselection_enabled: bool, } pub static SHUTDOWN: AtomicBool = AtomicBool::new(false); @@ -135,29 +139,84 @@ fn restore_shaper( } } +fn resolve_reflectors( + definitions: &[ReflectorDefinition], + protocol: ReflectorProtocol, + countries: &[String], + filter_country: bool, +) -> anyhow::Result> { + let mut resolved = HashMap::new(); + for definition in definitions { + if !definition.protocols.contains(&protocol) { + continue; + } + if protocol != ReflectorProtocol::Irtt + && definition + .host + .parse::() + .is_ok_and(|ip| ip.is_ipv6()) + { + continue; + } + if filter_country + && !countries.is_empty() + && !definition.country.as_ref().is_some_and(|country| { + countries + .iter() + .any(|allowed| allowed.eq_ignore_ascii_case(country)) + }) + { + continue; + } + let port = if protocol == ReflectorProtocol::Irtt { + definition.irtt_port.unwrap_or(2112) + } else { + 0 + }; + for endpoint in (definition.host.as_str(), port).to_socket_addrs()? { + if protocol != ReflectorProtocol::Irtt && endpoint.is_ipv6() { + continue; + } + resolved.insert(endpoint, definition.clone()); + } + } + Ok(resolved) +} + fn setup_reflectors(settings: &Settings) -> anyhow::Result { - let configured = settings.load_reflectors()?; - let configured_count = configured.len(); - - let default_reflectors = [ - IpAddr::from_str("9.9.9.9")?, - IpAddr::from_str("8.238.120.14")?, - IpAddr::from_str("74.82.42.42")?, - IpAddr::from_str("194.242.2.2")?, - IpAddr::from_str("208.67.222.222")?, - IpAddr::from_str("94.140.14.14")?, - ]; - - let reselection_enabled = configured_count > settings.reflectors.num_reflectors as usize; - let pool = if reselection_enabled { + let protocol = settings.advanced_settings.measurement_type; + let configured = resolve_reflectors( + &settings.reflectors.entries, + protocol, + &settings.reflectors.countries, + true, + )?; + let bootstrap = resolve_reflectors(&settings.reflectors.bootstrap, protocol, &[], false)?; + let mut reflectors = bootstrap.clone(); + reflectors.extend( configured - } else { - Vec::new() - }; + .iter() + .map(|(endpoint, definition)| (*endpoint, definition.clone())), + ); + let mut pool = configured.keys().copied().collect::>(); + pool.sort(); + let mut initial = bootstrap.keys().copied().collect::>(); + initial.sort(); + if initial.is_empty() { + initial = pool + .iter() + .take(settings.reflectors.num_reflectors as usize) + .copied() + .collect(); + } + let reselection_enabled = pool.len() > settings.reflectors.num_reflectors as usize; Ok(ReflectorSetup { - peers: Arc::new(RwLock::new(default_reflectors.to_vec())), + peers: Arc::new(RwLock::new(initial)), + active_streams: Arc::new(RwLock::new(Vec::new())), + endpoint_streams: Arc::new(RwLock::new(HashMap::new())), pool, + reflectors, reselection_enabled, }) } @@ -230,7 +289,10 @@ fn run(settings: &Settings) -> anyhow::Result<()> { let ReflectorSetup { peers: reflector_peers, + active_streams, + endpoint_streams, pool: reflector_pool, + reflectors, reselection_enabled, } = setup_reflectors(settings)?; @@ -248,7 +310,20 @@ fn run(settings: &Settings) -> anyhow::Result<()> { (None, None) }; - let controller = Controller::new(reflector_rx, icmp_binding_tx); + let source = match settings.advanced_settings.measurement_type { + ReflectorProtocol::IcmpEcho => MeasurementSource::IcmpEcho, + ReflectorProtocol::IcmpTimestamps => MeasurementSource::IcmpTimestamp, + ReflectorProtocol::Irtt => MeasurementSource::Irtt, + }; + let controller = Controller::new( + reflector_rx, + icmp_binding_tx, + irtt_binding_tx, + reflectors, + Arc::clone(&active_streams), + Arc::clone(&endpoint_streams), + source, + ); let icmp_engine = { let _guard = rt.enter(); IcmpEngine::new( @@ -336,6 +411,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { .expect("reselection snapshot receiver must exist when reselection is enabled"), reflector_peers_lock: Arc::clone(&reflector_peers), reflector_pool, + endpoint_streams, trigger_channel: reselect_rx, metrics: event_metrics, reflector_tx, @@ -349,7 +425,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { let mut ratecontroller = Ratecontroller::new( settings.clone(), control_snapshot_rx, - reflector_peers, + active_streams, reselect_tx, rate_metrics, )?; diff --git a/src/measurement/irtt.rs b/src/measurement/irtt.rs index 77c7f8d..c100ca6 100644 --- a/src/measurement/irtt.rs +++ b/src/measurement/irtt.rs @@ -8,7 +8,7 @@ use irtt_client::{ ManagedCommandAcknowledgement, ManagedCommandApplyError, ManagedCommandError, ManagedCompletionPolicy, ManagedConfigError, ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedSubscribeError, ManagedTargetConfig, ManagedTargetEndReason, - TargetId, + TargetAuth, TargetId, }, }; use log::{debug, trace, warn}; @@ -54,14 +54,21 @@ pub struct IrttBinding { stream: MeasurementStream, schedule: ProbeSchedule, port: u16, + hmac: Option, } impl IrttBinding { - pub fn new(stream: MeasurementStream, schedule: ProbeSchedule, port: u16) -> Self { + pub fn new( + stream: MeasurementStream, + schedule: ProbeSchedule, + port: u16, + hmac: Option, + ) -> Self { Self { stream, schedule, port, + hmac, } } } @@ -112,10 +119,16 @@ impl IrttEngine { let targets = bindings .iter() .map(|binding| { - ManagedTargetConfig::new( + let mut target = ManagedTargetConfig::new( TargetId::new(binding.stream.id.0.to_string()), SocketAddr::new(binding.stream.peer, binding.port).to_string(), - ) + ); + if let Some(key) = &binding.hmac { + target.auth = TargetAuth::Override(irtt_client::Authentication::Hmac( + irtt_client::HmacKey::new(key.as_bytes().to_vec().into_boxed_slice()), + )); + } + target }) .collect(); diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index 230ba81..40263fa 100644 --- a/src/ratecontroller.rs +++ b/src/ratecontroller.rs @@ -7,6 +7,7 @@ use crate::SHUTDOWN; use crate::baseliner::ControlSnapshot; +use crate::measurement::model::MeasurementStream; use crate::metrics::{Metric, MetricsSender}; use crate::platform::{ InterfaceStatsProvider, PlatformInterfaceStats, PlatformTrafficControl, TrafficControlBackend, @@ -19,7 +20,6 @@ use flume::{Receiver, Sender}; use log::{debug, info, warn}; use std::fs::File; use std::io::Write; -use std::net::IpAddr; use std::sync::atomic::Ordering; use std::thread::sleep; use std::time::{Duration, Instant}; @@ -100,7 +100,7 @@ fn drain_latest_snapshot( fn deltas_from_snapshot( snapshot: Option<&ControlSnapshot>, - reflectors: &[IpAddr], + reflectors: &[MeasurementStream], now: Instant, tick_interval: f64, ) -> (Vec, Vec) { @@ -123,8 +123,8 @@ fn deltas_from_snapshot( up.push(up_delta); debug!( - "Reflector: {} down_delay: {} up_delay: {}", - reflector, down_delta, up_delta + "Reflector: {} ({:?}) down_delay: {} up_delay: {}", + reflector.peer, reflector.source, down_delta, up_delta ); } } @@ -139,7 +139,7 @@ pub struct Ratecontroller { settings: Settings, control_snapshot: Option, control_snapshot_rx: Receiver, - reflectors_lock: ArcRwLock>, + reflectors_lock: ArcRwLock>, reselect_trigger: Sender, state_dl: State, state_ul: State, @@ -153,7 +153,7 @@ impl Ratecontroller { fn new_with_backends( settings: Settings, control_snapshot_rx: Receiver, - reflectors_lock: ArcRwLock>, + reflectors_lock: ArcRwLock>, reselect_trigger: Sender, metrics: MetricsSender, mut stats_provider: S, @@ -509,7 +509,7 @@ impl Ratecontroller { pub fn new( settings: Settings, control_snapshot_rx: Receiver, - reflectors_lock: ArcRwLock>, + reflectors_lock: ArcRwLock>, reselect_trigger: Sender, metrics: MetricsSender, ) -> anyhow::Result { @@ -530,7 +530,17 @@ impl Ratecontroller { mod tests { use super::*; use crate::baseliner::{EwmaStats, ReflectorState}; + use crate::measurement::model::{MeasurementSource, MeasurementStreamId}; use std::collections::HashMap; + use std::net::IpAddr; + + fn stream(id: u64, peer: IpAddr) -> MeasurementStream { + MeasurementStream { + id: MeasurementStreamId(id), + peer, + source: MeasurementSource::IcmpTimestamp, + } + } #[test] fn fixed_seed_reproduces_initial_safe_rates() { @@ -564,7 +574,7 @@ mod tests { } fn snapshot(generated_at: Instant, marker: f64) -> ControlSnapshot { - let reflector = "192.0.2.1".parse().unwrap(); + let reflector = stream(0, "192.0.2.1".parse().unwrap()); ControlSnapshot { generated_at, reflectors: HashMap::from([( @@ -590,7 +600,9 @@ mod tests { assert!(rx.is_empty()); assert_eq!(latest.generated_at, newest_at); assert_eq!( - latest.reflectors[&"192.0.2.1".parse().unwrap()].recent.down, + latest.reflectors[&stream(0, "192.0.2.1".parse().unwrap())] + .recent + .down, 2.0 ); } @@ -598,10 +610,10 @@ mod tests { #[test] fn calculates_sorted_deltas_from_active_fresh_snapshot_state() { let now = Instant::now(); - let first: IpAddr = "192.0.2.1".parse().unwrap(); - let second: IpAddr = "192.0.2.2".parse().unwrap(); - let stale: IpAddr = "192.0.2.3".parse().unwrap(); - let inactive: IpAddr = "192.0.2.4".parse().unwrap(); + let first = stream(0, "192.0.2.1".parse().unwrap()); + let second = stream(1, "192.0.2.2".parse().unwrap()); + let stale = stream(2, "192.0.2.3".parse().unwrap()); + let inactive = stream(3, "192.0.2.4".parse().unwrap()); let snapshot = ControlSnapshot { generated_at: now, reflectors: HashMap::from([ diff --git a/src/reflector_selector.rs b/src/reflector_selector.rs index c38b979..5078218 100644 --- a/src/reflector_selector.rs +++ b/src/reflector_selector.rs @@ -7,12 +7,13 @@ use crate::SHUTDOWN; use crate::baseliner::ControlSnapshot; +use crate::measurement::model::MeasurementStream; use crate::metrics::{Metric, MetricsSender}; use crate::settings::Settings; use crate::util::RwLockExt; use flume::{Receiver, RecvError, RecvTimeoutError, Selector, Sender}; use log::{debug, info}; -use std::net::IpAddr; +use std::net::SocketAddr; use std::sync::atomic::Ordering; use std::sync::{Arc, RwLock}; use std::thread::sleep; @@ -98,18 +99,25 @@ fn receive_snapshots_until( } } -fn recent_rtt_for_peer(snapshot: Option<&ControlSnapshot>, peer: &IpAddr) -> Option { +fn recent_rtt_for_peer( + snapshot: Option<&ControlSnapshot>, + stream: Option<&MeasurementStream>, + now: Instant, + max_age: Duration, +) -> Option { snapshot - .and_then(|snapshot| snapshot.reflectors.get(peer)) + .and_then(|snapshot| stream.and_then(|stream| snapshot.reflectors.get(stream))) + .filter(|state| now.saturating_duration_since(state.last_receive_at) < max_age) .map(|state| (state.recent.down + state.recent.up) as u64) } pub struct ReflectorSelector { pub settings: Settings, pub snapshot_rx: Receiver, - pub reflector_peers_lock: Arc>>, - pub reflector_tx: Sender>, - pub reflector_pool: Vec, + pub reflector_peers_lock: Arc>>, + pub reflector_tx: Sender>, + pub reflector_pool: Vec, + pub endpoint_streams: Arc>>, pub trigger_channel: Receiver, pub metrics: MetricsSender, } @@ -163,8 +171,9 @@ impl ReflectorSelector { Duration::new(self.settings.reflectors.peer_reselection_time * 60, 0); } - let mut next_peers: Vec = Vec::new(); + let mut next_peers: Vec = Vec::new(); let mut reflectors_peers = self.reflector_peers_lock.write_anyhow()?; + let current_peers = reflectors_peers.clone(); // Include all current peers for reflector in reflectors_peers.iter() { @@ -172,6 +181,9 @@ impl ReflectorSelector { next_peers.push(*reflector); } + if self.reflector_pool.is_empty() { + continue; + } for _ in 0..20 { let next_candidate = &self.reflector_pool[fastrand::usize(..self.reflector_pool.len())]; @@ -205,7 +217,13 @@ impl ReflectorSelector { let mut candidates = Vec::new(); for peer in next_peers { - if let Some(rtt) = recent_rtt_for_peer(latest_snapshot.as_ref(), &peer) { + let stream = self.endpoint_streams.read_anyhow()?.get(&peer).copied(); + if let Some(rtt) = recent_rtt_for_peer( + latest_snapshot.as_ref(), + stream.as_ref(), + Instant::now(), + Duration::from_secs_f64(self.settings.advanced_settings.tick_interval * 2.0), + ) { candidates.push((peer, rtt)); info!("Candidate reflector: {} RTT: {}", peer, rtt); } else { @@ -220,7 +238,7 @@ impl ReflectorSelector { candidates.sort_by(|a, b| a.1.cmp(&b.1)); // Now we will just limit the candidates down to 2 * num_reflectors - let mut num_reflectors = self.settings.reflectors.num_reflectors; + let num_reflectors = self.settings.reflectors.num_reflectors; let candidate_pool_num = (2 * num_reflectors) as usize; candidates.truncate(candidate_pool_num); @@ -235,7 +253,14 @@ impl ReflectorSelector { } if (candidates.len() as u8) < num_reflectors { - num_reflectors = candidates.len() as u8; + info!( + "Only {} usable reflectors, retaining current set of {}", + candidates.len(), + num_reflectors + ); + self.reflector_tx.send(current_peers.clone())?; + *reflectors_peers = current_peers; + continue; } let mut new_peers = Vec::new(); @@ -246,7 +271,7 @@ impl ReflectorSelector { self.metrics.send(Metric::Event { name: "reflector_selected", reason: "", - reflector: Some(peer), + reflector: Some(peer.ip()), tags: &[], }); } @@ -261,7 +286,9 @@ impl ReflectorSelector { mod tests { use super::*; use crate::baseliner::{EwmaStats, ReflectorState}; + use crate::measurement::model::MeasurementSource; use std::collections::HashMap; + use std::net::IpAddr; use std::thread; fn snapshot(generated_at: Instant) -> ControlSnapshot { @@ -274,7 +301,11 @@ mod tests { fn snapshot_with_rtt(generated_at: Instant, peer: IpAddr, rtt: f64) -> ControlSnapshot { let mut reflectors = HashMap::new(); reflectors.insert( - peer, + crate::measurement::model::MeasurementStream { + id: crate::measurement::model::MeasurementStreamId(0), + peer, + source: MeasurementSource::IcmpTimestamp, + }, ReflectorState { baseline: EwmaStats { down: rtt / 2.0, @@ -374,7 +405,7 @@ mod tests { #[test] fn baseline_wait_retains_snapshot_for_evaluation() { let (snapshot_tx, snapshot_rx) = flume::unbounded(); - let peer = "192.0.2.1".parse().unwrap(); + let peer: IpAddr = "192.0.2.1".parse().unwrap(); let generated_at = Instant::now(); snapshot_tx .send(snapshot_with_rtt(generated_at, peer, 20.0)) @@ -395,7 +426,12 @@ mod tests { ); assert_eq!( - recent_rtt_for_peer(latest_snapshot.as_ref(), &peer), + recent_rtt_for_peer( + latest_snapshot.as_ref(), + latest_snapshot.as_ref().unwrap().reflectors.keys().next(), + generated_at + Duration::from_secs(1), + Duration::from_secs(2) + ), Some(42) ); } diff --git a/src/settings/mod.rs b/src/settings/mod.rs index ec757fa..514a9c4 100644 --- a/src/settings/mod.rs +++ b/src/settings/mod.rs @@ -140,6 +140,16 @@ pub(crate) struct ReflectorDefinition { pub(crate) irtt_hmac: Option, } +#[derive(Debug, Deserialize)] +struct ReflectorRow { + #[serde(alias = "reflector_ip")] + host: String, + country: Option, + protocols: Option, + irtt_port: Option, + irtt_hmac: Option, +} + #[derive(Clone, Debug, Deserialize, UciConfig)] #[uci(package = "sqm-autorate-rust")] pub(crate) struct Settings { @@ -176,15 +186,11 @@ impl Settings { ); let builder = builder.add_source(Environment::with_prefix("SQMA").separator("__")); - let built_config = builder.build()?; + let mut settings: Self = builder.build()?.try_deserialize()?; - built_config.try_deserialize() - } + settings.reflectors.load_files()?; - pub fn load_reflectors(&self) -> anyhow::Result> { - let mut reflectors = Vec::with_capacity(50); - - Ok(reflectors) + Ok(settings) } } @@ -233,13 +239,77 @@ pub(crate) struct ReflectorSettings { pub(crate) peer_reselection_time: u64, } +impl ReflectorSettings { + fn load_files(&mut self) -> Result<(), ConfigError> { + for path in &self.files { + let mut reader = csv::Reader::from_path(path) + .map_err(|error| ConfigError::Message(error.to_string()))?; + for row in reader.deserialize::() { + let row = row.map_err(|error| ConfigError::Message(error.to_string()))?; + let protocols = match row.protocols { + Some(protocols) => protocols + .split(['|', ';']) + .map(str::trim) + .filter(|protocol| !protocol.is_empty()) + .map(ReflectorProtocol::from_str) + .collect::, _>>()?, + None => vec![ + ReflectorProtocol::IcmpEcho, + ReflectorProtocol::IcmpTimestamps, + ], + }; + self.entries.push(ReflectorDefinition { + host: row.host, + country: row.country, + protocols, + irtt_port: row.irtt_port, + irtt_hmac: row.irtt_hmac, + }); + } + } + + Ok(()) + } +} + impl Default for ReflectorSettings { fn default() -> Self { + let icmp = |host: &str| ReflectorDefinition { + host: host.to_owned(), + country: None, + protocols: vec![ + ReflectorProtocol::IcmpEcho, + ReflectorProtocol::IcmpTimestamps, + ], + irtt_port: None, + irtt_hmac: None, + }; + + let irtt = |host: &str, country: Option<&str>| ReflectorDefinition { + host: host.to_owned(), + country: country.map(str::to_owned), + protocols: vec![ReflectorProtocol::Irtt], + irtt_port: Some(2112), + irtt_hmac: None, + }; + Self { files: vec!["/etc/sqm-autorate-rust/reflectors.csv".into()], countries: Vec::new(), entries: Vec::new(), - bootstrap: Vec::new(), + bootstrap: vec![ + icmp("9.9.9.9"), + icmp("8.238.120.14"), + icmp("74.82.42.42"), + icmp("194.242.2.2"), + icmp("208.67.222.222"), + icmp("94.140.14.14"), + irtt("global.irtt.lochnair.net", None), + irtt("uk-cov1.irtt.lochnair.net", Some("GB")), + irtt("uk-cov2.irtt.lochnair.net", Some("GB")), + irtt("de-fra.irtt.lochnair.net", Some("DE")), + irtt("nl-ams1.irtt.lochnair.net", Some("NL")), + ], num_reflectors: 5, peer_reselection_time: 15, }