diff --git a/src/baseliner.rs b/src/baseliner.rs index a2b00fa..1f05116 100644 --- a/src/baseliner.rs +++ b/src/baseliner.rs @@ -5,10 +5,10 @@ // // SPDX-License-Identifier: MPL-2.0 -use crate::Config; use crate::SHUTDOWN; use crate::metrics::{Metric, MetricsSender}; use crate::pinger::PingReply; +use crate::settings::Settings; use flume::{Receiver, Sender}; use log::{debug, info}; use std::collections::HashMap; @@ -70,7 +70,7 @@ pub struct ControlSnapshot { } pub struct Baseliner { - pub config: Config, + pub settings: Settings, pub reselect_trigger: Sender, pub start_time: Instant, pub stats_rx: Receiver, @@ -150,8 +150,14 @@ impl Baseliner { pub fn run(&self) -> anyhow::Result<()> { let mut reflectors = HashMap::::new(); - let slow_factor = ewma_factor(self.config.tick_interval, BASELINE_HALF_LIFE_SECS); - let fast_factor = ewma_factor(self.config.tick_interval, RECENT_HALF_LIFE_SECS); + let slow_factor = ewma_factor( + self.settings.advanced_settings.tick_interval, + BASELINE_HALF_LIFE_SECS, + ); + let fast_factor = ewma_factor( + self.settings.advanced_settings.tick_interval, + RECENT_HALF_LIFE_SECS, + ); loop { if SHUTDOWN.load(Ordering::Relaxed) { @@ -212,7 +218,7 @@ impl Baseliner { #[cfg(test)] mod tests { use super::*; - use crate::config::MeasurementType; + use crate::settings::MeasurementType; use std::time::Duration; fn state( diff --git a/src/config.rs b/src/config.rs deleted file mode 100644 index 21764ea..0000000 --- a/src/config.rs +++ /dev/null @@ -1,436 +0,0 @@ -// SPDX-FileCopyrightText: 2022-Present Charles Corrigan mailto:chas-iot@runegate.org (github @chas-iot) -// SPDX-FileCopyrightText: 2022-Present Daniel Lakeland mailto:dlakelan@street-artists.org (github @dlakelan) -// SPDX-FileCopyrightText: 2022-Present Mark Baker mailto:mark@vpost.net (github @Fail-Safe) -// SPDX-FileCopyrightText: 2022-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair) -// -// SPDX-License-Identifier: MPL-2.0 - -use anyhow::Result; -use log::Level; -#[cfg(all(target_os = "linux", feature = "uci"))] -use log::warn; -#[cfg(all(target_os = "linux", feature = "uci"))] -use rust_uci::Uci; -use std::fmt; -use std::fmt::Display; -use std::fs::File; -use std::io::BufRead; -use std::net::IpAddr; -use std::path::Path; -use std::str::FromStr; -use std::{env, io}; - -use thiserror::Error; - -#[derive(Error, Debug)] -pub enum ConfigError { - #[error("Invalid measurement type")] - InvalidMeasurementType(String), - #[error("Couldn't parse value for key: `{0}`: invalid value")] - ParseError(String), - #[error("No config value found for key: `{0}`")] - MissingValue(String), - #[error("Reflector list not found at: {0}")] - ReflectorListNotFound(String), -} - -fn read_lines

