Lochnair-Labs/sqm-autorate-rust
2
0 Fork 0
Code Issues 5 Pull Requests Actions 14 Packages Projects Releases Wiki Activity

feat(config): migrate configuration loading to config-rs #68

Merged
Lochnair merged 9 commits from feature/config into master 2026-08-07 09:10:14 +00:00
Conversation 2 Commits 9 Files Changed 21
+174 -563
8 changed files with 174 additions and 563 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 7e0f434251 - Show all commits

No files matched your search

src/baseliner.rs
+11 -5
View File
@@ -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<bool>,
pub start_time: Instant,
pub stats_rx: Receiver<PingReply>,
@@ -150,8 +150,14 @@ impl Baseliner {
pub fn run(&self) -> anyhow::Result<()> {
let mut reflectors = HashMap::<IpAddr, ReflectorState>::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(
src/config.rs
-436
View File
@@ -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<P>(filename: P) -> io::Result<io::Lines<io::BufReader<File>>>
where
P: AsRef<Path>,
{
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<Self, Self::Err> {
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<FlexiBool> 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<Self, Self::Err> {
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<Self, Self::Err> {
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<String>,
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<Self> {
let config = Self {
// Network section
download_base_kbits: Self::get::<f64>(
"SQMA_DOWNLOAD_BASE_KBITS",
"sqm-autorate-rust.@network[0].download_base_kbits",
None,
)?,
download_interface: Self::get::<String>(
"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::<f64>(
"SQMA_UPLOAD_BASE_KBITS",
"sqm-autorate-rust.@network[0].upload_base_kbits",
None,
)?,
upload_interface: Self::get::<String>(
"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::<Level>(
"SQMA_LOG_LEVEL",
"sqm-autorate-rust.@output[0].log_level",
Some(Level::Error),
)?,
speed_hist_file: Self::get::<String>(
"SQMA_SPEED_HIST_FILE",
"sqm-autorate-rust.@output[0].speed_hist_file",
Some("/tmp/sqm-speedhist.csv".parse()?),
)?,
stats_file: Self::get::<String>(
"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::<ObservabilityProtocol>(
"SQMA_OBSERVABILITY_PROTOCOL",
"sqm-autorate-rust.@observability[0].protocol",
Some(ObservabilityProtocol::Udp),
)?,
observability_host: Self::get_optional::<String>(
"SQMA_OBSERVABILITY_HOST",
"sqm-autorate-rust.@observability[0].host",
),
observability_port: Self::get::<u16>(
"SQMA_OBSERVABILITY_PORT",
"sqm-autorate-rust.@observability[0].port",
Some(8089),
)?,
observability_batch_size: Self::get::<usize>(
"SQMA_OBSERVABILITY_BATCH_SIZE",
"sqm-autorate-rust.@observability[0].batch_size",
Some(25),
)?,
observability_batch_timeout_ms: Self::get::<u64>(
"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::<String>(
"SQMA_OBSERVABILITY_HOST_TAG",
"sqm-autorate-rust.@observability[0].host_tag",
Some(Self::get_host_tag()),
)?,
// Advanced section
download_delay_ms: Self::get::<f64>(
"SQMA_DOWNLOAD_DELAY_MS",
"sqm-autorate-rust.@advanced_settings[0].download_delay_ms",
Some(15.0),
)?,
high_load_level: Self::get::<f64>(
"SQMA_HIGH_LOAD_LEVEL",
"sqm-autorate-rust.@advanced_settings[0].high_load_level",
Some(0.8),
)?,
measurement_type: Self::get::<MeasurementType>(
"SQMA_MEASUREMENT_TYPE",
"sqm-autorate-rust.@advanced_settings[0].measurement_type",
Some(MeasurementType::IcmpTimestamps),
)?,
min_change_interval: Self::get::<f64>(
"SQMA_MIN_CHANGE_INTERVAL",
"sqm-autorate-rust.@advanced_settings[0].min_change_interval",
Some(0.5),
)?,
num_reflectors: Self::get::<u8>(
"SQMA_NUM_REFLECTORS",
"sqm-autorate-rust.@advanced_settings[0].num_reflectors",
Some(5),
)?,
peer_reselection_time: Self::get::<u64>(
"SQMA_PEER_RESELECTION_TIME",
"sqm-autorate-rust.@advanced_settings[0].peer_reselection_time",
Some(15),
)?,
reflector_list_file: Self::get::<String>(
"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::<u32>(
"SQMA_SPEED_HIST_SIZE",
"sqm-autorate-rust.@advanced_settings[0].speed_hist_size",
Some(100),
)?,
tick_interval: Self::get::<f64>(
"SQMA_TICK_INTERVAL",
"sqm-autorate-rust.@advanced_settings[0].tick_interval",
Some(0.5),
)?,
upload_delay_ms: Self::get::<f64>(
"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::<f64>(
"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::<f64>(
"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<T: FromStr>(env_key: &str, uci_key: &str, default: Option<T>) -> Result<T, ConfigError> {
match Self::get_value(env_key, uci_key) {
Some(val) => match val.parse::<T>() {
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<bool>) -> Result<bool, ConfigError> {
Self::get::<FlexiBool>(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<T: FromStr>(env_key: &str, uci_key: &str) -> Option<T> {
Self::get_value(env_key, uci_key).and_then(|val| val.parse::<T>().ok())
}
fn get_value(env_key: &str, uci_key: &str) -> Option<String> {
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<String> {
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<String> {
None
}
pub fn load_reflectors(&self) -> Result<Vec<IpAddr>> {
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<IpAddr> = 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)
}
}
src/main.rs
+91 -64
View File
@@ -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<RwLock<Vec<IpAddr>>>,
pool: Vec<IpAddr>,
@@ -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<T: TrafficControlBackend>(
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<T: TrafficControlBackend>(
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<T: TrafficControlBackend>(
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<ReflectorSetup> {
let configured = config.load_reflectors()?;
fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
let configured = settings.load_reflectors()?;
let configured_count = configured.len();
let default_reflectors = [
@@ -207,7 +221,7 @@ fn setup_reflectors(config: &Config) -> anyhow::Result<ReflectorSetup> {
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<ReflectorSetup> {
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::Error>) -> 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)
}
src/metrics.rs
+9 -9
View File
@@ -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<AtomicU32>,
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)?
src/pinger_icmp.rs
+1 -1
View File
@@ -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;
src/pinger_icmp_ts.rs
+1 -1
View File
@@ -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;
src/ratecontroller.rs
+51 -41
View File
@@ -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<S: InterfaceStatsProvider, T: TrafficControlBackend> {
config: Config,
settings: Settings,
control_snapshot: Option<ControlSnapshot>,
control_snapshot_rx: Receiver<ControlSnapshot>,
down_direction: StatsDirection,
@@ -173,7 +173,7 @@ pub struct Ratecontroller<S: InterfaceStatsProvider, T: TrafficControlBackend> {
impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
fn new_with_backends(
config: Config,
settings: Settings,
control_snapshot_rx: Receiver<ControlSnapshot>,
reflectors_lock: ArcRwLock<Vec<IpAddr>>,
reselect_trigger: Sender<bool>,
@@ -183,23 +183,28 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
mut stats_provider: S,
mut traffic_control: T,
) -> anyhow::Result<Self> {
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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
/ 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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
* (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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
pub fn new(
config: Config,
settings: Settings,
control_snapshot_rx: Receiver<ControlSnapshot>,
reflectors_lock: ArcRwLock<Vec<IpAddr>>,
reselect_trigger: Sender<bool>,
@@ -348,7 +355,7 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
metrics: MetricsSender,
) -> anyhow::Result<Self> {
Self::new_with_backends(
config,
settings,
control_snapshot_rx,
reflectors_lock,
reselect_trigger,
@@ -363,7 +370,8 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
let mut stats_fd: Option<File> = 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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
.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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
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!(
src/reflector_selector.rs
+10 -6
View File
@@ -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<ControlSnapshot>,
pub reflector_peers_lock: Arc<RwLock<Vec<IpAddr>>>,
pub reflector_pool: Vec<IpAddr>,
@@ -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<IpAddr> = 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);