diff --git a/Cargo.lock b/Cargo.lock index c44c291..452bdfb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -60,6 +60,15 @@ version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] + [[package]] name = "bumpalo" version = "3.20.3" @@ -123,6 +132,12 @@ dependencies = [ "cc", ] +[[package]] +name = "cmov" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" + [[package]] name = "config" version = "0.15.27" @@ -135,6 +150,21 @@ dependencies = [ "winnow", ] +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + [[package]] name = "csv" version = "1.4.0" @@ -156,6 +186,27 @@ dependencies = [ "memchr", ] +[[package]] +name = "ctutils" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03bb0e1cc970d482d121d9a1744999169b69a07470b3d644a7894e53fcaf4574" +dependencies = [ + "cmov", +] + +[[package]] +name = "digest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" +dependencies = [ + "block-buffer", + "const-oid", + "crypto-common", + "ctutils", +] + [[package]] name = "either" version = "1.19.0" @@ -242,6 +293,24 @@ version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +[[package]] +name = "hmac" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" +dependencies = [ + "digest", +] + +[[package]] +name = "hybrid-array" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17" +dependencies = [ + "typenum", +] + [[package]] name = "icmp-socket2" version = "0.6.4" @@ -265,6 +334,31 @@ dependencies = [ "hashbrown", ] +[[package]] +name = "irtt-client" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a59c24704bd94f19b84b2b0190aa9163efede499f8cf4eb02922443cfee08339" +dependencies = [ + "irtt-proto", + "nix", + "socket2", + "thiserror", + "tokio", +] + +[[package]] +name = "irtt-proto" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a56ff307983b8f6477724e08cf726c5e24975b9affeb853d0830d91c60026ab2" +dependencies = [ + "hmac", + "md-5", + "subtle", + "thiserror", +] + [[package]] name = "itertools" version = "0.13.0" @@ -337,6 +431,16 @@ version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest", +] + [[package]] name = "memchr" version = "2.8.3" @@ -679,6 +783,7 @@ dependencies = [ "fastrand", "flume", "icmp-socket2", + "irtt-client", "libc", "log", "netlink-bindings", @@ -697,6 +802,12 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "85ec932792e62b2d3936163eacbd2e09984453bcd114f665c1e1694e36ff59c9" +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "2.0.119" @@ -836,6 +947,12 @@ dependencies = [ "toml", ] +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "uci-derive" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 777f1c5..b4df466 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,7 @@ csv = "1.4.0" fastrand = "2" flume = { version = "0.12.0", default-features = false, features = ["async", "eventual-fairness", "select"] } icmp-socket2 = { version = "0.6.4", features = ["tokio"] } +irtt-client = { version = "0.6.0", optional = true, features = ["ancillary", "tokio"] } libc = "0.2" rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] } serde = { version = "1", features = ["derive"] } @@ -44,7 +45,8 @@ netlink-socket2 = { version = "0.3.5" } rust-uci = { git = "https://github.com/Lochnair/rust-uci.git", rev = "6835397f4f4ba8b69e75e727105186865c357417", optional = true, default-features = false } [features] -default = ["toml", "uci-vendored"] +default = ["irtt", "toml", "uci-vendored"] +irtt = ["dep:irtt-client"] toml = ["config/toml"] uci = ["dep:rust-uci"] uci-vendored = ["uci", "rust-uci/vendored"] diff --git a/src/baseliner.rs b/src/baseliner.rs index 4908c43..8363050 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( @@ -289,6 +289,7 @@ mod tests { }, started_at: received_at, observed_at: received_at, + round_trip: None, one_way: Some(OneWayLatency { uplink: SignedDuration::from_millis(up_time as i64), downlink: SignedDuration::from_millis(down_time as i64), @@ -434,7 +435,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 +452,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 +471,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..1573c13 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -1,100 +1,150 @@ -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}, +#[cfg(feature = "irtt")] +use crate::measurement::irtt::IrttBinding; +use crate::{ + measurement::{ + icmp::IcmpBinding, + model::{MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule}, + }, + settings::ReflectorDefinition, }; #[derive(Debug, Error)] pub enum ControllerError { #[error("ICMP channel error")] IcmpChannelError(#[from] SendError>), + #[cfg(feature = "irtt")] + #[error("IRTT channel error")] + IrttChannelError(#[from] SendError>), } pub struct Controller { - reflector_rx: Receiver>, + reflector_rx: Receiver>, icmp_binding_tx: Sender>, - + #[cfg(feature = "irtt")] + 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>, + #[cfg(feature = "irtt")] irtt_binding_tx: Sender>, + reflectors: HashMap, + active_streams: Arc>>, + endpoint_streams: Arc>>, + source: MeasurementSource, + period: Duration, ) -> Self { Self { reflector_rx, icmp_binding_tx, - source: MeasurementSource::IcmpTimestamp, - period: Duration::from_millis(500), + #[cfg(feature = "irtt")] + irtt_binding_tx, + reflectors, + active_streams, + endpoint_streams, + source, + period, 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(); + #[cfg(feature = "irtt")] + 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(_definition) = self.reflectors.get(&endpoint) else { + continue; + }; + #[cfg(feature = "irtt")] + let (port, hmac) = { + ( + _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)) + }; + 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)) + } + #[cfg(feature = "irtt")] + MeasurementSource::Irtt => irtt.push(IrttBinding::new(stream, port, hmac)), + #[cfg(not(feature = "irtt"))] + MeasurementSource::Irtt => unreachable!("IRTT feature is checked before startup"), + _ => 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?; + #[cfg(feature = "irtt")] + 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 560cab9..1bda3c4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -22,16 +22,19 @@ mod util; use crate::baseliner::Baseliner; use crate::controller::Controller; use crate::measurement::icmp::IcmpEngine; +#[cfg(feature = "irtt")] +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; @@ -39,8 +42,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); @@ -134,29 +140,126 @@ fn restore_shaper( } } -fn setup_reflectors(settings: &Settings) -> anyhow::Result { - let configured = settings.load_reflectors()?; - let configured_count = configured.len(); +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 + }; + let endpoints = match (definition.host.as_str(), port).to_socket_addrs() { + Ok(endpoints) => endpoints, + Err(error) => { + warn!("Failed to resolve reflector {}: {error}", definition.host); + continue; + } + }; + for endpoint in endpoints { + if protocol != ReflectorProtocol::Irtt && endpoint.is_ipv6() { + continue; + } + insert_reflector(&mut resolved, endpoint, definition.clone())?; + } + } + Ok(resolved) +} - 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.advanced_settings.num_reflectors as usize; - let pool = if reselection_enabled { - configured +fn insert_reflector( + reflectors: &mut HashMap, + endpoint: SocketAddr, + definition: ReflectorDefinition, +) -> anyhow::Result<()> { + if let Some(existing) = reflectors.get_mut(&endpoint) { + let same_country = match (&existing.country, &definition.country) { + (Some(a), Some(b)) => a.eq_ignore_ascii_case(b), + _ => true, + }; + let same_protocols = existing + .protocols + .iter() + .all(|p| definition.protocols.contains(p)) + && definition + .protocols + .iter() + .all(|p| existing.protocols.contains(p)); + if !same_country + || !same_protocols + || existing.irtt_port.unwrap_or(2112) != definition.irtt_port.unwrap_or(2112) + || existing.irtt_hmac != definition.irtt_hmac + { + anyhow::bail!("conflicting reflector definitions for resolved endpoint {endpoint}"); + } + if existing.country.is_none() { + existing.country = definition.country; + } } else { - Vec::new() - }; + reflectors.insert(endpoint, definition); + } + Ok(()) +} + +fn setup_reflectors(settings: &Settings) -> anyhow::Result { + 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(); + for (endpoint, definition) in &configured { + insert_reflector(&mut reflectors, *endpoint, definition.clone())?; + } + if reflectors.is_empty() { + anyhow::bail!("no usable reflectors for measurement protocol {protocol}"); + } + 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, }) } @@ -219,6 +322,12 @@ fn wait_for_exit(error_rx: &flume::Receiver) -> anyhow::Result<() } fn run(settings: &Settings) -> anyhow::Result<()> { + #[cfg(not(feature = "irtt"))] + if settings.advanced_settings.measurement_type == ReflectorProtocol::Irtt { + anyhow::bail!( + "IRTT measurement selected, but this binary was built without the `irtt` feature" + ); + } let rt = tokio::runtime::Runtime::new()?; if settings.advanced_settings.dry_run { @@ -229,13 +338,18 @@ fn run(settings: &Settings) -> anyhow::Result<()> { let ReflectorSetup { peers: reflector_peers, + active_streams, + endpoint_streams, pool: reflector_pool, + reflectors, reselection_enabled, } = setup_reflectors(settings)?; let (control_snapshot_tx, control_snapshot_rx) = flume::unbounded(); let (error_tx, error_rx) = flume::unbounded::(); let (icmp_binding_tx, icmp_binding_rx) = flume::unbounded(); + #[cfg(feature = "irtt")] + let (irtt_binding_tx, irtt_binding_rx) = flume::unbounded(); let (measurement_tx, measurement_rx) = flume::unbounded(); let (reflector_tx, reflector_rx) = flume::unbounded(); let (reselect_tx, reselect_rx) = flume::bounded(1); @@ -246,18 +360,45 @@ fn run(settings: &Settings) -> anyhow::Result<()> { (None, None) }; - let controller = Controller::new(reflector_rx, icmp_binding_tx); - let engine = { + 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, + #[cfg(feature = "irtt")] + irtt_binding_tx, + reflectors, + Arc::clone(&active_streams), + Arc::clone(&endpoint_streams), + source, + Duration::from_secs_f64(settings.advanced_settings.tick_interval), + ); + let icmp_engine = { let _guard = rt.enter(); IcmpEngine::new( start_time, &settings.network, - measurement_tx, + measurement_tx.clone(), icmp_binding_rx, )? }; + #[cfg(feature = "irtt")] + let (irtt_task, irtt_engine) = { + let _guard = rt.enter(); + IrttEngine::new( + Duration::from_secs_f64(settings.advanced_settings.tick_interval), + &settings.network, + measurement_tx, + irtt_binding_rx, + )? + }; - spawn_task(&rt, &error_tx, engine.run()); + spawn_task(&rt, &error_tx, icmp_engine.run()); + #[cfg(feature = "irtt")] + spawn_task(&rt, &error_tx, irtt_engine.run(irtt_task)); let dropped = Arc::new(AtomicU32::new(0)); @@ -329,6 +470,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, @@ -342,7 +484,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/icmp.rs b/src/measurement/icmp.rs index 3d96b3b..cbdeb59 100644 --- a/src/measurement/icmp.rs +++ b/src/measurement/icmp.rs @@ -326,6 +326,7 @@ impl IcmpEngine { stream: probe.stream, started_at: probe.sent_at, observed_at, + round_trip: None, one_way: None, }) } @@ -385,6 +386,7 @@ impl IcmpEngine { stream: probe.stream, started_at: probe.sent_at, observed_at, + round_trip: None, one_way: Some(OneWayLatency { uplink: SignedDuration::from_millis(uplink_ms), downlink: SignedDuration::from_millis(downlink_ms), diff --git a/src/measurement/irtt.rs b/src/measurement/irtt.rs new file mode 100644 index 0000000..ad6b818 --- /dev/null +++ b/src/measurement/irtt.rs @@ -0,0 +1,418 @@ +use std::{collections::HashMap, net::SocketAddr}; + +use flume::{Receiver, Sender}; +use irtt_client::{ + ClientConfig, ClientEvent, NegotiationPolicy, OpenPolicy, + managed::{ + ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask, + ManagedCommandApplyError, ManagedCommandError, ManagedCompletionPolicy, ManagedConfigError, + ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedSubscribeError, + ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetLifecycle, TargetAuth, TargetId, + }, +}; +use log::{debug, trace, warn}; +use thiserror::Error; +use tokio::sync::broadcast::error::RecvError; + +use crate::SHUTDOWN; +use crate::measurement::model::{ + MeasurementEvent, MeasurementLoss, MeasurementObservation, MeasurementStream, + MeasurementStreamId, OneWayClock, OneWayLatency, SignedDuration, +}; +use crate::settings::NetworkSettings; + +#[derive(Debug, Error)] +pub enum IrttError { + #[error("failed to build IRTT config")] + InvalidConfig(#[from] ManagedConfigError), + + #[error("event channel closed")] + EventChannelClosed, + + #[error("binding channel closed")] + BindingChannelClosed, + + #[error("error while reciving on channel")] + ChannelRecvError(#[from] RecvError), + + #[error("unable to subscribe to irtt events")] + SubscriptionError(#[from] ManagedSubscribeError), + + #[error("failed to submit IRTT target update")] + CommandError(#[from] ManagedCommandError), + + #[error("failed to apply IRTT target update")] + CommandApplyError(#[from] ManagedCommandApplyError), + + #[error("IRTT managed client failed: {0:?}")] + ManagedFailure(ManagedEndReason), + + #[error("IRTT managed task was abandoned")] + ManagedAbandoned, +} + +pub struct IrttBinding { + stream: MeasurementStream, + port: u16, + hmac: Option, +} + +impl IrttBinding { + pub fn new(stream: MeasurementStream, port: u16, hmac: Option) -> Self { + Self { stream, port, hmac } + } +} + +pub struct IrttEngine { + client: ManagedClientHandle, + + bindings: HashMap, + targets: Vec, + + binding_rx: Receiver>, + observation_tx: Sender, +} + +impl IrttEngine { + pub fn new( + interval: std::time::Duration, + settings: &NetworkSettings, + observation_tx: Sender, + binding_rx: Receiver>, + ) -> Result<(ManagedClientTask, Self), IrttError> { + let config = managed_client_config(interval, settings); + let (task, handle) = ManagedClient::task(config, Vec::new())?; + + Ok(( + task, + Self { + client: handle, + bindings: HashMap::new(), + targets: Vec::new(), + binding_rx, + observation_tx, + }, + )) + } + + async fn set_bindings(&mut self, bindings: Vec) -> Result<(), IrttError> { + let targets = bindings + .iter() + .map(|binding| { + 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.as_ref().filter(|key| !key.trim().is_empty()) { + target.auth = TargetAuth::Override(irtt_client::Authentication::Hmac( + irtt_client::HmacKey::new(key.as_bytes().to_vec().into_boxed_slice()), + )); + } + target + }) + .collect::>(); + + self.client.update_targets(targets.clone())?.await?; + self.targets = targets; + + self.bindings = bindings + .into_iter() + .map(|binding| (binding.stream.id, binding)) + .collect(); + + Ok(()) + } + + pub async fn run(mut self, task: ManagedClientTask) -> Result<(), IrttError> { + let mut task = Box::pin(task); + let client = self.client.clone(); + + // Keep driving the client even while the event loop awaits command receipts. + let result = tokio::select! { + biased; + outcome = task.as_mut() => { + return match outcome.end_reason { + ManagedEndReason::StopRequested => Ok(()), + end_reason => Err(IrttError::ManagedFailure(end_reason)), + }; + } + result = self.run_events() => result, + }; + + // Every event-loop exit stops and joins the driver, including channel errors. + let stop = client.stop(); + let (outcome, _) = tokio::join!(task, stop); + result?; + match outcome.end_reason { + ManagedEndReason::StopRequested => Ok(()), + end_reason => Err(IrttError::ManagedFailure(end_reason)), + } + } + + async fn run_events(&mut self) -> Result<(), IrttError> { + let mut events = self.client.subscribe()?; + // Durable status catches failures even if their presentation events were lost. + // Resubmitting the full desired set restarts terminal targets only. + let mut retry = tokio::time::interval(std::time::Duration::from_secs(5)); + retry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + retry.tick().await; + loop { + tokio::select! { + bindings = self.binding_rx.recv_async() => { + let bindings = bindings + .map_err(|_| IrttError::BindingChannelClosed)?; + + self.set_bindings(bindings).await?; + } + + _ = retry.tick() => { + if SHUTDOWN.load(std::sync::atomic::Ordering::Relaxed) { + return Ok(()); + } + if self.client.status().targets.iter().any(|target| { + target.desired && target.lifecycle == ManagedTargetLifecycle::Terminal + }) { + self.client.update_targets(self.targets.clone())?.await?; + } + } + + event = events.recv() => { + let event = match event { + Ok(event) => event, + + Err(RecvError::Lagged(n)) => { + warn!("IRTT event receiver dropped {n} events"); + continue; + } + + Err(RecvError::Closed) => { + let status = self.client.status(); + + match status.lifecycle { + ManagedLifecycle::Completed => return Ok(()), + + ManagedLifecycle::Failed => { + if let Some(outcome) = &status.final_outcome { + return Err(IrttError::ManagedFailure( + outcome.end_reason.clone() + )); + } + + return Err(IrttError::ManagedAbandoned); + } + + _ => return Err(IrttError::ManagedAbandoned), + } + } + }; + + match event { + ManagedEvent::Started => { + debug!("IRTT managed client started"); + } + + ManagedEvent::TargetStateChanged { + target, + lifecycle, + } => { + debug!("IRTT target {target:?}: {lifecycle:?}"); + } + + ManagedEvent::Client { target, event } => { + let stream = target.id.as_str() + .parse::() + .ok() + .and_then(|id| { + self.bindings.get(&MeasurementStreamId(id)) + }) + .map(|binding| binding.stream); + + match event { + ClientEvent::SessionStarted(session) => { + debug!( + "IRTT session started for {}: {}", + target.id, + session.remote + ); + } + + ClientEvent::NoTestCompleted(outcome) => { + debug!( + "IRTT no-test completed for {}: {}", + target.id, + outcome.remote + ); + } + + ClientEvent::SessionClosed { .. } => { + debug!("IRTT session closed for {}", target.id); + } + + ClientEvent::EchoSent { seq, .. } => { + trace!("IRTT {} sent probe {seq}", target.id); + } + + ClientEvent::EchoReply { + sent_at, + received_at, + one_way, + rtt, + .. + } => { + let Some(stream) = stream else { + continue; + }; + + let observation = MeasurementObservation { + stream, + started_at: sent_at.mono, + observed_at: received_at.mono, + round_trip: i64::try_from(rtt.effective.as_nanos()) + .ok().map(SignedDuration::from_nanos), + one_way: one_way.and_then(map_one_way), + }; + + self.observation_tx + .send_async(MeasurementEvent::Observation(observation)) + .await + .map_err(|_| IrttError::EventChannelClosed)?; + } + + ClientEvent::EchoLoss { + sent_at, + timeout_at, + .. + } => { + let Some(stream) = stream else { + continue; + }; + + let loss = MeasurementLoss { + stream, + started_at: sent_at.mono, + detected_at: timeout_at, + }; + + self.observation_tx + .send_async(MeasurementEvent::Loss(loss)) + .await + .map_err(|_| IrttError::EventChannelClosed)?; + } + + ClientEvent::DuplicateReply { seq, .. } => { + trace!( + "IRTT {} duplicate reply {seq}", + target.id + ); + } + + ClientEvent::LateReply { seq, .. } => { + trace!( + "IRTT {} late reply {seq}", + target.id + ); + } + + ClientEvent::Warning { kind, message, .. } => { + debug!( + "IRTT {} warning ({kind:?}): {message}", + target.id + ); + } + } + } + + ManagedEvent::TargetFinished { outcome } => { + match &outcome.end_reason { + ManagedTargetEndReason::Failed(failure) => { + warn!( + "IRTT target {} failed: {failure:?}", + outcome.target.id + ); + } + + reason => { + debug!( + "IRTT target {} finished: {reason:?}", + outcome.target.id + ); + } + } + } + + ManagedEvent::Stopping => { + debug!("IRTT managed client stopping"); + } + + ManagedEvent::Completed { outcome } => { + debug!( + "IRTT managed client completed: {:?}", + outcome.end_reason + ); + return Ok(()); + } + + ManagedEvent::Failed { outcome } => { + return Err(IrttError::ManagedFailure( + outcome.end_reason.clone() + )); + } + + ManagedEvent::Abandoned => { + return Err(IrttError::ManagedAbandoned); + } + } + } + + _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => { + if SHUTDOWN.load(std::sync::atomic::Ordering::Relaxed) { + return Ok(()); + } + } + } + } + } +} + +fn managed_client_config( + interval: std::time::Duration, + #[allow(unused)] settings: &NetworkSettings, +) -> ManagedClientConfig { + ManagedClientConfig { + client: ClientConfig { + socket: irtt_client::SocketConfig { + #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))] + bind_to_device: settings.measurement_bind_device.clone(), + #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))] + mark: settings.measurement_mark, + #[cfg(target_os = "freebsd")] + fib: settings.measurement_fib, + ..Default::default() + }, + request: irtt_client::SessionRequest { + duration: None, + interval, + ..Default::default() + }, + open: OpenPolicy { + negotiation: NegotiationPolicy::Strict, + ..Default::default() + }, + ..Default::default() + }, + completion: ManagedCompletionPolicy::ExplicitStop, + ..Default::default() + } +} + +fn map_one_way(sample: irtt_client::OneWayDelaySample) -> Option { + let uplink = i64::try_from(sample.client_to_server?.as_nanos()).ok()?; + + let downlink = i64::try_from(sample.server_to_client?.as_nanos()).ok()?; + + Some(OneWayLatency { + uplink: SignedDuration::from_nanos(uplink), + downlink: SignedDuration::from_nanos(downlink), + clock: OneWayClock::Uncalibrated, + }) +} diff --git a/src/measurement/mod.rs b/src/measurement/mod.rs index 0ee60f2..836dd96 100644 --- a/src/measurement/mod.rs +++ b/src/measurement/mod.rs @@ -1,2 +1,4 @@ pub(crate) mod icmp; +#[cfg(feature = "irtt")] +pub(crate) mod irtt; pub(crate) mod model; diff --git a/src/measurement/model.rs b/src/measurement/model.rs index 12b7bf5..643afec 100644 --- a/src/measurement/model.rs +++ b/src/measurement/model.rs @@ -20,7 +20,7 @@ pub enum MeasurementSource { impl Display for MeasurementSource { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { - MeasurementSource::IcmpEcho => write!(f, "icmp"), + MeasurementSource::IcmpEcho => write!(f, "icmp-echo"), MeasurementSource::IcmpTimestamp => write!(f, "icmp-timestamps"), MeasurementSource::Irtt => write!(f, "irtt"), MeasurementSource::TcpSeqAck => write!(f, "tcp-seqack"), @@ -121,6 +121,9 @@ pub struct MeasurementObservation { /// Local monotonic time at which the result was observed. pub observed_at: Instant, + /// Measured round-trip latency when it differs from elapsed local time. + pub round_trip: Option, + /// Directional timing, when the measurement mechanism can provide it. pub one_way: Option, } @@ -131,7 +134,10 @@ impl MeasurementObservation { } pub fn rtt_ms(&self) -> f64 { - self.rtt().as_secs_f64() * 1000.0 + self.round_trip + .map(SignedDuration::as_secs_f64) + .unwrap_or_else(|| self.rtt().as_secs_f64()) + * 1000.0 } } diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index b099177..d35e205 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, @@ -491,8 +491,7 @@ impl Ratecontroller { self.state_dl.deltas = down_deltas; self.state_ul.deltas = up_deltas; - let minimum = - usize::from(self.settings.advanced_settings.num_reflectors).min(reflectors.len()); + let minimum = usize::from(self.settings.reflectors.num_reflectors).min(reflectors.len()); if reflectors.is_empty() || self.state_dl.deltas.len() < minimum || self.state_ul.deltas.len() < minimum @@ -515,7 +514,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 { @@ -536,7 +535,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() { @@ -570,7 +579,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([( @@ -596,7 +605,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 ); } @@ -604,10 +615,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 d348003..3cbbc89 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, } @@ -159,13 +167,11 @@ impl ReflectorSelector { // After 40 reselections, slow down to every 15 minutes if reselection_count > 40 { - selector_sleep_time = Duration::new( - self.settings.advanced_settings.peer_reselection_time * 60, - 0, - ); + selector_sleep_time = + 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(); @@ -175,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())]; @@ -208,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 { @@ -223,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.advanced_settings.num_reflectors; + let num_reflectors = self.settings.reflectors.num_reflectors; let candidate_pool_num = (2 * num_reflectors) as usize; candidates.truncate(candidate_pool_num); @@ -245,8 +260,12 @@ impl ReflectorSelector { } if (candidates.len() as u8) < num_reflectors { - num_reflectors = candidates.len() as u8; + info!( + "Only {} usable reflectors, selecting available candidates", + candidates.len() + ); } + let num_reflectors = num_reflectors.min(candidates.len() as u8); let mut new_peers = Vec::new(); for i in 0..num_reflectors { @@ -256,7 +275,7 @@ impl ReflectorSelector { self.metrics.send(Metric::Event { name: "reflector_selected", reason: "", - reflector: Some(peer), + reflector: Some(peer.ip()), tags: &[], }); } @@ -271,7 +290,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 { @@ -284,7 +305,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, @@ -384,7 +409,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)) @@ -405,7 +430,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 3539729..1aa1e5f 100644 --- a/src/settings/mod.rs +++ b/src/settings/mod.rs @@ -2,12 +2,10 @@ // // SPDX-License-Identifier: MPL-2.0 -use anyhow::{Context, bail}; use config::{Config, ConfigError, Environment}; use log::Level; use serde::{Deserialize, Deserializer, de}; use std::fmt::{self, Display}; -use std::net::IpAddr; use std::str::FromStr; use uci_derive::{UciConfig, UciSection}; @@ -48,7 +46,7 @@ where type Value = bool; fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - formatter.write_str("a boolean or a UCI-compatible boolean string") + formatter.write_str("a boolean, integer 0 or 1, or a UCI-compatible boolean string") } fn visit_bool(self, value: bool) -> Result @@ -58,6 +56,28 @@ where Ok(value) } + fn visit_i64(self, value: i64) -> Result + where + E: de::Error, + { + match value { + 0 => Ok(false), + 1 => Ok(true), + _ => Err(E::custom("boolean integer must be 0 or 1")), + } + } + + fn visit_u64(self, value: u64) -> Result + where + E: de::Error, + { + match value { + 0 => Ok(false), + 1 => Ok(true), + _ => Err(E::custom("boolean integer must be 0 or 1")), + } + } + fn visit_str(self, value: &str) -> Result where E: de::Error, @@ -79,28 +99,32 @@ where deserializer.deserialize_any(FlexiBoolVisitor) } -#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] -pub enum MeasurementType { - Icmp = 1, +#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Hash)] +#[serde(rename_all = "kebab-case")] +pub enum ReflectorProtocol { + IcmpEcho = 1, IcmpTimestamps, + Irtt, } -impl Display for MeasurementType { +impl Display for ReflectorProtocol { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { - MeasurementType::Icmp => write!(f, "icmp"), - MeasurementType::IcmpTimestamps => write!(f, "icmp-timestamps"), + ReflectorProtocol::IcmpEcho => write!(f, "icmp-echo"), + ReflectorProtocol::IcmpTimestamps => write!(f, "icmp-timestamps"), + ReflectorProtocol::Irtt => write!(f, "irtt"), } } } -impl FromStr for MeasurementType { +impl FromStr for ReflectorProtocol { type Err = ConfigError; fn from_str(s: &str) -> Result { match s.to_lowercase().as_str() { - "icmp" => Ok(MeasurementType::Icmp), - "icmp-timestamps" | "icmp_timestamps" => Ok(MeasurementType::IcmpTimestamps), + "icmp-echo" => Ok(ReflectorProtocol::IcmpEcho), + "icmp-timestamps" => Ok(ReflectorProtocol::IcmpTimestamps), + "irtt" => Ok(ReflectorProtocol::Irtt), _ => Err(ConfigError::Message(format!( "invalid measurement type: {s}" ))), @@ -128,12 +152,23 @@ impl FromStr for ObservabilityProtocol { } } -#[derive(Debug, Deserialize)] -struct ReflectorRecord { - reflector_ip: IpAddr, - ip_version: u8, +#[derive(Debug, Clone, Deserialize, UciSection)] +pub(crate) struct ReflectorDefinition { + pub(crate) host: String, + pub(crate) country: Option, + pub(crate) protocols: Vec, + pub(crate) irtt_port: Option, #[allow(dead_code)] - description: String, + pub(crate) irtt_hmac: Option, +} + +#[derive(Debug, Deserialize)] +struct ReflectorRow { + host: String, + country: Option, + protocols: String, + irtt_port: Option, + irtt_hmac: Option, } #[derive(Clone, Debug, Deserialize, UciConfig)] @@ -142,6 +177,10 @@ pub(crate) struct Settings { #[uci(section)] pub(crate) network: NetworkSettings, + #[serde(default)] + #[uci(section)] + pub(crate) reflectors: ReflectorSettings, + #[serde(default)] #[uci(section)] pub(crate) output: OutputSettings, @@ -160,55 +199,41 @@ impl Settings { let builder = Config::builder(); #[cfg(all(feature = "uci", unix))] - let builder = builder.add_source(UciSource::from_schema::().required(false)); + let builder = builder.add_source( + UciSource::from_schema::() + .with_repeated_schema(&[ + uci_schema::UciRepeatedSectionMapping { + uci_section: "reflector", + config_path: "reflectors.entries", + options: ::UCI_OPTIONS, + }, + uci_schema::UciRepeatedSectionMapping { + uci_section: "bootstrap_reflector", + config_path: "reflectors.bootstrap", + options: ::UCI_OPTIONS, + }, + ]) + .required(false), + ); #[cfg(feature = "toml")] let builder = builder.add_source( config::File::with_name("/etc/sqm-autorate-rust/config.toml").required(false), ); - let builder = builder.add_source(Environment::with_prefix("SQMA").separator("__")); - let built_config = builder.build()?; + let builder = builder.add_source( + Environment::with_prefix("SQMA") + .separator("__") + .try_parsing(true) + .list_separator(",") + .with_list_parse_key("reflectors.files") + .with_list_parse_key("reflectors.countries"), + ); + 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 reader = csv::ReaderBuilder::new() - .has_headers(true) - .trim(csv::Trim::All) - .from_path(&self.advanced_settings.reflector_list_file) - .with_context(|| { - format!( - "failed to open reflector list: {}", - self.advanced_settings.reflector_list_file - ) - })?; - - let mut reflectors = Vec::with_capacity(50); - - for record in reader.deserialize::() { - let record = record.with_context(|| { - format!( - "failed to parse reflector list: {}", - self.advanced_settings.reflector_list_file - ) - })?; - - match (record.ip_version, record.reflector_ip) { - (4, IpAddr::V4(_)) => {} - (6, address @ IpAddr::V6(_)) => { - bail!("reflector {address} is IPv6; only IPv4 reflectors are supported"); - } - (version, address) => { - bail!("reflector {address} has inconsistent IP version {version}"); - } - } - - reflectors.push(record.reflector_ip); - } - - Ok(reflectors) + Ok(settings) } } @@ -241,6 +266,100 @@ impl NetworkSettings { } } +#[derive(Clone, Debug, Deserialize, UciSection)] +#[serde(default)] +pub(crate) struct ReflectorSettings { + #[serde(deserialize_with = "deserialize_reflector_files")] + pub(crate) files: Vec, + pub(crate) countries: Vec, + + #[uci(skip)] + pub(crate) entries: Vec, + + #[uci(skip)] + pub(crate) bootstrap: Vec, + + pub(crate) num_reflectors: u8, + 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 = row + .protocols + .split(['|', ';']) + .map(str::trim) + .filter(|protocol| !protocol.is_empty()) + .map(ReflectorProtocol::from_str) + .collect::, _>>()?; + if protocols.is_empty() { + return Err(ConfigError::Message(format!( + "reflector row in {path} has no protocols" + ))); + } + 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![ + 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, + } + } +} + #[derive(Clone, Debug, Deserialize, UciSection)] #[serde(default)] pub(crate) struct OutputSettings { @@ -310,10 +429,7 @@ pub(crate) struct AdvancedSettings { pub(crate) high_load_level: f64, pub(crate) min_change_interval: f64, #[serde(deserialize_with = "deserialize_from_str")] - pub(crate) measurement_type: MeasurementType, - pub(crate) num_reflectors: u8, - pub(crate) peer_reselection_time: u64, - pub(crate) reflector_list_file: String, + pub(crate) measurement_type: ReflectorProtocol, pub(crate) speed_hist_size: u32, pub(crate) tick_interval: f64, pub(crate) upload_delay_ms: f64, @@ -327,10 +443,7 @@ impl Default for AdvancedSettings { download_delay_ms: 15.0, high_load_level: 0.8, min_change_interval: 0.5, - measurement_type: MeasurementType::IcmpTimestamps, - num_reflectors: 5, - peer_reselection_time: 15, - reflector_list_file: "/etc/sqm-autorate/reflectors-icmp.csv".to_owned(), + measurement_type: ReflectorProtocol::IcmpTimestamps, speed_hist_size: 100, tick_interval: 0.5, upload_delay_ms: 15.0, @@ -359,3 +472,24 @@ where let value = String::deserialize(deserializer)?; value.parse().map_err(de::Error::custom) } + +// UCI drops empty string options; `option files 'none'` disables file loading. +fn deserialize_reflector_files<'de, D>(deserializer: D) -> Result, D::Error> +where + D: Deserializer<'de>, +{ + #[derive(Deserialize)] + #[serde(untagged)] + enum Files { + List(Vec), + String(String), + } + + match Files::deserialize(deserializer)? { + Files::List(files) => Ok(files), + Files::String(value) if value.trim().is_empty() || value.trim() == "none" => Ok(Vec::new()), + Files::String(_) => Err(de::Error::custom( + "reflector files must be a list, an empty string, or none", + )), + } +} diff --git a/src/settings/validation.rs b/src/settings/validation.rs index 14a54cd..e6818a3 100644 --- a/src/settings/validation.rs +++ b/src/settings/validation.rs @@ -3,6 +3,8 @@ // SPDX-License-Identifier: MPL-2.0 // +use crate::settings::ReflectorSettings; + use super::{AdvancedSettings, NetworkSettings, ObservabilitySettings, OutputSettings, Settings}; use std::error::Error; use std::fmt::{self, Display, Formatter}; @@ -88,6 +90,7 @@ impl Settings { let mut report = ValidationReport::default(); self.network.validate(&mut report); + self.reflectors.validate(&mut report); self.output.validate(&mut report); self.observability.validate(&mut report); self.advanced_settings.validate(&mut report); @@ -192,6 +195,60 @@ impl NetworkSettings { } } +impl ReflectorSettings { + fn validate(&self, report: &mut ValidationReport) { + if self.num_reflectors == 0 { + report.error("reflectors.num_reflectors", "must be greater than 0"); + } + + if self.peer_reselection_time == 0 { + report.error("reflectors.peer_reselection_time", "must be greater than 0"); + } + + if self.files.is_empty() && self.entries.is_empty() { + report.error( + "reflectors", + "at least one reflector file or inline entry is required", + ); + } + + for (index, reflector) in self.entries.iter().enumerate() { + if reflector.host.trim().is_empty() { + report.error( + "reflectors.entries", + format!("entry {index} must have a nonempty host"), + ); + } + if reflector.protocols.is_empty() { + report.error( + "reflectors.entries", + format!("entry {index} must have at least one measurement protocol"), + ); + } + } + + for (index, file) in self.files.iter().enumerate() { + if file.trim().is_empty() { + report.error( + "reflectors.files", + format!("entry {index} must not be empty"), + ); + } + } + + for (index, country) in self.countries.iter().enumerate() { + if country.len() != 2 || !country.bytes().all(|c| c.is_ascii_alphabetic()) { + report.error( + "reflectors.countries", + format!( + "entry {index} ({country:?}) must be an ISO 3166-1 alpha-2 country code" + ), + ); + } + } + } +} + impl OutputSettings { fn validate(&self, report: &mut ValidationReport) { if self.suppress_statistics { @@ -298,24 +355,6 @@ impl AdvancedSettings { self.tick_interval, ); - if self.num_reflectors < 5 { - report.error( - "advanced_settings.num_reflectors", - format!("must be at least 5, got {}", self.num_reflectors,), - ); - } - - if self.peer_reselection_time == 0 { - report.error( - "advanced_settings.peer_reselection_time", - "must be greater than 0", - ); - } - - if self.reflector_list_file.trim().is_empty() { - report.error("advanced_settings.reflector_list_file", "must not be empty"); - } - if self.speed_hist_size == 0 { report.error( "advanced_settings.speed_hist_size",