3 Commits
Author SHA1 Message Date
Nils Andreas Svee 8ef47e769b feat(reflectors): refactor reflector configuration
Build binaries / Build (arm-musleabi) (push) Successful in 1m46s
Build binaries / Build (aarch64-musl) (push) Successful in 1m46s
Build binaries / Build (arm-musleabihf) (push) Successful in 2m1s
Build binaries / Build (armv7-musleabi) (push) Successful in 2m1s
Build binaries / Build (armv7-musleabihf) (push) Successful in 1m53s
Build binaries / Build (i686-musl) (push) Successful in 1m53s
Build binaries / Build (mips-musl) (push) Successful in 2m4s
Build binaries / Build (mips64-muslabi64) (push) Successful in 2m1s
Build binaries / Build (mips64el-muslabi64) (push) Successful in 2m5s
Build binaries / Build (mipsel-musl) (push) Successful in 2m3s
Build binaries / Build (riscv64gc-musl) (push) Successful in 1m43s
Build binaries / Build (x86_64-musl) (push) Successful in 1m47s
2026-10-13 00:34:50 +07:00
Nils Andreas Svee 15f037afa4 feat(uci): allow skipping derived fields and map repeated sections
Build binaries / Build (aarch64-musl) (push) Successful in 1m47s
Build binaries / Build (arm-musleabi) (push) Successful in 1m49s
Build binaries / Build (arm-musleabihf) (push) Successful in 1m39s
Build binaries / Build (armv7-musleabi) (push) Successful in 1m47s
Build binaries / Build (armv7-musleabihf) (push) Successful in 1m37s
Build binaries / Build (i686-musl) (push) Successful in 1m41s
Build binaries / Build (mips-musl) (push) Successful in 1m57s
Build binaries / Build (mips64-muslabi64) (push) Successful in 2m4s
Build binaries / Build (mips64el-muslabi64) (push) Successful in 2m5s
Build binaries / Build (mipsel-musl) (push) Successful in 2m10s
Build binaries / Build (riscv64gc-musl) (push) Successful in 1m52s
Build binaries / Build (x86_64-musl) (push) Successful in 1m45s
2026-10-12 00:53:31 +07:00
Nils Andreas Svee d93f507c3f feat(measurement): add IRTT engine 2026-10-12 00:13:14 +07:00
11 changed files with 749 additions and 157 deletions

No files matched your search

