Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Nils Andreas Svee
|
4b9e510ab3 | ||
|
Nils Andreas Svee
|
03b9c72ef6 | ||
|
Nils Andreas Svee
|
8dbae6df91 |
No files matched your search
Generated
+117
@@ -60,6 +60,15 @@ version = "2.13.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06"
|
||||
|
||||
[[package]]
|
||||
name = "block-buffer"
|
||||
version = "0.12.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa"
|
||||
dependencies = [
|
||||
"hybrid-array",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "bumpalo"
|
||||
version = "3.20.3"
|
||||
@@ -123,6 +132,12 @@ dependencies = [
|
||||
"cc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "cmov"
|
||||
version = "0.5.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a"
|
||||
|
||||
[[package]]
|
||||
name = "config"
|
||||
version = "0.15.27"
|
||||
@@ -135,6 +150,21 @@ dependencies = [
|
||||
"winnow",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "const-oid"
|
||||
version = "0.10.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c"
|
||||
|
||||
[[package]]
|
||||
name = "crypto-common"
|
||||
version = "0.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453"
|
||||
dependencies = [
|
||||
"hybrid-array",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "csv"
|
||||
version = "1.4.0"
|
||||
@@ -156,6 +186,27 @@ dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ctutils"
|
||||
version = "0.4.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "03bb0e1cc970d482d121d9a1744999169b69a07470b3d644a7894e53fcaf4574"
|
||||
dependencies = [
|
||||
"cmov",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "digest"
|
||||
version = "0.11.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2"
|
||||
dependencies = [
|
||||
"block-buffer",
|
||||
"const-oid",
|
||||
"crypto-common",
|
||||
"ctutils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "either"
|
||||
version = "1.19.0"
|
||||
@@ -242,6 +293,24 @@ version = "0.17.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
|
||||
|
||||
[[package]]
|
||||
name = "hmac"
|
||||
version = "0.13.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f"
|
||||
dependencies = [
|
||||
"digest",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hybrid-array"
|
||||
version = "0.4.15"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17"
|
||||
dependencies = [
|
||||
"typenum",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "icmp-socket2"
|
||||
version = "0.6.4"
|
||||
@@ -265,6 +334,31 @@ dependencies = [
|
||||
"hashbrown",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "irtt-client"
|
||||
version = "0.6.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a59c24704bd94f19b84b2b0190aa9163efede499f8cf4eb02922443cfee08339"
|
||||
dependencies = [
|
||||
"irtt-proto",
|
||||
"nix",
|
||||
"socket2",
|
||||
"thiserror",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "irtt-proto"
|
||||
version = "0.5.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a56ff307983b8f6477724e08cf726c5e24975b9affeb853d0830d91c60026ab2"
|
||||
dependencies = [
|
||||
"hmac",
|
||||
"md-5",
|
||||
"subtle",
|
||||
"thiserror",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "itertools"
|
||||
version = "0.13.0"
|
||||
@@ -337,6 +431,16 @@ version = "0.4.34"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
|
||||
|
||||
[[package]]
|
||||
name = "md-5"
|
||||
version = "0.11.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"digest",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "memchr"
|
||||
version = "2.8.3"
|
||||
@@ -679,6 +783,7 @@ dependencies = [
|
||||
"fastrand",
|
||||
"flume",
|
||||
"icmp-socket2",
|
||||
"irtt-client",
|
||||
"libc",
|
||||
"log",
|
||||
"netlink-bindings",
|
||||
@@ -697,6 +802,12 @@ version = "1.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "85ec932792e62b2d3936163eacbd2e09984453bcd114f665c1e1694e36ff59c9"
|
||||
|
||||
[[package]]
|
||||
name = "subtle"
|
||||
version = "2.6.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
|
||||
|
||||
[[package]]
|
||||
name = "syn"
|
||||
version = "2.0.119"
|
||||
@@ -836,6 +947,12 @@ dependencies = [
|
||||
"toml",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "typenum"
|
||||
version = "1.20.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20"
|
||||
|
||||
[[package]]
|
||||
name = "uci-derive"
|
||||
version = "0.1.0"
|
||||
|
||||
+3
-1
@@ -25,6 +25,7 @@ csv = "1.4.0"
|
||||
fastrand = "2"
|
||||
flume = { version = "0.12.0", default-features = false, features = ["async", "eventual-fairness", "select"] }
|
||||
icmp-socket2 = { version = "0.6.4", features = ["tokio"] }
|
||||
irtt-client = { version = "0.6.0", optional = true, features = ["ancillary", "tokio"] }
|
||||
libc = "0.2"
|
||||
rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] }
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
@@ -44,7 +45,8 @@ netlink-socket2 = { version = "0.3.5" }
|
||||
rust-uci = { git = "https://github.com/Lochnair/rust-uci.git", rev = "6835397f4f4ba8b69e75e727105186865c357417", optional = true, default-features = false }
|
||||
|
||||
[features]
|
||||
default = ["toml", "uci-vendored"]
|
||||
default = ["irtt", "toml", "uci-vendored"]
|
||||
irtt = ["dep:irtt-client"]
|
||||
toml = ["config/toml"]
|
||||
uci = ["dep:rust-uci"]
|
||||
uci-vendored = ["uci", "rust-uci/vendored"]
|
||||
|
||||
@@ -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
@@ -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));
|
||||
|
||||
|
||||
@@ -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,
|
||||
})
|
||||
}
|
||||
@@ -1,2 +1,4 @@
|
||||
pub(crate) mod icmp;
|
||||
#[cfg(feature = "irtt")]
|
||||
pub(crate) mod irtt;
|
||||
pub(crate) mod model;
|
||||
@@ -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
@@ -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
@@ -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)
|
||||
|
||||
@@ -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
@@ -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",
|
||||
|
||||
Reference in new issue
Block a user