(filename: P) -> io::Result>> -where - P: AsRef, -{ - let file = File::open(filename)?; - Ok(io::BufReader::new(file).lines()) -} - -struct FlexiBool(bool); - -impl FromStr for FlexiBool { - type Err = ConfigError; - - fn from_str(s: &str) -> Result { - match s.trim().to_lowercase().as_str() { - "1" | "true" | "yes" | "y" | "on" | "enabled" => Ok(FlexiBool(true)), - "0" | "false" | "no" | "n" | "off" | "disabled" => Ok(FlexiBool(false)), - _ => Err(ConfigError::ParseError(s.to_string())), - } - } -} - -impl From for bool { - fn from(f: FlexiBool) -> bool { - f.0 - } -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] -pub enum MeasurementType { - Icmp = 1, - IcmpTimestamps, -} - -impl Display for MeasurementType { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - MeasurementType::Icmp => write!(f, "icmp"), - MeasurementType::IcmpTimestamps => write!(f, "icmp-timestamps"), - } - } -} - -impl FromStr for MeasurementType { - type Err = ConfigError; - - fn from_str(s: &str) -> Result { - return match s.to_lowercase().as_str() { - "icmp" => Ok(MeasurementType::Icmp), - "icmp-timestamps" => Ok(MeasurementType::IcmpTimestamps), - &_ => Err(ConfigError::InvalidMeasurementType(s.to_string())), - }; - } -} - -#[derive(Clone, Copy, Debug)] -pub enum ObservabilityProtocol { - Udp, - Tcp, -} - -impl FromStr for ObservabilityProtocol { - type Err = ConfigError; - - fn from_str(s: &str) -> Result { - match s.to_lowercase().as_str() { - "udp" => Ok(ObservabilityProtocol::Udp), - "tcp" => Ok(ObservabilityProtocol::Tcp), - _ => Err(ConfigError::InvalidMeasurementType(s.to_string())), - } - } -} - -#[derive(Clone, Debug)] -pub struct Config { - // Network section - pub download_interface: String, - pub upload_interface: String, - pub download_base_kbits: f64, - pub download_min_percent: f64, - pub download_min_kbits: f64, - pub upload_base_kbits: f64, - pub upload_min_percent: f64, - pub upload_min_kbits: f64, - - // Output section - pub log_level: Level, - pub speed_hist_file: String, - pub stats_file: String, - pub suppress_statistics: bool, - - // Observability section - pub observability_enabled: bool, - pub observability_protocol: ObservabilityProtocol, - pub observability_host: Option, - pub observability_port: u16, - pub observability_batch_size: usize, - pub observability_batch_timeout_ms: u64, - pub observability_export_ping_metrics: bool, - pub observability_export_rate_metrics: bool, - pub observability_export_baseline_metrics: bool, - pub observability_export_events: bool, - pub observability_host_tag: String, - - // Advanced section - pub download_delay_ms: f64, - pub high_load_level: f64, - pub min_change_interval: f64, - pub measurement_type: MeasurementType, - pub num_reflectors: u8, - pub peer_reselection_time: u64, - pub reflector_list_file: String, - pub speed_hist_size: u32, - pub tick_interval: f64, - pub upload_delay_ms: f64, - - // Monitoring / dry-run - pub dry_run: bool, -} - -impl Config { - pub fn new() -> Result { - let config = Self { - // Network section - download_base_kbits: Self::get::( - "SQMA_DOWNLOAD_BASE_KBITS", - "sqm-autorate-rust.@network[0].download_base_kbits", - None, - )?, - download_interface: Self::get::( - "SQMA_DOWNLOAD_INTERFACE", - "sqm-autorate-rust.@network[0].download_interface", - None, - )?, - download_min_percent: 0.0, // placeholder, computed below - download_min_kbits: 0.0, // placeholder, computed below - upload_base_kbits: Self::get::( - "SQMA_UPLOAD_BASE_KBITS", - "sqm-autorate-rust.@network[0].upload_base_kbits", - None, - )?, - upload_interface: Self::get::( - "SQMA_UPLOAD_INTERFACE", - "sqm-autorate-rust.@network[0].upload_interface", - None, - )?, - upload_min_percent: 0.0, // placeholder, computed below - upload_min_kbits: 0.0, // placeholder, computed below - // Output section - log_level: Self::get::( - "SQMA_LOG_LEVEL", - "sqm-autorate-rust.@output[0].log_level", - Some(Level::Error), - )?, - speed_hist_file: Self::get::( - "SQMA_SPEED_HIST_FILE", - "sqm-autorate-rust.@output[0].speed_hist_file", - Some("/tmp/sqm-speedhist.csv".parse()?), - )?, - stats_file: Self::get::( - "SQMA_STATS_FILE", - "sqm-autorate-rust.@output[0].stats_file", - Some("/tmp/sqm-autorate-rust.csv".parse()?), - )?, - suppress_statistics: Self::get_bool( - "SQMA_SUPPRESS_STATISTICS", - "sqm-autorate-rust.@output[0].suppress_statistics", - Some(false), - )?, - - // Observability section - observability_enabled: Self::get_bool( - "SQMA_OBSERVABILITY_ENABLED", - "sqm-autorate-rust.@observability[0].enabled", - Some(false), - )?, - observability_protocol: Self::get::( - "SQMA_OBSERVABILITY_PROTOCOL", - "sqm-autorate-rust.@observability[0].protocol", - Some(ObservabilityProtocol::Udp), - )?, - observability_host: Self::get_optional::( - "SQMA_OBSERVABILITY_HOST", - "sqm-autorate-rust.@observability[0].host", - ), - observability_port: Self::get::( - "SQMA_OBSERVABILITY_PORT", - "sqm-autorate-rust.@observability[0].port", - Some(8089), - )?, - observability_batch_size: Self::get::( - "SQMA_OBSERVABILITY_BATCH_SIZE", - "sqm-autorate-rust.@observability[0].batch_size", - Some(25), - )?, - observability_batch_timeout_ms: Self::get::( - "SQMA_OBSERVABILITY_BATCH_TIMEOUT_MS", - "sqm-autorate-rust.@observability[0].batch_timeout_ms", - Some(100), - )?, - observability_export_ping_metrics: Self::get_bool( - "SQMA_OBSERVABILITY_EXPORT_PING_METRICS", - "sqm-autorate-rust.@observability[0].export_ping_metrics", - Some(false), - )?, - observability_export_rate_metrics: Self::get_bool( - "SQMA_OBSERVABILITY_EXPORT_RATE_METRICS", - "sqm-autorate-rust.@observability[0].export_rate_metrics", - Some(true), - )?, - observability_export_baseline_metrics: Self::get_bool( - "SQMA_OBSERVABILITY_EXPORT_BASELINE_METRICS", - "sqm-autorate-rust.@observability[0].export_baseline_metrics", - Some(false), - )?, - observability_export_events: Self::get_bool( - "SQMA_OBSERVABILITY_EXPORT_EVENTS", - "sqm-autorate-rust.@observability[0].export_events", - Some(true), - )?, - observability_host_tag: Self::get::( - "SQMA_OBSERVABILITY_HOST_TAG", - "sqm-autorate-rust.@observability[0].host_tag", - Some(Self::get_host_tag()), - )?, - - // Advanced section - download_delay_ms: Self::get::( - "SQMA_DOWNLOAD_DELAY_MS", - "sqm-autorate-rust.@advanced_settings[0].download_delay_ms", - Some(15.0), - )?, - high_load_level: Self::get::( - "SQMA_HIGH_LOAD_LEVEL", - "sqm-autorate-rust.@advanced_settings[0].high_load_level", - Some(0.8), - )?, - measurement_type: Self::get::( - "SQMA_MEASUREMENT_TYPE", - "sqm-autorate-rust.@advanced_settings[0].measurement_type", - Some(MeasurementType::IcmpTimestamps), - )?, - min_change_interval: Self::get::( - "SQMA_MIN_CHANGE_INTERVAL", - "sqm-autorate-rust.@advanced_settings[0].min_change_interval", - Some(0.5), - )?, - num_reflectors: Self::get::( - "SQMA_NUM_REFLECTORS", - "sqm-autorate-rust.@advanced_settings[0].num_reflectors", - Some(5), - )?, - peer_reselection_time: Self::get::( - "SQMA_PEER_RESELECTION_TIME", - "sqm-autorate-rust.@advanced_settings[0].peer_reselection_time", - Some(15), - )?, - reflector_list_file: Self::get::( - "SQMA_REFLECTOR_LIST_FILE", - "sqm-autorate-rust.@advanced_settings[0].reflector_list_file", - Some("/etc/sqm-autorate/reflectors-icmp.csv".parse()?), - )?, - speed_hist_size: Self::get::( - "SQMA_SPEED_HIST_SIZE", - "sqm-autorate-rust.@advanced_settings[0].speed_hist_size", - Some(100), - )?, - tick_interval: Self::get::( - "SQMA_TICK_INTERVAL", - "sqm-autorate-rust.@advanced_settings[0].tick_interval", - Some(0.5), - )?, - upload_delay_ms: Self::get::( - "SQMA_UPLOAD_DELAY_MS", - "sqm-autorate-rust.@advanced_settings[0].upload_delay_ms", - Some(15.0), - )?, - dry_run: Self::get_bool( - "SQMA_DRY_RUN", - "sqm-autorate-rust.@advanced_settings[0].dry_run", - Some(false), - )?, - }; - - let mut config = config; - - config.download_min_percent = Self::get::( - "SQMA_DOWNLOAD_MIN_PERCENT", - "sqm-autorate-rust.@network[0].download_min_percent", - Some(20.0), - )? - .clamp(1.0, 80.0); - - config.upload_min_percent = Self::get::( - "SQMA_UPLOAD_MIN_PERCENT", - "sqm-autorate-rust.@network[0].upload_min_percent", - Some(20.0), - )? - .clamp(1.0, 80.0); - - config.download_min_kbits = - (config.download_base_kbits * config.download_min_percent / 100.0).floor(); - config.upload_min_kbits = - (config.upload_base_kbits * config.upload_min_percent / 100.0).floor(); - - Ok(config) - } - - fn get(env_key: &str, uci_key: &str, default: Option) -> Result { - match Self::get_value(env_key, uci_key) { - Some(val) => match val.parse::() { - Ok(parsed_val) => Ok(parsed_val), - // Ran into an compilation error while trying to return the - // error as-is, so using my own error type to indicate something went wrong while parsing - Err(_) => Err(ConfigError::ParseError(env_key.to_string())), - }, - None => match default { - Some(val) => Ok(val), - None => Err(ConfigError::MissingValue(env_key.to_string())), - }, - } - } - - fn get_bool(env_key: &str, uci_key: &str, default: Option) -> Result { - Self::get::(env_key, uci_key, default.map(FlexiBool)).map(|f| f.0) - } - - fn get_host_tag() -> String { - // rustix can give us the hostname without shelling out - rustix::system::uname() - .nodename() - .to_string_lossy() - .into_owned() - } - - fn get_optional(env_key: &str, uci_key: &str) -> Option { - Self::get_value(env_key, uci_key).and_then(|val| val.parse::().ok()) - } - - fn get_value(env_key: &str, uci_key: &str) -> Option { - if let Ok(val) = env::var(env_key) { - return Some(val); - } - - if let Some(val) = Self::get_from_uci(uci_key) { - return Some(val); - } - - None - } - - #[cfg(all(target_os = "linux", feature = "uci"))] - fn get_from_uci(key: &str) -> Option { - let mut uci = match Uci::new() { - Ok(val) => val, - Err(e) => { - warn!("Error opening UCI instance: {}", e); - return None; - } - }; - - return match uci.get(key) { - Ok(val) => Some(val), - Err(e) => { - warn!("Problem getting config from UCI: {}", e); - None - } - }; - } - - #[cfg(not(all(target_os = "linux", feature = "uci")))] - fn get_from_uci(_: &str) -> Option { - None - } - - pub fn load_reflectors(&self) -> Result> { - let lines = read_lines(self.reflector_list_file.clone()).map_err(|e| { - ConfigError::ReflectorListNotFound( - self.reflector_list_file.clone() + ": " + e.to_string().as_str(), - ) - })?; - - let mut reflectors: Vec = Vec::with_capacity(50); - - let mut first = true; - - for line in lines { - if first { - first = false; - continue; - } - - let line = line?; - let columns: Vec<&str> = line.split(',').collect(); - reflectors.push(IpAddr::from_str(columns[0])?); - } - - Ok(reflectors) - } -} diff --git a/src/main.rs b/src/main.rs index 88b73fd..47b4a1f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -8,7 +8,6 @@ extern crate core; mod baseliner; -mod config; mod log; mod metrics; mod pinger; @@ -23,6 +22,14 @@ mod util; use crate::baseliner::Baseliner; use crate::metrics::{Metric, Metrics, MetricsSender}; +use crate::pinger::{InFlightProbeCache, PingListener, PingSender}; +use crate::pinger_icmp::{PingerICMPEchoListener, PingerICMPEchoSender}; +use crate::pinger_icmp_ts::{PingerICMPTimestampListener, PingerICMPTimestampSender}; +use crate::platform::{TrafficControlBackend, traffic_control_backend, warn_platform_limitations}; +use crate::ratecontroller::{Ratecontroller, StatsDirection}; +use crate::reflector_selector::ReflectorSelector; +use crate::settings::MeasurementType; +use crate::settings::Settings; use ::log::{debug, info, warn}; use flume::RecvTimeoutError; use std::collections::HashMap; @@ -34,14 +41,6 @@ use std::thread::sleep; use std::time::{Duration, Instant}; use std::{process, thread}; -use crate::config::{Config, MeasurementType}; -use crate::pinger::{InFlightProbeCache, PingListener, PingSender}; -use crate::pinger_icmp::{PingerICMPEchoListener, PingerICMPEchoSender}; -use crate::pinger_icmp_ts::{PingerICMPTimestampListener, PingerICMPTimestampSender}; -use crate::platform::{TrafficControlBackend, traffic_control_backend, warn_platform_limitations}; -use crate::ratecontroller::{Ratecontroller, StatsDirection}; -use crate::reflector_selector::ReflectorSelector; - struct ReflectorSetup { peers: Arc>>, pool: Vec, @@ -64,12 +63,14 @@ const INFLIGHT_CAPACITY_DUPLICATE_FACTOR: usize = 2; const INFLIGHT_CAPACITY_MIN: usize = 256; fn compute_inflight_probe_capacity( - config: &Config, + settings: &Settings, active_reflector_count: usize, reselection_enabled: bool, ) -> usize { - let tick_interval_ms = (config.tick_interval * 1000.0).max(1.0); - let expected_path_delay_ms = (config.download_delay_ms + config.upload_delay_ms).max(10.0); + let tick_interval_ms = (settings.advanced_settings.tick_interval * 1000.0).max(1.0); + let expected_path_delay_ms = (settings.advanced_settings.download_delay_ms + + settings.advanced_settings.upload_delay_ms) + .max(10.0); // Keep enough room for severe queueing events while still adapting to configured delay budgets. let max_rtt_ms = (expected_path_delay_ms * 20.0).clamp(1000.0, 10000.0); @@ -100,27 +101,36 @@ fn create_pinger(measurement_type: MeasurementType) -> (PingListenerBox, PingSen } fn initialize_shaper( - config: &Config, + settings: &Settings, traffic_control: &mut T, ) -> anyhow::Result<(T::Handle, T::Handle)> { - initialize_shaper_with_settle_time(config, traffic_control, Duration::from_secs(2)) + initialize_shaper_with_settle_time(settings, traffic_control, Duration::from_secs(2)) } fn initialize_shaper_with_settle_time( - config: &Config, + settings: &Settings, traffic_control: &mut T, settle_time: Duration, ) -> anyhow::Result<(T::Handle, T::Handle)> { - let down = traffic_control.find_shaper(&config.download_interface)?; - let up = traffic_control.find_shaper(&config.upload_interface)?; + let down = traffic_control.find_shaper(&settings.network.download_interface)?; + let up = traffic_control.find_shaper(&settings.network.upload_interface)?; info!( "Requesting minimum shaper rates (D/U): {} / {}", - config.download_min_kbits, config.upload_min_kbits, + settings.network.download_min_kbits(), + settings.network.upload_min_kbits(), ); - traffic_control.set_rate(&down, config.download_min_kbits as u64, config.dry_run)?; - traffic_control.set_rate(&up, config.upload_min_kbits as u64, config.dry_run)?; + traffic_control.set_rate( + &down, + settings.network.download_min_kbits() as u64, + settings.advanced_settings.dry_run, + )?; + traffic_control.set_rate( + &up, + settings.network.upload_min_kbits() as u64, + settings.advanced_settings.dry_run, + )?; info!( "Sleeping for {} seconds to give the shaper a chance to control existing bloat", @@ -145,17 +155,17 @@ fn install_signal_handlers() { } } -fn interface_stats_directions(config: &Config) -> (StatsDirection, StatsDirection) { - let download = if config.download_interface.starts_with("ifb") - || config.download_interface.starts_with("veth") +fn interface_stats_directions(settings: &Settings) -> (StatsDirection, StatsDirection) { + let download = if settings.network.download_interface.starts_with("ifb") + || settings.network.download_interface.starts_with("veth") { StatsDirection::TX } else { StatsDirection::RX }; - let upload = if config.upload_interface.starts_with("ifb") - || config.upload_interface.starts_with("veth") + let upload = if settings.network.upload_interface.starts_with("ifb") + || settings.network.upload_interface.starts_with("veth") { StatsDirection::RX } else { @@ -171,31 +181,35 @@ fn probe_identifier() -> u16 { } fn restore_shaper( - config: &Config, + settings: &Settings, traffic_control: &mut T, down: &T::Handle, up: &T::Handle, ) { info!( "Requesting restoration to base shaper rates (D/U): {} / {}", - config.download_base_kbits, config.upload_base_kbits, + settings.network.download_base_kbits, settings.network.upload_base_kbits, ); - if let Err(error) = - traffic_control.set_rate(down, config.download_base_kbits as u64, config.dry_run) - { + if let Err(error) = traffic_control.set_rate( + down, + settings.network.download_base_kbits as u64, + settings.advanced_settings.dry_run, + ) { warn!("Failed to restore download shaper rate: {error}"); } - if let Err(error) = - traffic_control.set_rate(up, config.upload_base_kbits as u64, config.dry_run) - { + if let Err(error) = traffic_control.set_rate( + up, + settings.network.upload_base_kbits as u64, + settings.advanced_settings.dry_run, + ) { warn!("Failed to restore upload shaper rate: {error}"); } } -fn setup_reflectors(config: &Config) -> anyhow::Result { - let configured = config.load_reflectors()?; +fn setup_reflectors(settings: &Settings) -> anyhow::Result { + let configured = settings.load_reflectors()?; let configured_count = configured.len(); let default_reflectors = [ @@ -207,7 +221,7 @@ fn setup_reflectors(config: &Config) -> anyhow::Result { IpAddr::from_str("94.140.14.14")?, ]; - let reselection_enabled = configured_count > config.num_reflectors as usize; + let reselection_enabled = configured_count > settings.advanced_settings.num_reflectors as usize; let pool = if reselection_enabled { configured } else { @@ -218,7 +232,9 @@ fn setup_reflectors(config: &Config) -> anyhow::Result { peers: Arc::new(RwLock::new(default_reflectors.to_vec())), pool, reselection_enabled, - active_count: default_reflectors.len().max(config.num_reflectors as usize), + active_count: default_reflectors + .len() + .max(settings.advanced_settings.num_reflectors as usize), }) } @@ -241,8 +257,8 @@ fn wait_for_exit(error_rx: &flume::Receiver) -> anyhow::Result<() } } -fn run(config: Config) -> anyhow::Result<()> { - if config.dry_run { +fn run(settings: &Settings) -> anyhow::Result<()> { + if settings.advanced_settings.dry_run { info!("*** MONITORING MODE ACTIVE — qdisc rates will NOT be changed ***"); } @@ -254,7 +270,7 @@ fn run(config: Config) -> anyhow::Result<()> { pool: reflector_pool, reselection_enabled, active_count: active_reflector_count, - } = setup_reflectors(&config)?; + } = setup_reflectors(settings)?; let (baseliner_stats_tx, baseliner_stats_rx) = flume::unbounded(); let (control_snapshot_tx, control_snapshot_rx) = flume::unbounded(); @@ -269,10 +285,10 @@ fn run(config: Config) -> anyhow::Result<()> { let dropped = Arc::new(AtomicU32::new(0)); - let (metrics_tx, metrics_thread_handle) = if config.observability_enabled { + let (metrics_tx, metrics_thread_handle) = if settings.observability.enabled { let (tx, rx) = flume::bounded(1000); let metrics = Metrics { - config: config.clone(), + settings: settings.clone(), metrics_rx: rx, metrics_dropped: Arc::clone(&dropped), }; @@ -297,36 +313,40 @@ fn run(config: Config) -> anyhow::Result<()> { .unwrap_or_else(MetricsSender::disabled) }; - let ping_metrics = make_sender(config.observability_export_ping_metrics); - let baseline_metrics = make_sender(config.observability_export_baseline_metrics); - let event_metrics = make_sender(config.observability_export_events); - let rate_metrics = make_sender(config.observability_export_rate_metrics); + let ping_metrics = make_sender(settings.observability.export_ping_metrics); + let baseline_metrics = make_sender(settings.observability.export_baseline_metrics); + let event_metrics = make_sender(settings.observability.export_events); + let rate_metrics = make_sender(settings.observability.export_rate_metrics); event_metrics.send(Metric::Event { name: "starting", reason: "", reflector: None, - tags: if config.dry_run { + tags: if settings.advanced_settings.dry_run { &[("dry_run", "true")] } else { &[] }, }); - let (mut ping_listener, mut ping_sender) = create_pinger(config.measurement_type); + let (mut ping_listener, mut ping_sender) = + create_pinger(settings.advanced_settings.measurement_type); let inflight_probe_capacity = - compute_inflight_probe_capacity(&config, active_reflector_count, reselection_enabled); + compute_inflight_probe_capacity(settings, active_reflector_count, reselection_enabled); info!( "In-flight probe cache capacity: {} (active_reflectors={}, reselection_enabled={}, tick_interval_s={})", - inflight_probe_capacity, active_reflector_count, reselection_enabled, config.tick_interval + inflight_probe_capacity, + active_reflector_count, + reselection_enabled, + settings.advanced_settings.tick_interval ); let inflight: InFlightProbeCache = Arc::new(Mutex::new(HashMap::with_capacity(inflight_probe_capacity))); let baseliner = Baseliner { - config: config.clone(), + settings: settings.clone(), reselect_trigger: reselect_tx.clone(), start_time, stats_rx: baseliner_stats_rx, @@ -337,12 +357,12 @@ fn run(config: Config) -> anyhow::Result<()> { }; let mut main_traffic_control = traffic_control_backend(); - let (down_shaper, up_shaper) = initialize_shaper(&config, &mut main_traffic_control)?; + let (down_shaper, up_shaper) = initialize_shaper(settings, &mut main_traffic_control)?; let err_tx = error_tx.clone(); let listener_peers = Arc::clone(&reflector_peers); let listener_inflight = Arc::clone(&inflight); - let measurement_type = config.measurement_type; + let measurement_type = settings.advanced_settings.measurement_type; thread::Builder::new() .name("receiver".to_string()) .spawn(move || { @@ -370,8 +390,8 @@ fn run(config: Config) -> anyhow::Result<()> { let err_tx = error_tx.clone(); let sender_peers = Arc::clone(&reflector_peers); let sender_inflight = Arc::clone(&inflight); - let measurement_type = config.measurement_type; - let tick_interval = config.tick_interval; + let measurement_type = settings.advanced_settings.measurement_type; + let tick_interval = settings.advanced_settings.tick_interval; thread::Builder::new() .name("sender".to_string()) .spawn(move || { @@ -390,7 +410,7 @@ fn run(config: Config) -> anyhow::Result<()> { if reselection_enabled { let reflector_selector = ReflectorSelector { - config: config.clone(), + settings: settings.clone(), snapshot_rx: selection_snapshot_rx .expect("reselection snapshot receiver must exist when reselection is enabled"), reflector_peers_lock: Arc::clone(&reflector_peers), @@ -411,19 +431,19 @@ fn run(config: Config) -> anyhow::Result<()> { // Give the baseliner time to collect initial samples before adjusting rates. sleep(Duration::from_secs(10)); - let (dl_direction, ul_direction) = interface_stats_directions(&config); + let (dl_direction, ul_direction) = interface_stats_directions(settings); debug!( "Download direction: {}:{:?}", - config.download_interface, dl_direction + settings.network.download_interface, dl_direction ); debug!( "Upload direction: {}:{:?}", - config.upload_interface, ul_direction + settings.network.upload_interface, ul_direction ); let mut ratecontroller = Ratecontroller::new( - config.clone(), + settings.clone(), control_snapshot_rx, reflector_peers, reselect_tx, @@ -466,7 +486,12 @@ fn run(config: Config) -> anyhow::Result<()> { let _ = handle.join(); } - restore_shaper(&config, &mut main_traffic_control, &down_shaper, &up_shaper); + restore_shaper( + settings, + &mut main_traffic_control, + &down_shaper, + &up_shaper, + ); result } @@ -476,9 +501,11 @@ fn main() -> anyhow::Result<()> { install_signal_handlers(); - let config = Config::new()?; - log::init(config.log_level)?; + let settings = Settings::load()?; + settings.validate()?; + + log::init(settings.output.log_level)?; warn_platform_limitations(); - run(config) + run(&settings) } diff --git a/src/metrics.rs b/src/metrics.rs index 35d8762..e8d59ac 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -2,8 +2,8 @@ // // SPDX-License-Identifier: MPL-2.0 -use crate::Config; -use crate::config::{MeasurementType, ObservabilityProtocol}; +use crate::settings::Settings; +use crate::settings::{MeasurementType, ObservabilityProtocol}; use crate::time::Time; use flume::{Receiver, RecvTimeoutError, Sender}; use log::{error, info, warn}; @@ -157,14 +157,14 @@ impl MetricsSender { } pub struct Metrics { - pub config: Config, + pub settings: Settings, pub metrics_dropped: Arc, pub metrics_rx: Receiver<(Metric, u64)>, } impl Metrics { pub fn run(self) -> anyhow::Result<()> { - let host = match &self.config.observability_host { + let host = match &self.settings.observability.host { Some(h) => h.clone(), None => { error!("Observability host not configured - metrics exporter disabled"); @@ -172,12 +172,12 @@ impl Metrics { } }; - let port = self.config.observability_port; - let batch_size = self.config.observability_batch_size; - let timeout = Duration::from_millis(self.config.observability_batch_timeout_ms); - let host_tag = &self.config.observability_host_tag.clone(); + let port = self.settings.observability.port; + let batch_size = self.settings.observability.batch_size; + let timeout = Duration::from_millis(self.settings.observability.batch_timeout_ms); + let host_tag = &self.settings.observability.host_tag.clone(); - let mut transport = match self.config.observability_protocol { + let mut transport = match self.settings.observability.protocol { ObservabilityProtocol::Udp => { info!("Metrics exporter configured for UDP to {}:{}", host, port); Transport::new_udp(&host, port)? diff --git a/src/pinger_icmp.rs b/src/pinger_icmp.rs index 3a36144..91f4b5a 100644 --- a/src/pinger_icmp.rs +++ b/src/pinger_icmp.rs @@ -8,8 +8,8 @@ use std::net::IpAddr; use std::time::Instant; -use crate::config::MeasurementType; use crate::pinger::{PingError, PingListener, PingReply, PingSender}; +use crate::settings::MeasurementType; use crate::time::Time; use icmp_socket2::Icmpv4Message; use icmp_socket2::Icmpv4Packet; diff --git a/src/pinger_icmp_ts.rs b/src/pinger_icmp_ts.rs index de8c815..c34cbf1 100644 --- a/src/pinger_icmp_ts.rs +++ b/src/pinger_icmp_ts.rs @@ -5,8 +5,8 @@ // // SPDX-License-Identifier: MPL-2.0 -use crate::config::MeasurementType; use crate::pinger::{PingError, PingListener, PingReply, PingSender}; +use crate::settings::MeasurementType; use crate::time::Time; use icmp_socket2::Icmpv4Message; use icmp_socket2::Icmpv4Packet; diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index 8b0bf49..60b49ac 100644 --- a/src/ratecontroller.rs +++ b/src/ratecontroller.rs @@ -5,7 +5,6 @@ // // SPDX-License-Identifier: MPL-2.0 -use crate::Config; use crate::SHUTDOWN; use crate::baseliner::ControlSnapshot; use crate::metrics::{Metric, MetricsSender}; @@ -13,6 +12,7 @@ use crate::platform::{ InterfaceStatsProvider, PlatformInterfaceStats, PlatformTrafficControl, TrafficControlBackend, interface_stats_provider, traffic_control_backend, }; +use crate::settings::Settings; use crate::time::Time; use crate::util::{ArcRwLock, RwLockExt}; use flume::{Receiver, Sender}; @@ -157,7 +157,7 @@ fn deltas_from_snapshot( } pub struct Ratecontroller { - config: Config, + settings: Settings, control_snapshot: Option, control_snapshot_rx: Receiver, down_direction: StatsDirection, @@ -173,7 +173,7 @@ pub struct Ratecontroller { impl Ratecontroller { fn new_with_backends( - config: Config, + settings: Settings, control_snapshot_rx: Receiver, reflectors_lock: ArcRwLock>, reselect_trigger: Sender, @@ -183,23 +183,28 @@ impl Ratecontroller { mut stats_provider: S, mut traffic_control: T, ) -> anyhow::Result { - let dl_shaper = traffic_control.find_shaper(config.download_interface.as_str())?; - let dl_safe_rates = - generate_initial_speeds(config.download_base_kbits, config.speed_hist_size); - let ul_shaper = traffic_control.find_shaper(config.upload_interface.as_str())?; - let ul_safe_rates = - generate_initial_speeds(config.upload_base_kbits, config.speed_hist_size); + let dl_shaper = + traffic_control.find_shaper(settings.network.download_interface.as_str())?; + let dl_safe_rates = generate_initial_speeds( + settings.network.download_base_kbits, + settings.advanced_settings.speed_hist_size, + ); + let ul_shaper = traffic_control.find_shaper(settings.network.upload_interface.as_str())?; + let ul_safe_rates = generate_initial_speeds( + settings.network.upload_base_kbits, + settings.advanced_settings.speed_hist_size, + ); let (cur_rx, cur_tx) = get_interface_stats( &mut stats_provider, - &config.download_interface, - &config.upload_interface, + &settings.network.download_interface, + &settings.network.upload_interface, down_direction, up_direction, )?; Ok(Self { - config, + settings, control_snapshot: None, control_snapshot_rx, down_direction, @@ -216,18 +221,18 @@ impl Ratecontroller { fn request_initial_rates(&mut self) -> anyhow::Result<()> { // Set rates to 60% of base rate to make sure we start with sane baselines. - self.state_dl.current_rate = self.config.download_base_kbits * 0.6; - self.state_ul.current_rate = self.config.upload_base_kbits * 0.6; + self.state_dl.current_rate = self.settings.network.download_base_kbits * 0.6; + self.state_ul.current_rate = self.settings.network.upload_base_kbits * 0.6; self.traffic_control.set_rate( &self.state_dl.shaper, self.state_dl.current_rate.round() as u64, - self.config.dry_run, + self.settings.advanced_settings.dry_run, )?; self.traffic_control.set_rate( &self.state_ul.shaper, self.state_ul.current_rate.round() as u64, - self.config.dry_run, + self.settings.advanced_settings.dry_run, )?; Ok(()) @@ -236,16 +241,16 @@ impl Ratecontroller { fn calculate_rate(&mut self, direction: Direction) -> anyhow::Result<()> { let (base_rate, delay_ms, min_rate, state) = if direction == Direction::Down { ( - self.config.download_base_kbits, - self.config.download_delay_ms, - self.config.download_min_kbits, + self.settings.network.download_base_kbits, + self.settings.advanced_settings.download_delay_ms, + self.settings.network.download_min_kbits(), &mut self.state_dl, ) } else { ( - self.config.upload_base_kbits, - self.config.upload_delay_ms, - self.config.upload_min_kbits, + self.settings.network.upload_base_kbits, + self.settings.advanced_settings.upload_delay_ms, + self.settings.network.upload_min_kbits(), &mut self.state_ul, ) }; @@ -273,7 +278,9 @@ impl Ratecontroller { / dur.as_secs_f64(); state.load = state.utilisation / state.current_rate; - if state.delta_stat < delay_ms && state.load > self.config.high_load_level { + if state.delta_stat < delay_ms + && state.load > self.settings.advanced_settings.high_load_level + { state.safe_rates[state.nrate] = (state.current_rate * state.load).floor(); let max_rate = state .safe_rates @@ -284,7 +291,7 @@ impl Ratecontroller { * (1.0 + 0.1 * (1.0_f64 - state.current_rate / max_rate).max(0.0)) + (base_rate * 0.03); state.nrate += 1; - state.nrate %= self.config.speed_hist_size as usize; + state.nrate %= self.settings.advanced_settings.speed_hist_size as usize; } if state.delta_stat > delay_ms { @@ -318,7 +325,7 @@ impl Ratecontroller { self.control_snapshot.as_ref(), &reflectors, now_t, - self.config.tick_interval, + self.settings.advanced_settings.tick_interval, ); self.state_dl.deltas = down_deltas; self.state_ul.deltas = up_deltas; @@ -339,7 +346,7 @@ impl Ratecontroller { impl Ratecontroller { pub fn new( - config: Config, + settings: Settings, control_snapshot_rx: Receiver, reflectors_lock: ArcRwLock>, reselect_trigger: Sender, @@ -348,7 +355,7 @@ impl Ratecontroller { metrics: MetricsSender, ) -> anyhow::Result { Self::new_with_backends( - config, + settings, control_snapshot_rx, reflectors_lock, reselect_trigger, @@ -363,7 +370,8 @@ impl Ratecontroller { impl Ratecontroller { pub fn run(&mut self) -> anyhow::Result<()> { - let sleep_time = Duration::from_secs_f64(self.config.min_change_interval); + let sleep_time = + Duration::from_secs_f64(self.settings.advanced_settings.min_change_interval); let mut lastchg_t = Instant::now(); let mut lastdump_t = Instant::now(); @@ -375,12 +383,12 @@ impl Ratecontroller { let mut stats_fd: Option = None; let mut stats_fd_inner: File; - if !self.config.suppress_statistics { + if !self.settings.output.suppress_statistics { speed_hist_fd_inner = File::options() .create(true) .write(true) .truncate(true) - .open(self.config.speed_hist_file.as_str())?; + .open(self.settings.output.speed_hist_file.as_str())?; speed_hist_fd_inner.write_all("time,counter,upspeed,downspeed\n".as_bytes())?; speed_hist_fd_inner.flush()?; @@ -391,7 +399,7 @@ impl Ratecontroller { .create(true) .write(true) .truncate(true) - .open(self.config.stats_file.as_str())?; + .open(self.settings.output.stats_file.as_str())?; stats_fd_inner.write_all( "times,timens,rxload,txload,deltadelaydown,deltadelayup,dlrate,uprate\n".as_bytes(), @@ -409,14 +417,16 @@ impl Ratecontroller { sleep(sleep_time); let now_t = Instant::now(); - if now_t.duration_since(lastchg_t).as_secs_f64() > self.config.min_change_interval { + if now_t.duration_since(lastchg_t).as_secs_f64() + > self.settings.advanced_settings.min_change_interval + { // if it's been long enough, and the stats indicate needing to change speeds // change speeds here (self.state_dl.current_bytes, self.state_ul.current_bytes) = get_interface_stats( &mut self.stats_provider, - &self.config.download_interface, - &self.config.upload_interface, + &self.settings.network.download_interface, + &self.settings.network.upload_interface, self.down_direction, self.up_direction, )?; @@ -430,18 +440,18 @@ impl Ratecontroller { reflector: None, tags: &[], }); - self.state_dl.next_rate = self.config.download_min_kbits; - self.state_ul.next_rate = self.config.upload_min_kbits; + self.state_dl.next_rate = self.settings.network.download_min_kbits(); + self.state_ul.next_rate = self.settings.network.upload_min_kbits(); self.traffic_control.set_rate( &self.state_dl.shaper, self.state_dl.next_rate as u64, - self.config.dry_run, + self.settings.advanced_settings.dry_run, )?; self.traffic_control.set_rate( &self.state_ul.shaper, self.state_ul.next_rate as u64, - self.config.dry_run, + self.settings.advanced_settings.dry_run, )?; self.state_dl.current_rate = self.state_dl.next_rate; @@ -465,7 +475,7 @@ impl Ratecontroller { self.traffic_control.set_rate( &self.state_dl.shaper, self.state_dl.next_rate as u64, - self.config.dry_run, + self.settings.advanced_settings.dry_run, )?; } @@ -473,7 +483,7 @@ impl Ratecontroller { self.traffic_control.set_rate( &self.state_ul.shaper, self.state_ul.next_rate as u64, - self.config.dry_run, + self.settings.advanced_settings.dry_run, )?; } @@ -529,7 +539,7 @@ impl Ratecontroller { if let Some(ref mut fd) = speed_hist_fd && now_t.duration_since(lastdump_t).as_secs_f64() > 300.0 { - for i in 0..self.config.speed_hist_size as usize { + for i in 0..self.settings.advanced_settings.speed_hist_size as usize { let hist_time = Time::new(ClockId::Realtime); if let Err(e) = fd.write_all( format!( diff --git a/src/reflector_selector.rs b/src/reflector_selector.rs index 6a6e210..ab78889 100644 --- a/src/reflector_selector.rs +++ b/src/reflector_selector.rs @@ -5,10 +5,10 @@ // // SPDX-License-Identifier: MPL-2.0 -use crate::Config; use crate::SHUTDOWN; use crate::baseliner::ControlSnapshot; use crate::metrics::{Metric, MetricsSender}; +use crate::settings::Settings; use crate::util::RwLockExt; use flume::{Receiver, RecvError, RecvTimeoutError, Selector}; use log::{debug, info}; @@ -105,7 +105,7 @@ fn recent_rtt_for_peer(snapshot: Option<&ControlSnapshot>, peer: &IpAddr) -> Opt } pub struct ReflectorSelector { - pub config: Config, + pub settings: Settings, pub snapshot_rx: Receiver, pub reflector_peers_lock: Arc>>, pub reflector_pool: Vec, @@ -118,8 +118,9 @@ impl ReflectorSelector { let mut selector_sleep_time = Duration::new(30, 0); let mut reselection_count = 0; let mut latest_snapshot = None; - let baseline_sleep_time = - Duration::from_secs_f64(self.config.tick_interval * std::f64::consts::PI); + let baseline_sleep_time = Duration::from_secs_f64( + self.settings.advanced_settings.tick_interval * std::f64::consts::PI, + ); // Initial wait of several seconds to allow some OWD data to build up receive_snapshots_until( &self.snapshot_rx, @@ -157,7 +158,10 @@ impl ReflectorSelector { // After 40 reselections, slow down to every 15 minutes if reselection_count > 40 { - selector_sleep_time = Duration::new(self.config.peer_reselection_time * 60, 0); + selector_sleep_time = Duration::new( + self.settings.advanced_settings.peer_reselection_time * 60, + 0, + ); } let mut next_peers: Vec = Vec::new(); @@ -216,7 +220,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.config.num_reflectors; + let mut num_reflectors = self.settings.advanced_settings.num_reflectors; let candidate_pool_num = (2 * num_reflectors) as usize; candidates.truncate(candidate_pool_num);