diff --git a/Cargo.lock b/Cargo.lock index 6095311..5545a97 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -94,12 +94,51 @@ dependencies = [ "cc", ] +[[package]] +name = "config" +version = "0.15.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b85f248a4de22d204ceabc6299d89d2c70fbd7f09fea53c06c852369652d8139" +dependencies = [ + "pathdiff", + "serde_core", + "toml", + "winnow", +] + +[[package]] +name = "csv" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52cd9d68cf7efc6ddfaaee42e7288d3a99d613d4b50f76ce9827ae0c6e14f938" +dependencies = [ + "csv-core", + "itoa", + "ryu", + "serde_core", +] + +[[package]] +name = "csv-core" +version = "0.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704a3c26996a80471189265814dbc2c257598b96b8a7feae2d31ace646bb9782" +dependencies = [ + "memchr", +] + [[package]] name = "either" version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + [[package]] name = "errno" version = "0.3.14" @@ -154,6 +193,12 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280" +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" + [[package]] name = "icmp-socket2" version = "0.6.0" @@ -163,6 +208,16 @@ dependencies = [ "socket2", ] +[[package]] +name = "indexmap" +version = "2.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" +dependencies = [ + "equivalent", + "hashbrown", +] + [[package]] name = "itertools" version = "0.13.0" @@ -172,6 +227,12 @@ dependencies = [ "either", ] +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + [[package]] name = "js-sys" version = "0.3.103" @@ -201,9 +262,10 @@ dependencies = [ [[package]] name = "libuci-sys" version = "1.1.0" -source = "git+https://github.com/namib-project/rust-uci.git?rev=3e7b6153b6df92b45428c32bb6461756563deecb#3e7b6153b6df92b45428c32bb6461756563deecb" +source = "git+https://github.com/Lochnair/rust-uci.git?rev=fcfedd204fb76928ef617bf9631b041c816e03d4#fcfedd204fb76928ef617bf9631b041c816e03d4" dependencies = [ "bindgen", + "cc", "cmake", ] @@ -284,6 +346,12 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "pathdiff" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df94ce210e5bc13cb6651479fa48d14f601d9858cfe0467f43ae157023b938d3" + [[package]] name = "prettyplease" version = "0.2.37" @@ -350,7 +418,7 @@ checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" [[package]] name = "rust-uci" version = "0.1.6" -source = "git+https://github.com/namib-project/rust-uci.git?rev=3e7b6153b6df92b45428c32bb6461756563deecb#3e7b6153b6df92b45428c32bb6461756563deecb" +source = "git+https://github.com/Lochnair/rust-uci.git?rev=fcfedd204fb76928ef617bf9631b041c816e03d4#fcfedd204fb76928ef617bf9631b041c816e03d4" dependencies = [ "libc", "libuci-sys", @@ -383,12 +451,70 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + [[package]] name = "scopeguard" version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.2", +] + +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "serde_spanned" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26" +dependencies = [ + "serde_core", +] + [[package]] name = "shlex" version = "1.3.0" @@ -425,6 +551,8 @@ name = "sqm-autorate-rust" version = "0.4.1" dependencies = [ "anyhow", + "config", + "csv", "fastrand", "flume", "icmp-socket2", @@ -434,7 +562,9 @@ dependencies = [ "netlink-socket2", "rust-uci", "rustix", + "serde", "thiserror", + "uci-derive", ] [[package]] @@ -459,6 +589,21 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "target-triple" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3a6bfce3d99adfa72d24750a61f782f3036a81e7f86d8841ee1326deaebd171" + +[[package]] +name = "termcolor" +version = "1.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06794f8f6c5c898b3275aebefa6b8a1cb24cd2c6c79397ab15774837a0bc5755" +dependencies = [ + "winapi-util", +] + [[package]] name = "thiserror" version = "2.0.19" @@ -479,6 +624,70 @@ dependencies = [ "syn 3.0.2", ] +[[package]] +name = "toml" +version = "1.1.3+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c96ecdfa941c8fc4fcaed14f99ada8ebed502eef533015095a07e3301d4c3c" +dependencies = [ + "indexmap", + "serde_core", + "serde_spanned", + "toml_datetime", + "toml_parser", + "toml_writer", + "winnow", +] + +[[package]] +name = "toml_datetime" +version = "1.1.1+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3165f65f62e28e0115a00b2ebdd37eb6f3b641855f9d636d3cd4103767159ad7" +dependencies = [ + "serde_core", +] + +[[package]] +name = "toml_parser" +version = "1.1.2+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526" +dependencies = [ + "winnow", +] + +[[package]] +name = "toml_writer" +version = "1.1.2+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2" + +[[package]] +name = "trybuild" +version = "1.0.120" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e605bf6b39357663d8ba4e984f8be8da8df6bb32e81031d6889024ea8fd68e4" +dependencies = [ + "glob", + "serde", + "serde_derive", + "serde_json", + "target-triple", + "termcolor", + "toml", +] + +[[package]] +name = "uci-derive" +version = "0.1.0" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "trybuild", +] + [[package]] name = "unicode-ident" version = "1.0.24" @@ -530,6 +739,15 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "winapi-util" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" +dependencies = [ + "windows-sys", +] + [[package]] name = "windows-link" version = "0.2.1" @@ -544,3 +762,18 @@ checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" dependencies = [ "windows-link", ] + +[[package]] +name = "winnow" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23b97319f7b8343df12cc98938e5c3eb436064524c8d2b4e30a1d3a36eecdf81" +dependencies = [ + "memchr", +] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/Cargo.toml b/Cargo.toml index fcf3c73..1ed5008 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,6 +26,10 @@ rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] } thiserror = "2.0" libc = "0.2" flume = { version = "0.12.0", default-features = false, features = ["eventual-fairness", "select"] } +config = { version = "0.15.25", default-features = false } +csv = "1.4.0" +serde = { version = "1", features = ["derive"] } +uci-derive = { path = "crates/uci-derive" } [dependencies.log] features = ["std"] @@ -34,7 +38,16 @@ version = "0.4" [target.'cfg(target_os = "linux")'.dependencies] netlink-bindings = { version = "0.3.3", features = ["rt-link", "tc"] } netlink-socket2 = { version = "0.3.3" } -rust-uci = { git = "https://github.com/namib-project/rust-uci.git", rev = "3e7b6153b6df92b45428c32bb6461756563deecb", optional = true } + +[target.'cfg(unix)'.dependencies] +rust-uci = { git = "https://github.com/Lochnair/rust-uci.git", rev = "fcfedd204fb76928ef617bf9631b041c816e03d4", optional = true, default-features = false, } [features] +default = ["toml", "uci-vendored"] +toml = ["config/toml"] uci = ["dep:rust-uci"] +uci-vendored = ["uci", "rust-uci/vendored"] + +[workspace] +members = ["crates/uci-derive"] +resolver = "3" diff --git a/Dockerfile.ci b/Dockerfile.ci new file mode 100644 index 0000000..d810f3a --- /dev/null +++ b/Dockerfile.ci @@ -0,0 +1,7 @@ +ARG TARGET + +FROM ghcr.io/rust-cross/rust-musl-cross:${TARGET} + +RUN apt-get update \ + && apt-get install -y --no-install-recommends pkg-config \ + && rm -rf /var/lib/apt/lists/* diff --git a/Jenkinsfile b/Jenkinsfile index 4047aca..027ec69 100644 --- a/Jenkinsfile +++ b/Jenkinsfile @@ -11,10 +11,10 @@ pipeline { } } - agent { - docker { - image "ghcr.io/rust-cross/rust-musl-cross:${TARGET}" - alwaysPull true + agent { + dockerfile { + filename 'Dockerfile.ci' + additionalBuildArgs "--pull --build-arg TARGET=${TARGET}" } } @@ -25,7 +25,7 @@ pipeline { # 1. Set CARGO_HOME to a writable directory in the workspace export CARGO_HOME="$(pwd)/.cargo" mkdir -p "$CARGO_HOME" - + # 2. Add local bin to PATH (for tools you might install) export PATH="$(pwd)/.local/bin:${PATH}" @@ -44,14 +44,17 @@ pipeline { echo "----------------------------------" # 5. Check Toolchain & Build + # Force standalone static artifacts across the entire CI matrix. + export RUSTFLAGS="${RUSTFLAGS:+$RUSTFLAGS }-C target-feature=+crt-static" + # We grep for "nightly" in the version string (e.g., "rustc 1.95.0-nightly...") if rustc -V | grep -q "nightly"; then echo "Detected Nightly toolchain. Enabling '-Z build-std' to support panic=abort..." - + # Ensure rust-src is installed (needed for build-std) # We use '|| true' so it doesn't fail if already installed or network flakes rustup component add rust-src || true - + cargo build --release -Z build-std=std,panic_abort else echo "Detected Stable toolchain. Building with standard pre-compiled library..." @@ -71,7 +74,7 @@ pipeline { # 2. Verify it exists if [ -f "$BINARY_PATH" ]; then echo "Found compiled binary at: $BINARY_PATH" - + # 3. Rename with the matrix tag (e.g., sqm-autorate-rust-aarch64-musl) # This ensures unique filenames in the shared workspace root cp "$BINARY_PATH" "sqm-autorate-rust-${TARGET}" @@ -82,7 +85,7 @@ pipeline { exit 1 fi ''' - + // 4. Archive the specific file we just copied archiveArtifacts artifacts: "sqm-autorate-rust-${TARGET}", fingerprint: true, onlyIfSuccessful: true } diff --git a/build.rs b/build.rs index fd0ee4e..dba8fb6 100644 --- a/build.rs +++ b/build.rs @@ -7,8 +7,8 @@ fn main() { if uci_enabled && target_os != "linux" { println!( - "cargo::warning=feature `uci` is unsupported for target `{target}`; \ - UCI integration will not be compiled" + "cargo:warning=UCI support on {target} is intended for development \ + and testing; Linux/OpenWrt remains the production-supported platform" ); } diff --git a/crates/uci-derive/Cargo.toml b/crates/uci-derive/Cargo.toml new file mode 100644 index 0000000..6e0abf1 --- /dev/null +++ b/crates/uci-derive/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "uci-derive" +version = "0.1.0" +edition = "2024" +publish = false + +[lib] +proc-macro = true + +[dependencies] +proc-macro2 = "1" +quote = "1" +syn = { version = "2", features = ["full"] } + +[dev-dependencies] +trybuild = "1" diff --git a/crates/uci-derive/src/lib.rs b/crates/uci-derive/src/lib.rs new file mode 100644 index 0000000..f379b76 --- /dev/null +++ b/crates/uci-derive/src/lib.rs @@ -0,0 +1,252 @@ +// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair) +// +// SPDX-License-Identifier: MPL-2.0 + +use proc_macro::TokenStream; +use quote::quote; +use syn::ext::IdentExt; +use syn::{Data, DeriveInput, Field, Fields, FieldsNamed, LitStr, Token, parse_macro_input}; + +/// Generates `UciSectionSchema` from a settings struct. +/// +/// Every named field is mapped one-to-one: +/// +/// ```text +/// Rust field name -> UCI option name +/// Rust field name -> config-rs field name +/// ``` +#[proc_macro_derive(UciSection)] +pub fn derive_uci_section(input: TokenStream) -> TokenStream { + let input = parse_macro_input!(input as DeriveInput); + + match expand_uci_section(&input) { + Ok(tokens) => tokens.into(), + Err(error) => error.to_compile_error().into(), + } +} + +/// Generates `UciConfigSchema` from the root settings struct. +/// +/// Required container attribute: +/// +/// ```ignore +/// #[uci(package = "sqm-autorate-rust")] +/// ``` +/// +/// Every field must specify its UCI section: +/// +/// ```ignore +/// #[uci(section)] +/// network: NetworkSettings, +/// +/// #[uci(section = "advanced_settings")] +/// advanced: AdvancedSettings, +/// ``` +#[proc_macro_derive(UciConfig, attributes(uci))] +pub fn derive_uci_config(input: TokenStream) -> TokenStream { + let input = parse_macro_input!(input as DeriveInput); + + match expand_uci_config(&input) { + Ok(tokens) => tokens.into(), + Err(error) => error.to_compile_error().into(), + } +} + +fn expand_uci_section(input: &DeriveInput) -> syn::Result { + let struct_name = &input.ident; + let fields = require_named_fields(input)?; + + let mappings = fields + .named + .iter() + .map(|field| { + let field_ident = field + .ident + .as_ref() + .expect("named fields always have identifiers"); + + let field_name = field_ident.unraw().to_string(); + let field_name = LitStr::new(&field_name, field_ident.span()); + + quote! { + crate::settings::uci_schema::UciOptionMapping { + uci_option: #field_name, + config_field: #field_name, + } + } + }) + .collect::>(); + + let (impl_generics, type_generics, where_clause) = input.generics.split_for_impl(); + + Ok(quote! { + #[automatically_derived] + impl #impl_generics + crate::settings::uci_schema::UciSectionSchema + for #struct_name #type_generics + #where_clause + { + const UCI_OPTIONS: &'static [ + crate::settings::uci_schema::UciOptionMapping + ] = &[ + #(#mappings),* + ]; + } + }) +} + +fn expand_uci_config(input: &DeriveInput) -> syn::Result { + let struct_name = &input.ident; + let fields = require_named_fields(input)?; + let package = parse_package_name(input)?; + + let mut sections = Vec::with_capacity(fields.named.len()); + + for field in &fields.named { + let field_ident = field + .ident + .as_ref() + .expect("named fields always have identifiers"); + + let field_type = &field.ty; + + let config_section_name = field_ident.unraw().to_string(); + let config_section_name = LitStr::new(&config_section_name, field_ident.span()); + + let uci_section_name = parse_section_name(field)?; + + sections.push(quote! { + crate::settings::uci_schema::UciSectionMapping { + uci_section: #uci_section_name, + config_section: #config_section_name, + options: < + #field_type as + crate::settings::uci_schema::UciSectionSchema + >::UCI_OPTIONS, + } + }); + } + + let (impl_generics, type_generics, where_clause) = input.generics.split_for_impl(); + + Ok(quote! { + #[automatically_derived] + impl #impl_generics + crate::settings::uci_schema::UciConfigSchema + for #struct_name #type_generics + #where_clause + { + const UCI_PACKAGE: &'static str = #package; + + const UCI_SECTIONS: &'static [ + crate::settings::uci_schema::UciSectionMapping + ] = &[ + #(#sections),* + ]; + } + }) +} + +fn require_named_fields(input: &DeriveInput) -> syn::Result<&FieldsNamed> { + match &input.data { + Data::Struct(data) => match &data.fields { + Fields::Named(fields) => Ok(fields), + + Fields::Unnamed(_) | Fields::Unit => Err(syn::Error::new_spanned( + input, + "this derive only supports structs with named fields", + )), + }, + + Data::Enum(_) | Data::Union(_) => Err(syn::Error::new_spanned( + input, + "this derive only supports structs", + )), + } +} + +fn parse_package_name(input: &DeriveInput) -> syn::Result { + let mut package = None; + + for attribute in &input.attrs { + if !attribute.path().is_ident("uci") { + continue; + } + + attribute.parse_nested_meta(|meta| { + if meta.path.is_ident("package") { + if package.is_some() { + return Err(meta.error("duplicate `package` attribute")); + } + + let value: LitStr = meta.value()?.parse()?; + + if value.value().is_empty() { + return Err(syn::Error::new(value.span(), "`package` cannot be empty")); + } + + package = Some(value); + return Ok(()); + } + + Err(meta.error("expected `package = \"...\"`")) + })?; + } + + package.ok_or_else(|| { + syn::Error::new_spanned( + &input.ident, + "UciConfig requires `#[uci(package = \"...\")]`", + ) + }) +} + +fn parse_section_name(field: &Field) -> syn::Result { + let field_ident = field + .ident + .as_ref() + .expect("named fields always have identifiers"); + + let default_name = field_ident.unraw().to_string(); + let default_name = LitStr::new(&default_name, field_ident.span()); + + let mut section = None; + + for attribute in &field.attrs { + if !attribute.path().is_ident("uci") { + continue; + } + + attribute.parse_nested_meta(|meta| { + if meta.path.is_ident("section") { + if section.is_some() { + return Err(meta.error("duplicate `section` attribute")); + } + + if meta.input.peek(Token![=]) { + let value: LitStr = meta.value()?.parse()?; + + if value.value().is_empty() { + return Err(syn::Error::new(value.span(), "`section` cannot be empty")); + } + + section = Some(value); + } else { + section = Some(default_name.clone()); + } + + return Ok(()); + } + + Err(meta.error("expected `section` or `section = \"...\"`")) + })?; + } + + section.ok_or_else(|| { + syn::Error::new_spanned( + field, + "UciConfig fields require `#[uci(section)]` \ + or `#[uci(section = \"...\")]`", + ) + }) +} 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 a8a0b51..a4449cd 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; @@ -17,11 +16,20 @@ mod pinger_icmp_ts; mod platform; mod ratecontroller; mod reflector_selector; +mod settings; mod time; 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; @@ -33,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, @@ -63,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); @@ -99,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", @@ -144,24 +155,8 @@ 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") - { - StatsDirection::TX - } else { - StatsDirection::RX - }; - - let upload = if config.upload_interface.starts_with("ifb") - || config.upload_interface.starts_with("veth") - { - StatsDirection::RX - } else { - StatsDirection::TX - }; - - (download, upload) +fn interface_stats_directions(_: &Settings) -> (StatsDirection, StatsDirection) { + (StatsDirection::RX, StatsDirection::TX) } fn probe_identifier() -> u16 { @@ -170,31 +165,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 = [ @@ -206,7 +205,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 { @@ -217,7 +216,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), }) } @@ -240,8 +241,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 ***"); } @@ -253,7 +254,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(); @@ -268,10 +269,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), }; @@ -296,36 +297,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, @@ -336,12 +341,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 || { @@ -369,8 +374,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 || { @@ -389,7 +394,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), @@ -410,19 +415,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, @@ -465,7 +470,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 } @@ -475,9 +485,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/platform.rs b/src/platform.rs index 170d41a..b5e959a 100644 --- a/src/platform.rs +++ b/src/platform.rs @@ -42,6 +42,9 @@ mod linux; #[cfg(target_os = "macos")] mod macos; +#[cfg(unix)] +pub mod unix; + #[cfg(not(any(target_os = "freebsd", target_os = "linux", target_os = "macos")))] mod unsupported; diff --git a/src/platform/unix/mod.rs b/src/platform/unix/mod.rs new file mode 100644 index 0000000..a3f1153 --- /dev/null +++ b/src/platform/unix/mod.rs @@ -0,0 +1,2 @@ +#[cfg(feature = "uci")] +pub mod uci; diff --git a/src/platform/unix/uci.rs b/src/platform/unix/uci.rs new file mode 100644 index 0000000..cd48c44 --- /dev/null +++ b/src/platform/unix/uci.rs @@ -0,0 +1,287 @@ +use std::path::{Path, PathBuf}; + +use crate::settings::uci_schema::{UciConfigSchema, UciSectionMapping}; +use config::{ConfigError as SourceError, Map, Source, Value, ValueKind}; + +use rust_uci::config::{ + Config as UciConfig, Package as UciPackage, SectionSelector, Value as UciValue, +}; + +#[derive(Clone, Debug)] +enum PackageSelection { + Selected(Vec), + All, +} + +#[derive(Clone, Debug)] +pub(crate) struct UciSource { + packages: PackageSelection, + required: bool, + schema: &'static [UciSectionMapping], + config_dir: Option, + save_dir: Option, +} + +impl UciSource { + fn with_selection(packages: PackageSelection) -> Self { + Self { + packages, + required: true, + schema: &[], + config_dir: None, + save_dir: None, + } + } + + pub(crate) fn new(packages: I) -> Self + where + I: IntoIterator, + S: Into, + { + Self::with_selection(PackageSelection::Selected( + packages.into_iter().map(Into::into).collect(), + )) + } + + #[allow(dead_code)] + pub(crate) fn all() -> Self { + Self::with_selection(PackageSelection::All) + } + + pub(crate) fn required(mut self, required: bool) -> Self { + self.required = required; + self + } + + pub(crate) fn from_schema() -> Self + where + T: UciConfigSchema, + { + Self::new([T::UCI_PACKAGE]).with_schema(T::UCI_SECTIONS) + } + + pub(crate) fn with_schema(mut self, schema: &'static [UciSectionMapping]) -> Self { + self.schema = schema; + self + } + + #[allow(dead_code)] + pub(crate) fn with_directories( + mut self, + config_dir: impl AsRef, + save_dir: impl AsRef, + ) -> Self { + self.config_dir = Some(config_dir.as_ref().to_owned()); + self.save_dir = Some(save_dir.as_ref().to_owned()); + self + } + + fn open(&self) -> Result { + let mut uci = rust_uci::Uci::new() + .map_err(|e| SourceError::Message(format!("failed to open UCI: {e}")))?; + + if let Some(config_dir) = &self.config_dir { + uci.set_config_dir(config_dir).map_err(|e| { + SourceError::Message(format!( + "failed to set UCI config directory {config_dir:?}: {e}" + )) + })?; + } + + if let Some(save_dir) = &self.save_dir { + uci.set_save_dir(save_dir).map_err(|e| { + SourceError::Message(format!( + "failed to set UCI save directory {save_dir:?}: {e}" + )) + })?; + } + + Ok(uci.into()) + } + + /// + /// Converts a [`SectionSelector`] into a string representation. + /// Anonymous sections are converted to `section_type[index]` format, + /// because config-rs does not allow @-prefixed section names. + fn config_selector(selector: &SectionSelector<'_>) -> String { + match selector { + SectionSelector::Named(name) => (*name).to_owned(), + + SectionSelector::Anonymous { + section_type, + index, + } => format!("{section_type}[{index}]"), + } + } + + fn mapped_config_key( + &self, + selector: &SectionSelector<'_>, + uci_option: &str, + ) -> Option { + let section_type = match selector { + SectionSelector::Anonymous { + section_type, + index, + } if *index == 0 => *section_type, + + SectionSelector::Anonymous { .. } | SectionSelector::Named(_) => return None, + }; + + self.schema + .iter() + .filter(|section| section.uci_section == section_type) + .find_map(|section| { + section + .options + .iter() + .find(|option| option.uci_option == uci_option) + .map(|option| format!("{}.{}", section.config_section, option.config_field,)) + }) + } + + fn collect_package( + &self, + package_name: &str, + package: &UciPackage, + values: &mut Map, + ) -> Result<(), SourceError> { + let sections = package.sections().map_err(|e| { + SourceError::Message(format!( + "failed to retrieve sections from UCI package \ + {package_name:?}: {e}" + )) + })?; + + for section in sections { + let selector = section.selector().ok_or_else(|| { + SourceError::Message(format!( + "loaded UCI section of type {:?} in package \ + {package_name:?} has no selector", + section.type_() + )) + })?; + + let options = section.options().map_err(|e| { + SourceError::Message(format!( + "failed to retrieve options from UCI section \ + {package_name}.{selector}: {e}" + )) + })?; + + for option in options { + let option_name = option.name(); + + let uci_relative_key = format!("{selector}.{option_name}"); + + let config_relative_key = + format!("{}.{option_name}", Self::config_selector(&selector),); + + let uci_key = format!("{package_name}.{uci_relative_key}"); + + let source_key = format!("{package_name}.{config_relative_key}"); + + let destination_key = self + .mapped_config_key(&selector, option_name) + .unwrap_or(source_key); + + let Some(value) = option.get().map_err(|e| { + SourceError::Message(format!( + "failed to retrieve UCI option \ + {uci_key}: {e}" + )) + })? + else { + continue; + }; + + let origin = format!("UCI option {uci_key}"); + + let value = match value { + UciValue::String(value) => Value::new(Some(&origin), ValueKind::String(value)), + + UciValue::List(items) => { + let items = items + .into_iter() + .map(|item| Value::new(Some(&origin), ValueKind::String(item))) + .collect(); + + Value::new(Some(&origin), ValueKind::Array(items)) + } + }; + + if values.insert(destination_key.clone(), value).is_some() { + return Err(SourceError::Message(format!( + "duplicate UCI destination key \ + {destination_key:?} while mapping {uci_key:?}" + ))); + } + } + } + + Ok(()) + } +} + +impl Source for UciSource { + fn clone_into_box(&self) -> Box { + Box::new(self.clone()) + } + + fn collect(&self) -> Result, SourceError> { + let resolved = (|| { + let uci = self.open()?; + + let packages: Vec = match &self.packages { + PackageSelection::Selected(package_names) => package_names + .iter() + .map(|package_name| { + uci.package(package_name) + .map_err(|e| { + SourceError::Message(format!( + "failed to open UCI package {package_name:?}: {e}" + )) + })? + .ok_or_else(|| { + SourceError::NotFound(format!("UCI package {package_name:?}")) + }) + }) + .collect::>()?, + + PackageSelection::All => uci + .packages() + .map_err(|e| { + SourceError::Message(format!("failed to enumerate UCI packages: {e}")) + })? + .collect(), + }; + + Ok::<_, SourceError>((uci, packages)) + })(); + + let (_uci, packages) = match resolved { + Ok(resolved) => resolved, + + Err(_) if !self.required => { + return Ok(Map::new()); + } + + Err(error) => { + return Err(error); + } + }; + + let mut values = Map::new(); + + for package in packages { + let package_name = package + .name() + .map_err(|e| SourceError::Message(format!("failed to read UCI package name: {e}")))? + .to_owned(); + + self.collect_package(&package_name, &package, &mut values)?; + } + + Ok(values) + } +} 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); diff --git a/src/settings/mod.rs b/src/settings/mod.rs new file mode 100644 index 0000000..15152e9 --- /dev/null +++ b/src/settings/mod.rs @@ -0,0 +1,347 @@ +// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair) +// +// 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}; + +pub(crate) mod uci_schema; +mod validation; + +#[cfg(all(feature = "uci", unix))] +use crate::platform::unix::uci::UciSource; + +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::Message(format!( + "invalid boolean string: {}", + s + ))), + } + } +} + +fn deserialize_flex_bool<'de, D>(deserializer: D) -> Result +where + D: Deserializer<'de>, +{ + struct FlexiBoolVisitor; + + impl<'de> de::Visitor<'de> for FlexiBoolVisitor { + 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") + } + + fn visit_bool(self, value: bool) -> Result + where + E: de::Error, + { + Ok(value) + } + + fn visit_str(self, value: &str) -> Result + where + E: de::Error, + { + value + .parse::() + .map(|value| value.0) + .map_err(E::custom) + } + + fn visit_string(self, value: String) -> Result + where + E: de::Error, + { + self.visit_str(&value) + } + } + + deserializer.deserialize_any(FlexiBoolVisitor) +} + +#[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 { + match s.to_lowercase().as_str() { + "icmp" => Ok(MeasurementType::Icmp), + "icmp-timestamps" | "icmp_timestamps" => Ok(MeasurementType::IcmpTimestamps), + _ => Err(ConfigError::Message(format!( + "invalid measurement type: {s}" + ))), + } + } +} + +#[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::Message(format!( + "invalid observability protocol: {s}", + ))), + } + } +} + +#[derive(Debug, Deserialize)] +struct ReflectorRecord { + reflector_ip: IpAddr, + ip_version: u8, + #[allow(dead_code)] + description: String, +} + +#[derive(Clone, Debug, Deserialize, UciConfig)] +#[uci(package = "sqm-autorate-rust")] +pub(crate) struct Settings { + #[uci(section)] + pub(crate) network: NetworkSettings, + + #[serde(default)] + #[uci(section)] + pub(crate) output: OutputSettings, + + #[serde(default)] + #[uci(section)] + pub(crate) observability: ObservabilitySettings, + + #[serde(default)] + #[uci(section)] + pub(crate) advanced_settings: AdvancedSettings, +} + +impl Settings { + pub(crate) fn load() -> Result { + let builder = Config::builder(); + + #[cfg(all(feature = "uci", unix))] + let builder = builder.add_source(UciSource::from_schema::().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()?; + + built_config.try_deserialize() + } + + 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, IpAddr::V6(_)) => {} + (version, address) => { + bail!("reflector {address} has inconsistent IP version {version}"); + } + } + + reflectors.push(record.reflector_ip); + } + + Ok(reflectors) + } +} + +#[derive(Clone, Debug, Deserialize, UciSection)] +pub(crate) struct NetworkSettings { + pub(crate) download_interface: String, + pub(crate) upload_interface: String, + pub(crate) download_base_kbits: f64, + #[serde(default = "default_min_percent")] + pub(crate) download_min_percent: f64, + pub(crate) upload_base_kbits: f64, + #[serde(default = "default_min_percent")] + pub(crate) upload_min_percent: f64, +} + +impl NetworkSettings { + pub(crate) fn download_min_kbits(&self) -> f64 { + (self.download_base_kbits * self.download_min_percent / 100.0).floor() + } + + pub(crate) fn upload_min_kbits(&self) -> f64 { + (self.upload_base_kbits * self.upload_min_percent / 100.0).floor() + } +} + +#[derive(Clone, Debug, Deserialize, UciSection)] +#[serde(default)] +pub(crate) struct OutputSettings { + #[serde(deserialize_with = "deserialize_from_str")] + pub(crate) log_level: Level, + pub(crate) speed_hist_file: String, + pub(crate) stats_file: String, + #[serde(deserialize_with = "deserialize_flex_bool")] + pub(crate) suppress_statistics: bool, +} + +impl Default for OutputSettings { + fn default() -> Self { + Self { + log_level: Level::Error, + speed_hist_file: "/tmp/sqm-speedhist.csv".to_owned(), + stats_file: "/tmp/sqm-autorate-rust.csv".to_owned(), + suppress_statistics: false, + } + } +} + +#[derive(Clone, Debug, Deserialize, UciSection)] +#[serde(default)] +pub(crate) struct ObservabilitySettings { + #[serde(deserialize_with = "deserialize_flex_bool")] + pub(crate) enabled: bool, + #[serde(deserialize_with = "deserialize_from_str")] + pub(crate) protocol: ObservabilityProtocol, + pub(crate) host: Option, + pub(crate) port: u16, + pub(crate) batch_size: usize, + pub(crate) batch_timeout_ms: u64, + #[serde(deserialize_with = "deserialize_flex_bool")] + pub(crate) export_ping_metrics: bool, + #[serde(deserialize_with = "deserialize_flex_bool")] + pub(crate) export_rate_metrics: bool, + #[serde(deserialize_with = "deserialize_flex_bool")] + pub(crate) export_baseline_metrics: bool, + #[serde(deserialize_with = "deserialize_flex_bool")] + pub(crate) export_events: bool, + pub(crate) host_tag: String, +} + +impl Default for ObservabilitySettings { + fn default() -> Self { + Self { + enabled: false, + protocol: ObservabilityProtocol::Udp, + host: None, + port: 8089, + batch_size: 25, + batch_timeout_ms: 100, + export_ping_metrics: false, + export_rate_metrics: true, + export_baseline_metrics: false, + export_events: true, + host_tag: default_host_tag(), + } + } +} + +#[derive(Clone, Debug, Deserialize, UciSection)] +#[serde(default)] +pub(crate) struct AdvancedSettings { + pub(crate) download_delay_ms: f64, + 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) speed_hist_size: u32, + pub(crate) tick_interval: f64, + pub(crate) upload_delay_ms: f64, + #[serde(deserialize_with = "deserialize_flex_bool")] + pub(crate) dry_run: bool, +} + +impl Default for AdvancedSettings { + fn default() -> Self { + Self { + 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(), + speed_hist_size: 100, + tick_interval: 0.5, + upload_delay_ms: 15.0, + dry_run: false, + } + } +} + +fn default_min_percent() -> f64 { + 20.0 +} + +fn default_host_tag() -> String { + rustix::system::uname() + .nodename() + .to_string_lossy() + .into_owned() +} + +fn deserialize_from_str<'de, D, T>(deserializer: D) -> Result +where + D: Deserializer<'de>, + T: FromStr, + T::Err: Display, +{ + let value = String::deserialize(deserializer)?; + value.parse().map_err(de::Error::custom) +} diff --git a/src/settings/uci_schema.rs b/src/settings/uci_schema.rs new file mode 100644 index 0000000..aedd51d --- /dev/null +++ b/src/settings/uci_schema.rs @@ -0,0 +1,29 @@ +// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair) +// +// SPDX-License-Identifier: MPL-2.0 + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct UciOptionMapping { + pub(crate) uci_option: &'static str, + pub(crate) config_field: &'static str, +} + +#[derive(Clone, Copy, Debug)] +pub(crate) struct UciSectionMapping { + /// UCI section type, corresponding to `@section_type[0]`. + pub(crate) uci_section: &'static str, + + /// Top-level config-rs / Serde settings key. + pub(crate) config_section: &'static str, + + pub(crate) options: &'static [UciOptionMapping], +} + +pub(crate) trait UciSectionSchema { + const UCI_OPTIONS: &'static [UciOptionMapping]; +} + +pub(crate) trait UciConfigSchema { + const UCI_PACKAGE: &'static str; + const UCI_SECTIONS: &'static [UciSectionMapping]; +} diff --git a/src/settings/validation.rs b/src/settings/validation.rs new file mode 100644 index 0000000..231c85d --- /dev/null +++ b/src/settings/validation.rs @@ -0,0 +1,378 @@ +// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair) +// +// SPDX-License-Identifier: MPL-2.0 +// + +use super::{AdvancedSettings, NetworkSettings, ObservabilitySettings, OutputSettings, Settings}; +use std::error::Error; +use std::fmt::{self, Display, Formatter}; + +const WARN_DOWNLOAD_MIN_KBITS: f64 = 1_000.0; +const WARN_UPLOAD_MIN_KBITS: f64 = 500.0; +const WARN_HIGH_MIN_PERCENT: f64 = 80.0; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum ValidationSeverity { + Warning, + Error, +} + +#[derive(Debug)] +struct ValidationIssue { + severity: ValidationSeverity, + path: &'static str, + message: String, +} + +#[derive(Debug, Default)] +pub(crate) struct ValidationReport { + issues: Vec, +} + +impl ValidationReport { + fn warning(&mut self, path: &'static str, message: impl Into) { + self.issues.push(ValidationIssue { + severity: ValidationSeverity::Warning, + path, + message: message.into(), + }); + } + + fn error(&mut self, path: &'static str, message: impl Into) { + self.issues.push(ValidationIssue { + severity: ValidationSeverity::Error, + path, + message: message.into(), + }); + } + + fn finish(self) -> Result<(), Self> { + let mut errors = Vec::new(); + + for issue in self.issues { + match issue.severity { + ValidationSeverity::Warning => { + eprintln!("warning: {}: {}", issue.path, issue.message); + } + + ValidationSeverity::Error => { + errors.push(issue); + } + } + } + + if errors.is_empty() { + Ok(()) + } else { + Err(Self { issues: errors }) + } + } +} + +impl Display for ValidationReport { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + write!(f, "settings validation failed:")?; + + for issue in &self.issues { + write!(f, "\n {}: {}", issue.path, issue.message)?; + } + + Ok(()) + } +} + +impl Error for ValidationReport {} + +impl Settings { + pub(crate) fn validate(&self) -> Result<(), ValidationReport> { + let mut report = ValidationReport::default(); + + self.network.validate(&mut report); + self.output.validate(&mut report); + self.observability.validate(&mut report); + self.advanced_settings.validate(&mut report); + + report.finish() + } +} + +impl NetworkSettings { + fn validate(&self, report: &mut ValidationReport) { + if self.download_interface.trim().is_empty() { + report.error("network.download_interface", "must not be empty"); + } + + if self.upload_interface.trim().is_empty() { + report.error("network.upload_interface", "must not be empty"); + } + + let download_base_valid = validate_greater_than_zero_f64( + report, + "network.download_base_kbits", + self.download_base_kbits, + ); + + let upload_base_valid = validate_greater_than_zero_f64( + report, + "network.upload_base_kbits", + self.upload_base_kbits, + ); + + let download_percent_valid = validate_percent( + report, + "network.download_min_percent", + self.download_min_percent, + ); + + let upload_percent_valid = validate_percent( + report, + "network.upload_min_percent", + self.upload_min_percent, + ); + + if download_percent_valid && self.download_min_percent > WARN_HIGH_MIN_PERCENT { + report.warning( + "network.download_min_percent", + format!( + "{}% leaves little room for the shaper to reduce the download rate", + self.download_min_percent, + ), + ); + } + + if upload_percent_valid && self.upload_min_percent > WARN_HIGH_MIN_PERCENT { + report.warning( + "network.upload_min_percent", + format!( + "{}% leaves little room for the shaper to reduce the upload rate", + self.upload_min_percent, + ), + ); + } + + if download_base_valid && download_percent_valid { + let minimum = self.download_min_kbits(); + + if minimum < WARN_DOWNLOAD_MIN_KBITS { + report.warning( + "network.download_min_percent", + format!( + "results in a minimum download rate of only {minimum:.0} kbit/s \ + ({:.1}% of {:.0} kbit/s); connectivity may become severely constrained", + self.download_min_percent, self.download_base_kbits, + ), + ); + } + } + + if upload_base_valid && upload_percent_valid { + let minimum = self.upload_min_kbits(); + + if minimum < WARN_UPLOAD_MIN_KBITS { + report.warning( + "network.upload_min_percent", + format!( + "results in a minimum upload rate of only {minimum:.0} kbit/s \ + ({:.1}% of {:.0} kbit/s); connectivity may become severely constrained", + self.upload_min_percent, self.upload_base_kbits, + ), + ); + } + } + } +} + +impl OutputSettings { + fn validate(&self, report: &mut ValidationReport) { + if self.suppress_statistics { + return; + } + + let speed_hist_file = self.speed_hist_file.trim(); + let stats_file = self.stats_file.trim(); + + if speed_hist_file.is_empty() { + report.error( + "output.speed_hist_file", + "must not be empty when statistics are enabled", + ); + } + + if stats_file.is_empty() { + report.error( + "output.stats_file", + "must not be empty when statistics are enabled", + ); + } + + if !speed_hist_file.is_empty() && !stats_file.is_empty() && speed_hist_file == stats_file { + report.error( + "output.stats_file", + "must not refer to the same file as output.speed_hist_file", + ); + } + } +} + +impl ObservabilitySettings { + fn validate(&self, report: &mut ValidationReport) { + if !self.enabled { + return; + } + + if self + .host + .as_deref() + .is_none_or(|host| host.trim().is_empty()) + { + report.error( + "observability.host", + "must be set when observability is enabled", + ); + } + + if self.port == 0 { + report.error("observability.port", "must be greater than 0"); + } + + if self.batch_size == 0 { + report.error("observability.batch_size", "must be greater than 0"); + } + + if self.batch_timeout_ms == 0 { + report.error("observability.batch_timeout_ms", "must be greater than 0"); + } + + if !self.export_ping_metrics + && !self.export_rate_metrics + && !self.export_baseline_metrics + && !self.export_events + { + report.warning( + "observability.enabled", + "observability is enabled but all metric and event exports are disabled", + ); + } + } +} + +impl AdvancedSettings { + fn validate(&self, report: &mut ValidationReport) { + validate_zero_or_greater_f64( + report, + "advanced_settings.download_delay_ms", + self.download_delay_ms, + ); + + validate_zero_or_greater_f64( + report, + "advanced_settings.upload_delay_ms", + self.upload_delay_ms, + ); + + let high_load_level_valid = validate_zero_or_greater_f64( + report, + "advanced_settings.high_load_level", + self.high_load_level, + ); + + validate_greater_than_zero_f64( + report, + "advanced_settings.min_change_interval", + self.min_change_interval, + ); + + validate_greater_than_zero_f64( + report, + "advanced_settings.tick_interval", + 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", + "must be greater than 0", + ); + } + + if high_load_level_valid { + if self.high_load_level == 0.0 { + report.warning( + "advanced_settings.high_load_level", + "0 causes any positive load to be treated as high load", + ); + } else if self.high_load_level > 1.0 { + report.warning( + "advanced_settings.high_load_level", + format!( + "{} requires utilisation to exceed the current shaper rate \ + before the high-load threshold is reached", + self.high_load_level, + ), + ); + } + } + } +} + +fn validate_greater_than_zero_f64( + report: &mut ValidationReport, + path: &'static str, + value: f64, +) -> bool { + if !value.is_finite() { + report.error(path, format!("must be finite, got {value}")); + false + } else if value <= 0.0 { + report.error(path, format!("must be greater than 0, got {value}")); + false + } else { + true + } +} + +fn validate_zero_or_greater_f64( + report: &mut ValidationReport, + path: &'static str, + value: f64, +) -> bool { + if !value.is_finite() { + report.error(path, format!("must be finite, got {value}")); + false + } else if value < 0.0 { + report.error(path, format!("must be 0 or greater, got {value}")); + false + } else { + true + } +} + +fn validate_percent(report: &mut ValidationReport, path: &'static str, value: f64) -> bool { + if !value.is_finite() { + report.error(path, format!("must be finite, got {value}")); + false + } else if !(1.0..=100.0).contains(&value) { + report.error(path, format!("must be between 1 and 100, got {value}")); + false + } else { + true + } +}