Generated
+165 -48
View File
@@ -4,9 +4,9 @@ version = 4
[[package]]
name = "aho-corasick"
version = "1.1.4"
version = "1.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301"
checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba"
dependencies = [
"memchr",
]
@@ -56,9 +56,18 @@ dependencies = [
[[package]]
name = "bitflags"
version = "2.13.1"
version = "2.13.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da"
checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06"
[[package]]
name = "block-buffer"
version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa"
dependencies = [
"hybrid-array",
]
[[package]]
name = "bumpalo"
@@ -74,9 +83,9 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
[[package]]
name = "cc"
version = "1.3.0"
version = "1.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c89588d05638b5b4594a3348a2d6c20277e43a7f5c5202b05cc56888475a47b8"
checksum = "f74872d07caf508b30a21f6836e7d7016a2eaf7d9ff4f48deaa58cd8a0407630"
dependencies = [
"find-msvc-tools",
"shlex 2.0.1",
@@ -93,9 +102,9 @@ dependencies = [
[[package]]
name = "cfg-if"
version = "1.0.4"
version = "1.0.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600"
[[package]]
name = "cfg_aliases"
@@ -105,9 +114,9 @@ checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527"
[[package]]
name = "clang-sys"
version = "1.8.1"
version = "1.9.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4"
checksum = "157a8ba7b480713b56f4c09fd13fc3e0a22a5dfab8097ba61cbc5feef950788a"
dependencies = [
"glob",
"libc",
@@ -123,6 +132,12 @@ dependencies = [
"cc",
]
[[package]]
name = "cmov"
version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a"
[[package]]
name = "config"
version = "0.15.27"
@@ -135,6 +150,21 @@ dependencies = [
"winnow",
]
[[package]]
name = "const-oid"
version = "0.10.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c"
[[package]]
name = "crypto-common"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453"
dependencies = [
"hybrid-array",
]
[[package]]
name = "csv"
version = "1.4.0"
@@ -157,10 +187,31 @@ dependencies = [
]
[[package]]
name = "either"
version = "1.16.0"
name = "ctutils"
version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e"
checksum = "03bb0e1cc970d482d121d9a1744999169b69a07470b3d644a7894e53fcaf4574"
dependencies = [
"cmov",
]
[[package]]
name = "digest"
version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2"
dependencies = [
"block-buffer",
"const-oid",
"crypto-common",
"ctutils",
]
[[package]]
name = "either"
version = "1.19.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0e9c71c2167ca323c882b99918929403426e2373ea17242ff5653e0d5e1058be"
[[package]]
name = "equivalent"
@@ -189,9 +240,9 @@ dependencies = [
[[package]]
name = "find-msvc-tools"
version = "0.1.9"
version = "0.1.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
checksum = "aedcfb3409746eddb02b9e19ebda1c3394f759a152e48ee875a0844d1b955484"
[[package]]
name = "flume"
@@ -213,9 +264,9 @@ checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e"
[[package]]
name = "futures-sink"
version = "0.3.32"
version = "0.3.34"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893"
checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d"
[[package]]
name = "getrandom"
@@ -232,9 +283,9 @@ dependencies = [
[[package]]
name = "glob"
version = "0.3.3"
version = "0.3.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280"
checksum = "e4eba85ea1d0a966a983acd07deee566e67395d2d96b6fb39e62b5a833f1eb0b"
[[package]]
name = "hashbrown"
@@ -242,6 +293,24 @@ version = "0.17.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
[[package]]
name = "hmac"
version = "0.13.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f"
dependencies = [
"digest",
]
[[package]]
name = "hybrid-array"
version = "0.4.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17"
dependencies = [
"typenum",
]
[[package]]
name = "icmp-socket2"
version = "0.6.4"
@@ -257,14 +326,39 @@ dependencies = [
[[package]]
name = "indexmap"
version = "2.14.0"
version = "2.14.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9"
checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855"
dependencies = [
"equivalent",
"hashbrown",
]
[[package]]
name = "irtt-client"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a59c24704bd94f19b84b2b0190aa9163efede499f8cf4eb02922443cfee08339"
dependencies = [
"irtt-proto",
"nix",
"socket2",
"thiserror",
"tokio",
]
[[package]]
name = "irtt-proto"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a56ff307983b8f6477724e08cf726c5e24975b9affeb853d0830d91c60026ab2"
dependencies = [
"hmac",
"md-5",
"subtle",
"thiserror",
]
[[package]]
name = "itertools"
version = "0.13.0"
@@ -282,9 +376,9 @@ checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
[[package]]
name = "js-sys"
version = "0.3.103"
version = "0.3.106"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102"
checksum = "7883d941dae510fb2d978fc3fe018c71c9e2892fd38854de3e8b92c2e5ad9cc5"
dependencies = [
"cfg-if",
"wasm-bindgen",
@@ -309,7 +403,7 @@ dependencies = [
[[package]]
name = "libuci-sys"
version = "1.1.0"
source = "git+https://github.com/Lochnair/rust-uci.git?rev=fcfedd204fb76928ef617bf9631b041c816e03d4#fcfedd204fb76928ef617bf9631b041c816e03d4"
source = "git+https://github.com/Lochnair/rust-uci.git?rev=6835397f4f4ba8b69e75e727105186865c357417#6835397f4f4ba8b69e75e727105186865c357417"
dependencies = [
"bindgen",
"cc",
@@ -337,6 +431,16 @@ version = "0.4.34"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
[[package]]
name = "md-5"
version = "0.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98"
dependencies = [
"cfg-if",
"digest",
]
[[package]]
name = "memchr"
version = "2.8.3"
@@ -507,9 +611,9 @@ dependencies = [
[[package]]
name = "regex-automata"
version = "0.4.16"
version = "0.4.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8fcfdb36bda0c880c5931cdc7a2bcdc8ba4556847b9d912bca70bc94708711ad"
checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2"
dependencies = [
"aho-corasick",
"memchr",
@@ -525,7 +629,7 @@ checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4"
[[package]]
name = "rust-uci"
version = "0.1.6"
source = "git+https://github.com/Lochnair/rust-uci.git?rev=fcfedd204fb76928ef617bf9631b041c816e03d4#fcfedd204fb76928ef617bf9631b041c816e03d4"
source = "git+https://github.com/Lochnair/rust-uci.git?rev=6835397f4f4ba8b69e75e727105186865c357417#6835397f4f4ba8b69e75e727105186865c357417"
dependencies = [
"libc",
"libuci-sys",
@@ -615,9 +719,9 @@ dependencies = [
[[package]]
name = "serde_spanned"
version = "1.1.1"
version = "1.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26"
checksum = "4e7523beb55eece201a2356bee0bbca0d1ab466c14c07703b2e0ee6d42cb0c2c"
dependencies = [
"serde_core",
]
@@ -679,6 +783,7 @@ dependencies = [
"fastrand",
"flume",
"icmp-socket2",
"irtt-client",
"libc",
"log",
"netlink-bindings",
@@ -697,6 +802,12 @@ version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "85ec932792e62b2d3936163eacbd2e09984453bcd114f665c1e1694e36ff59c9"
[[package]]
name = "subtle"
version = "2.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
[[package]]
name = "syn"
version = "2.0.119"
@@ -784,9 +895,9 @@ dependencies = [
[[package]]
name = "toml"
version = "1.1.3+spec-1.1.0"
version = "1.1.7+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "53c96ecdfa941c8fc4fcaed14f99ada8ebed502eef533015095a07e3301d4c3c"
checksum = "7874cd34dd99040fc24e30923aa64a60a81ff3ddfe2b8129ccc84ba24949deed"
dependencies = [
"indexmap",
"serde_core",
@@ -799,27 +910,27 @@ dependencies = [
[[package]]
name = "toml_datetime"
version = "1.1.1+spec-1.1.0"
version = "1.1.2+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3165f65f62e28e0115a00b2ebdd37eb6f3b641855f9d636d3cd4103767159ad7"
checksum = "2b86d767906c6c42421dcba507eb9d203e779497710a47782a224bb871653053"
dependencies = [
"serde_core",
]
[[package]]
name = "toml_parser"
version = "1.1.2+spec-1.1.0"
version = "1.1.4+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526"
checksum = "bed810ee007d8e7f9cf20b4774e7198142fa27186e0f805871da7f3c39eaf333"
dependencies = [
"winnow",
]
[[package]]
name = "toml_writer"
version = "1.1.2+spec-1.1.0"
version = "1.1.3+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2"
checksum = "06bdbd8cfc056b8d2e2e85f29b56a3bdbecb527cef81eb39e3e7b98af4652770"
[[package]]
name = "trybuild"
@@ -836,6 +947,12 @@ dependencies = [
"toml",
]
[[package]]
name = "typenum"
version = "1.20.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20"
[[package]]
name = "uci-derive"
version = "0.1.0"
@@ -848,9 +965,9 @@ dependencies = [
[[package]]
name = "unicode-ident"
version = "1.0.24"
version = "1.0.26"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954"
[[package]]
name = "wasi"
@@ -860,9 +977,9 @@ checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
[[package]]
name = "wasm-bindgen"
version = "0.2.126"
version = "0.2.129"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4"
checksum = "9bb54f33acc68fd454578d9820b0bde1a1a3d17aa17bb7b6595806d02886d409"
dependencies = [
"cfg-if",
"once_cell",
@@ -873,9 +990,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-macro"
version = "0.2.126"
version = "0.2.129"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1"
checksum = "2e29d0c35b16e224a7eeb5cd2d25e3e1968fbd65604117b44d3b789d00ee8535"
dependencies = [
"quote",
"wasm-bindgen-macro-support",
@@ -883,22 +1000,22 @@ dependencies = [
[[package]]
name = "wasm-bindgen-macro-support"
version = "0.2.126"
version = "0.2.129"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e"
checksum = "6f501a8bc3719dba86ef8ae4728879c08001bea749eb1333ac5b91e040e2a6b7"
dependencies = [
"bumpalo",
"proc-macro2",
"quote",
"syn 2.0.119",
"syn 3.0.6",
"wasm-bindgen-shared",
]
[[package]]
name = "wasm-bindgen-shared"
version = "0.2.126"
version = "0.2.129"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24"
checksum = "23f0c9c52aa7cd7d77769a4cfe2a9adb1b331f489a41d912ce14513d5ab995c6"
dependencies = [
"unicode-ident",
]
+4 -2
View File
@@ -25,6 +25,7 @@ csv = "1.4.0"
fastrand = "2"
flume = { version = "0.12.0", default-features = false, features = ["async", "eventual-fairness", "select"] }
icmp-socket2 = { version = "0.6.4", features = ["tokio"] }
irtt-client = { version = "0.6.0", optional = true, features = ["ancillary", "tokio"] }
libc = "0.2"
rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] }
serde = { version = "1", features = ["derive"] }
@@ -41,10 +42,11 @@ netlink-bindings = { version = "0.3.3", features = ["rt-link", "tc"] }
netlink-socket2 = { version = "0.3.3" }
[target.'cfg(unix)'.dependencies]
rust-uci = { git = "https://github.com/Lochnair/rust-uci.git", rev = "fcfedd204fb76928ef617bf9631b041c816e03d4", optional = true, default-features = false, }
rust-uci = { git = "https://github.com/Lochnair/rust-uci.git", rev = "6835397f4f4ba8b69e75e727105186865c357417", optional = true, default-features = false, }
[features]
default = ["toml", "uci-vendored"]
default = ["irtt", "toml", "uci-vendored"]
irtt = ["dep:irtt-client"]
toml = ["config/toml"]
uci = ["dep:rust-uci"]
uci-vendored = ["uci", "rust-uci/vendored"]
+44 -19
View File
@@ -9,13 +9,13 @@ use syn::{Data, DeriveInput, Field, Fields, FieldsNamed, LitStr, Token, parse_ma
/// Generates `UciSectionSchema` from a settings struct.
///
/// Every named field is mapped one-to-one:
/// Every named field without `#[uci(skip)]` is mapped one-to-one:
///
/// ```text
/// Rust field name -> UCI option name
/// Rust field name -> config-rs field name
/// ```
#[proc_macro_derive(UciSection)]
#[proc_macro_derive(UciSection, attributes(uci))]
pub fn derive_uci_section(input: TokenStream) -> TokenStream {
let input = parse_macro_input!(input as DeriveInput);
@@ -56,26 +56,28 @@ fn expand_uci_section(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStre
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 mut mappings = Vec::with_capacity(fields.named.len());
let field_name = field_ident.unraw().to_string();
let field_name = LitStr::new(&field_name, field_ident.span());
for field in &fields.named {
if parse_skip(field)? {
continue;
}
quote! {
crate::settings::uci_schema::UciOptionMapping {
uci_option: #field_name,
config_field: #field_name,
}
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());
mappings.push(quote! {
crate::settings::uci_schema::UciOptionMapping {
uci_option: #field_name,
config_field: #field_name,
}
})
.collect::<Vec<_>>();
});
}
let (impl_generics, type_generics, where_clause) = input.generics.split_for_impl();
@@ -95,6 +97,29 @@ fn expand_uci_section(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStre
})
}
fn parse_skip(field: &Field) -> syn::Result<bool> {
let mut skip = false;
for attribute in &field.attrs {
if !attribute.path().is_ident("uci") {
continue;
}
attribute.parse_nested_meta(|meta| {
if !meta.path.is_ident("skip") {
return Err(meta.error("expected `skip`"));
}
if skip {
return Err(meta.error("duplicate `skip` attribute"));
}
skip = true;
Ok(())
})?;
}
Ok(skip)
}
fn expand_uci_config(input: &DeriveInput) -> syn::Result<proc_macro2::TokenStream> {
let struct_name = &input.ident;
let fields = require_named_fields(input)?;
+11 -4
View File
@@ -22,6 +22,7 @@ mod util;
use crate::baseliner::Baseliner;
use crate::controller::Controller;
use crate::measurement::icmp::IcmpEngine;
use crate::measurement::irtt::IrttEngine;
use crate::metrics::{Metric, Metrics, MetricsSender};
use crate::platform::{TrafficControlBackend, traffic_control_backend, warn_platform_limitations};
use crate::ratecontroller::Ratecontroller;
@@ -147,7 +148,7 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
IpAddr::from_str("94.140.14.14")?,
];
let reselection_enabled = configured_count > settings.advanced_settings.num_reflectors as usize;
let reselection_enabled = configured_count > settings.reflectors.num_reflectors as usize;
let pool = if reselection_enabled {
configured
} else {
@@ -236,6 +237,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
let (control_snapshot_tx, control_snapshot_rx) = flume::unbounded();
let (error_tx, error_rx) = flume::unbounded::<anyhow::Error>();
let (icmp_binding_tx, icmp_binding_rx) = flume::unbounded();
let (irtt_binding_tx, irtt_binding_rx) = flume::unbounded();
let (measurement_tx, measurement_rx) = flume::unbounded();
let (reflector_tx, reflector_rx) = flume::unbounded();
let (reselect_tx, reselect_rx) = flume::bounded(1);
@@ -247,17 +249,22 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
};
let controller = Controller::new(reflector_rx, icmp_binding_tx);
let engine = {
let icmp_engine = {
let _guard = rt.enter();
IcmpEngine::new(
start_time,
&settings.network,
measurement_tx,
measurement_tx.clone(),
icmp_binding_rx,
)?
};
let (_irtt_task, irtt_engine) = {
let _guard = rt.enter();
IrttEngine::new(start_time, measurement_tx, irtt_binding_rx)?
};
spawn_task(&rt, &error_tx, engine.run());
spawn_task(&rt, &error_tx, icmp_engine.run());
spawn_task(&rt, &error_tx, irtt_engine.run());
let dropped = Arc::new(AtomicU32::new(0));
+343
View File
@@ -0,0 +1,343 @@
use std::{collections::HashMap, net::SocketAddr, time::Instant};
use flume::{Receiver, Sender};
use irtt_client::{
ClientConfig, ClientEvent,
managed::{
ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask,
ManagedCommandAcknowledgement, ManagedCommandApplyError, ManagedCommandError,
ManagedCompletionPolicy, ManagedConfigError, ManagedEndReason, ManagedEvent,
ManagedLifecycle, ManagedSubscribeError, ManagedTargetConfig, ManagedTargetEndReason,
TargetId,
},
};
use log::{debug, trace, warn};
use thiserror::Error;
use tokio::sync::broadcast::error::RecvError;
use crate::measurement::model::{
MeasurementEvent, MeasurementLoss, MeasurementObservation, MeasurementStream,
MeasurementStreamId, OneWayClock, OneWayLatency, ProbeSchedule, SignedDuration,
};
#[derive(Debug, Error)]
pub enum IrttError {
#[error("failed to build IRTT config")]
InvalidConfig(#[from] ManagedConfigError),
#[error("event channel closed")]
EventChannelClosed,
#[error("binding channel closed")]
BindingChannelClosed,
#[error("error while reciving on channel")]
ChannelRecvError(#[from] RecvError),
#[error("unable to subscribe to irtt events")]
SubscriptionError(#[from] ManagedSubscribeError),
#[error("failed to submit IRTT target update")]
CommandError(#[from] ManagedCommandError),
#[error("failed to apply IRTT target update")]
CommandApplyError(#[from] ManagedCommandApplyError),
#[error("IRTT managed client failed: {0:?}")]
ManagedFailure(ManagedEndReason),
#[error("IRTT managed task was abandoned")]
ManagedAbandoned,
}
pub struct IrttBinding {
stream: MeasurementStream,
schedule: ProbeSchedule,
port: u16,
}
impl IrttBinding {
pub fn new(stream: MeasurementStream, schedule: ProbeSchedule, port: u16) -> Self {
Self {
stream,
schedule,
port,
}
}
}
pub struct IrttEngine {
client: ManagedClientHandle,
bindings: HashMap<MeasurementStreamId, IrttBinding>,
origin: Instant,
binding_rx: Receiver<Vec<IrttBinding>>,
observation_tx: Sender<MeasurementEvent>,
}
impl IrttEngine {
pub fn new(
origin: Instant,
observation_tx: Sender<MeasurementEvent>,
binding_rx: Receiver<Vec<IrttBinding>>,
) -> Result<(ManagedClientTask, Self), IrttError> {
let config = ManagedClientConfig {
client: ClientConfig {
request: irtt_client::SessionRequest {
duration: None,
..Default::default()
},
..Default::default()
},
completion: ManagedCompletionPolicy::ExplicitStop,
..Default::default()
};
let (task, handle) = ManagedClient::task(config, Vec::new())?;
Ok((
task,
Self {
client: handle,
bindings: HashMap::new(),
origin,
binding_rx,
observation_tx,
},
))
}
async fn set_bindings(&mut self, bindings: Vec<IrttBinding>) -> Result<(), IrttError> {
let targets = bindings
.iter()
.map(|binding| {
ManagedTargetConfig::new(
TargetId::new(binding.stream.id.0.to_string()),
SocketAddr::new(binding.stream.peer, binding.port).to_string(),
)
})
.collect();
self.client.update_targets(targets)?.await?;
self.bindings = bindings
.into_iter()
.map(|binding| (binding.stream.id, binding))
.collect();
Ok(())
}
pub async fn run(mut self) -> Result<(), IrttError> {
let mut events = self.client.subscribe()?;
loop {
tokio::select! {
bindings = self.binding_rx.recv_async() => {
let bindings = bindings
.map_err(|_| IrttError::BindingChannelClosed)?;
self.set_bindings(bindings).await?;
}
event = events.recv() => {
let event = match event {
Ok(event) => event,
Err(RecvError::Lagged(n)) => {
warn!("IRTT event receiver dropped {n} events");
continue;
}
Err(RecvError::Closed) => {
let status = self.client.status();
match status.lifecycle {
ManagedLifecycle::Completed => return Ok(()),
ManagedLifecycle::Failed => {
if let Some(outcome) = &status.final_outcome {
return Err(IrttError::ManagedFailure(
outcome.end_reason.clone()
));
}
return Err(IrttError::ManagedAbandoned);
}
_ => return Err(IrttError::ManagedAbandoned),
}
}
};
match event {
ManagedEvent::Started => {
debug!("IRTT managed client started");
}
ManagedEvent::TargetStateChanged {
target,
lifecycle,
} => {
debug!("IRTT target {target:?}: {lifecycle:?}");
}
ManagedEvent::Client { target, event } => {
let stream = target.id.as_str()
.parse::<u64>()
.ok()
.and_then(|id| {
self.bindings.get(&MeasurementStreamId(id))
})
.map(|binding| binding.stream);
match event {
ClientEvent::SessionStarted(session) => {
debug!(
"IRTT session started for {}: {}",
target.id,
session.remote
);
}
ClientEvent::NoTestCompleted(outcome) => {
debug!(
"IRTT no-test completed for {}: {}",
target.id,
outcome.remote
);
}
ClientEvent::SessionClosed { .. } => {
debug!("IRTT session closed for {}", target.id);
}
ClientEvent::EchoSent { seq, .. } => {
trace!("IRTT {} sent probe {seq}", target.id);
}
ClientEvent::EchoReply {
sent_at,
received_at,
one_way,
..
} => {
let Some(stream) = stream else {
continue;
};
let observation = MeasurementObservation {
stream,
started_at: sent_at.mono,
observed_at: received_at.mono,
one_way: one_way.and_then(map_one_way),
};
self.observation_tx
.send_async(MeasurementEvent::Observation(observation))
.await
.map_err(|_| IrttError::EventChannelClosed)?;
}
ClientEvent::EchoLoss {
sent_at,
timeout_at,
..
} => {
let Some(stream) = stream else {
continue;
};
let loss = MeasurementLoss {
stream,
started_at: sent_at.mono,
detected_at: timeout_at,
};
self.observation_tx
.send_async(MeasurementEvent::Loss(loss))
.await
.map_err(|_| IrttError::EventChannelClosed)?;
}
ClientEvent::DuplicateReply { seq, .. } => {
trace!(
"IRTT {} duplicate reply {seq}",
target.id
);
}
ClientEvent::LateReply { seq, .. } => {
trace!(
"IRTT {} late reply {seq}",
target.id
);
}
ClientEvent::Warning { kind, message, .. } => {
debug!(
"IRTT {} warning ({kind:?}): {message}",
target.id
);
}
}
}
ManagedEvent::TargetFinished { outcome } => {
match &outcome.end_reason {
ManagedTargetEndReason::Failed(failure) => {
warn!(
"IRTT target {} failed: {failure:?}",
outcome.target.id
);
}
reason => {
debug!(
"IRTT target {} finished: {reason:?}",
outcome.target.id
);
}
}
}
ManagedEvent::Stopping => {
debug!("IRTT managed client stopping");
}
ManagedEvent::Completed { outcome } => {
debug!(
"IRTT managed client completed: {:?}",
outcome.end_reason
);
return Ok(());
}
ManagedEvent::Failed { outcome } => {
return Err(IrttError::ManagedFailure(
outcome.end_reason.clone()
));
}
ManagedEvent::Abandoned => {
return Err(IrttError::ManagedAbandoned);
}
}
}
}
}
}
}
fn map_one_way(sample: irtt_client::OneWayDelaySample) -> Option<OneWayLatency> {
let uplink = i64::try_from(sample.client_to_server?.as_nanos()).ok()?;
let downlink = i64::try_from(sample.server_to_client?.as_nanos()).ok()?;
Some(OneWayLatency {
uplink: SignedDuration::from_nanos(uplink),
downlink: SignedDuration::from_nanos(downlink),
clock: OneWayClock::Uncalibrated,
})
}
+2
View File
@@ -1,2 +1,4 @@
pub(crate) mod icmp;
#[cfg(feature = "irtt")]
pub(crate) mod irtt;
pub(crate) mod model;
+3 -5
View File
@@ -159,10 +159,8 @@ impl ReflectorSelector {
// After 40 reselections, slow down to every 15 minutes
if reselection_count > 40 {
selector_sleep_time = Duration::new(
self.settings.advanced_settings.peer_reselection_time * 60,
0,
);
selector_sleep_time =
Duration::new(self.settings.reflectors.peer_reselection_time * 60, 0);
}
let mut next_peers: Vec<IpAddr> = Vec::new();
@@ -222,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.settings.advanced_settings.num_reflectors;
let mut num_reflectors = self.settings.reflectors.num_reflectors;
let candidate_pool_num = (2 * num_reflectors) as usize;
candidates.truncate(candidate_pool_num);
+55 -55
View File
@@ -78,28 +78,32 @@ where
deserializer.deserialize_any(FlexiBoolVisitor)
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum MeasurementType {
Icmp = 1,
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Hash)]
#[serde(rename_all = "kebab-case")]
pub enum ReflectorProtocol {
IcmpEcho = 1,
IcmpTimestamps,
Irtt,
}
impl Display for MeasurementType {
impl Display for ReflectorProtocol {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
MeasurementType::Icmp => write!(f, "icmp"),
MeasurementType::IcmpTimestamps => write!(f, "icmp-timestamps"),
ReflectorProtocol::IcmpEcho => write!(f, "icmp"),
ReflectorProtocol::IcmpTimestamps => write!(f, "icmp-timestamps"),
ReflectorProtocol::Irtt => write!(f, "irtt"),
}
}
}
impl FromStr for MeasurementType {
impl FromStr for ReflectorProtocol {
type Err = ConfigError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s.to_lowercase().as_str() {
"icmp" => Ok(MeasurementType::Icmp),
"icmp-timestamps" | "icmp_timestamps" => Ok(MeasurementType::IcmpTimestamps),
"icmp" => Ok(ReflectorProtocol::IcmpEcho),
"icmp-timestamps" | "icmp_timestamps" => Ok(ReflectorProtocol::IcmpTimestamps),
"irtt" => Ok(ReflectorProtocol::Irtt),
_ => Err(ConfigError::Message(format!(
"invalid measurement type: {s}"
))),
@@ -127,12 +131,13 @@ impl FromStr for ObservabilityProtocol {
}
}
#[derive(Debug, Deserialize)]
struct ReflectorRecord {
reflector_ip: IpAddr,
ip_version: u8,
#[allow(dead_code)]
description: String,
#[derive(Debug, Clone, Deserialize, UciSection)]
pub(crate) struct ReflectorDefinition {
pub(crate) host: String,
pub(crate) country: Option<String>,
pub(crate) protocols: Vec<ReflectorProtocol>,
pub(crate) irtt_port: Option<u16>,
pub(crate) irtt_hmac: Option<String>,
}
#[derive(Clone, Debug, Deserialize, UciConfig)]
@@ -141,6 +146,10 @@ pub(crate) struct Settings {
#[uci(section)]
pub(crate) network: NetworkSettings,
#[serde(default)]
#[uci(section)]
pub(crate) reflectors: ReflectorSettings,
#[serde(default)]
#[uci(section)]
pub(crate) output: OutputSettings,
@@ -173,40 +182,8 @@ impl Settings {
}
pub fn load_reflectors(&self) -> anyhow::Result<Vec<IpAddr>> {
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::<ReflectorRecord>() {
let record = record.with_context(|| {
format!(
"failed to parse reflector list: {}",
self.advanced_settings.reflector_list_file
)
})?;
match (record.ip_version, record.reflector_ip) {
(4, IpAddr::V4(_)) => {}
(6, address @ IpAddr::V6(_)) => {
bail!("reflector {address} is IPv6; only IPv4 reflectors are supported");
}
(version, address) => {
bail!("reflector {address} has inconsistent IP version {version}");
}
}
reflectors.push(record.reflector_ip);
}
Ok(reflectors)
}
}
@@ -240,6 +217,35 @@ impl NetworkSettings {
}
}
#[derive(Clone, Debug, Deserialize, UciSection)]
#[serde(default)]
pub(crate) struct ReflectorSettings {
pub(crate) files: Vec<String>,
pub(crate) countries: Vec<String>,
#[uci(skip)]
pub(crate) entries: Vec<ReflectorDefinition>,
#[uci(skip)]
pub(crate) bootstrap: Vec<ReflectorDefinition>,
pub(crate) num_reflectors: u8,
pub(crate) peer_reselection_time: u64,
}
impl Default for ReflectorSettings {
fn default() -> Self {
Self {
files: vec!["/etc/sqm-autorate-rust/reflectors.csv".into()],
countries: Vec::new(),
entries: Vec::new(),
bootstrap: Vec::new(),
num_reflectors: 5,
peer_reselection_time: 15,
}
}
}
#[derive(Clone, Debug, Deserialize, UciSection)]
#[serde(default)]
pub(crate) struct OutputSettings {
@@ -309,10 +315,7 @@ pub(crate) struct AdvancedSettings {
pub(crate) high_load_level: f64,
pub(crate) min_change_interval: f64,
#[serde(deserialize_with = "deserialize_from_str")]
pub(crate) measurement_type: MeasurementType,
pub(crate) num_reflectors: u8,
pub(crate) peer_reselection_time: u64,
pub(crate) reflector_list_file: String,
pub(crate) measurement_type: ReflectorProtocol,
pub(crate) speed_hist_size: u32,
pub(crate) tick_interval: f64,
pub(crate) upload_delay_ms: f64,
@@ -326,10 +329,7 @@ impl Default for AdvancedSettings {
download_delay_ms: 15.0,
high_load_level: 0.8,
min_change_interval: 0.5,
measurement_type: MeasurementType::IcmpTimestamps,
num_reflectors: 5,
peer_reselection_time: 15,
reflector_list_file: "/etc/sqm-autorate/reflectors-icmp.csv".to_owned(),
measurement_type: ReflectorProtocol::IcmpTimestamps,
speed_hist_size: 100,
tick_interval: 0.5,
upload_delay_ms: 15.0,
+68 -6
View File
@@ -1,6 +1,6 @@
use std::path::{Path, PathBuf};
use crate::settings::uci_schema::{UciConfigSchema, UciSectionMapping};
use crate::settings::uci_schema::{UciConfigSchema, UciRepeatedSectionMapping, UciSectionMapping};
use config::{ConfigError as SourceError, Map, Source, Value, ValueKind};
use rust_uci::config::{
@@ -18,6 +18,7 @@ pub(crate) struct UciSource {
packages: PackageSelection,
required: bool,
schema: &'static [UciSectionMapping],
repeated_schema: &'static [UciRepeatedSectionMapping],
config_dir: Option<PathBuf>,
save_dir: Option<PathBuf>,
}
@@ -28,6 +29,7 @@ impl UciSource {
packages,
required: true,
schema: &[],
repeated_schema: &[],
config_dir: None,
save_dir: None,
}
@@ -65,6 +67,17 @@ impl UciSource {
self
}
/// Registers section types collected as arrays of tables at their config paths.
/// Absent section types leave the destination unset for Serde/source defaults.
#[allow(dead_code)]
pub(crate) fn with_repeated_schema(
mut self,
schema: &'static [UciRepeatedSectionMapping],
) -> Self {
self.repeated_schema = schema;
self
}
#[allow(dead_code)]
pub(crate) fn with_directories(
mut self,
@@ -145,6 +158,7 @@ impl UciSource {
package_name: &str,
package: &UciPackage,
values: &mut Map<String, Value>,
repeated_values: &mut Map<String, Vec<Value>>,
) -> Result<(), SourceError> {
let sections = package.sections().map_err(|e| {
SourceError::Message(format!(
@@ -154,6 +168,12 @@ impl UciSource {
})?;
for section in sections {
let repeated_mapping = self
.repeated_schema
.iter()
.find(|mapping| mapping.uci_section == section.type_());
let mut section_values = Map::new();
let selector = section.selector().ok_or_else(|| {
SourceError::Message(format!(
"loaded UCI section of type {:?} in package \
@@ -181,9 +201,21 @@ impl UciSource {
let source_key = format!("{package_name}.{config_relative_key}");
let destination_key = self
.mapped_config_key(&selector, option_name)
.unwrap_or(source_key);
let repeated_option = repeated_mapping.and_then(|mapping| {
mapping
.options
.iter()
.find(|option| option.uci_option == option_name)
});
let destination_key = if let Some(mapping) = repeated_option {
mapping.config_field.to_owned()
} else if repeated_mapping.is_some() {
option_name.to_owned()
} else {
self.mapped_config_key(&selector, option_name)
.unwrap_or(source_key)
};
let Some(value) = option.get().map_err(|e| {
SourceError::Message(format!(
@@ -210,13 +242,27 @@ impl UciSource {
}
};
if values.insert(destination_key.clone(), value).is_some() {
let destination = if repeated_mapping.is_some() {
&mut section_values
} else {
&mut *values
};
if destination.insert(destination_key.clone(), value).is_some() {
return Err(SourceError::Message(format!(
"duplicate UCI destination key \
{destination_key:?} while mapping {uci_key:?}"
)));
}
}
if let Some(mapping) = repeated_mapping {
let origin = format!("UCI section {package_name}.{selector}");
repeated_values
.entry(mapping.config_path.to_owned())
.or_default()
.push(Value::new(Some(&origin), ValueKind::Table(section_values)));
}
}
Ok(())
@@ -272,6 +318,7 @@ impl Source for UciSource {
};
let mut values = Map::new();
let mut repeated_values = Map::new();
for package in packages {
let package_name = package
@@ -279,7 +326,22 @@ impl Source for UciSource {
.map_err(|e| SourceError::Message(format!("failed to read UCI package name: {e}")))?
.to_owned();
self.collect_package(&package_name, &package, &mut values)?;
self.collect_package(&package_name, &package, &mut values, &mut repeated_values)?;
}
for (config_path, sections) in repeated_values {
let origin = "UCI repeated sections".to_owned();
if values
.insert(
config_path.clone(),
Value::new(Some(&origin), ValueKind::Array(sections)),
)
.is_some()
{
return Err(SourceError::Message(format!(
"duplicate UCI destination key {config_path:?} while mapping repeated sections"
)));
}
}
Ok(values)
+12
View File
@@ -19,6 +19,18 @@ pub(crate) struct UciSectionMapping {
pub(crate) options: &'static [UciOptionMapping],
}
#[cfg(all(feature = "uci", unix))]
#[derive(Clone, Copy, Debug)]
pub(crate) struct UciRepeatedSectionMapping {
/// UCI section type, for both named and anonymous sections.
pub(crate) uci_section: &'static str,
/// Config-rs / Serde path receiving an array of section tables.
pub(crate) config_path: &'static str,
pub(crate) options: &'static [UciOptionMapping],
}
pub(crate) trait UciSectionSchema {
const UCI_OPTIONS: &'static [UciOptionMapping];
}
+42 -18
View File
@@ -3,6 +3,8 @@
// SPDX-License-Identifier: MPL-2.0
//
use crate::settings::ReflectorSettings;
use super::{AdvancedSettings, NetworkSettings, ObservabilitySettings, OutputSettings, Settings};
use std::error::Error;
use std::fmt::{self, Display, Formatter};
@@ -88,6 +90,7 @@ impl Settings {
let mut report = ValidationReport::default();
self.network.validate(&mut report);
self.reflectors.validate(&mut report);
self.output.validate(&mut report);
self.observability.validate(&mut report);
self.advanced_settings.validate(&mut report);
@@ -192,6 +195,45 @@ impl NetworkSettings {
}
}
impl ReflectorSettings {
fn validate(&self, report: &mut ValidationReport) {
if self.num_reflectors == 0 {
report.error("reflectors.num_reflectors", "must be greater than 0");
}
if self.peer_reselection_time == 0 {
report.error("reflectors.peer_reselection_time", "must be greater than 0");
}
if self.files.is_empty() && self.entries.is_empty() {
report.error(
"reflectors",
"at least one reflector file or inline entry is required",
);
}
for (index, file) in self.files.iter().enumerate() {
if file.trim().is_empty() {
report.error(
"reflectors.files",
format!("entry {index} must not be empty"),
);
}
}
for (index, country) in self.countries.iter().enumerate() {
if country.len() != 2 || !country.bytes().all(|c| c.is_ascii_alphabetic()) {
report.error(
"reflectors.countries",
format!(
"entry {index} ({country:?}) must be an ISO 3166-1 alpha-2 country code"
),
);
}
}
}
}
impl OutputSettings {
fn validate(&self, report: &mut ValidationReport) {
if self.suppress_statistics {
@@ -298,24 +340,6 @@ impl AdvancedSettings {
self.tick_interval,
);
if self.num_reflectors < 5 {
report.error(
"advanced_settings.num_reflectors",
format!("must be at least 5, got {}", self.num_reflectors,),
);
}
if self.peer_reselection_time == 0 {
report.error(
"advanced_settings.peer_reselection_time",
"must be greater than 0",
);
}
if self.reflector_list_file.trim().is_empty() {
report.error("advanced_settings.reflector_list_file", "must not be empty");
}
if self.speed_hist_size == 0 {
report.error(
"advanced_settings.speed_hist_size",