From da3487e5068f53a4a5355f301390d3e5eb76baca Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 13 Apr 2026 15:22:15 +0700 Subject: [PATCH] feat(observability): make metrics collection optional and granular Wrap metrics senders in Option and conditionally initialize the monitoring thread to support granular configuration and reduced overhead. --- src/baseliner.rs | 35 +++++++++++--------- src/main.rs | 70 +++++++++++++++++++++++---------------- src/pinger.rs | 20 ++++++----- src/ratecontroller.rs | 4 +-- src/reflector_selector.rs | 2 +- 5 files changed, 76 insertions(+), 55 deletions(-) diff --git a/src/baseliner.rs b/src/baseliner.rs index df19941..51cfc41 100644 --- a/src/baseliner.rs +++ b/src/baseliner.rs @@ -25,7 +25,8 @@ pub struct Baseliner { pub reselect_trigger: Sender, pub start_time: Instant, pub stats_rx: Receiver, - pub metrics_tx: SyncSender, + pub baseline_metrics_tx: Option>, + pub event_metrics_tx: Option>, } fn ewma_factor(tick: f64, dur: f64) -> f64 { @@ -107,12 +108,14 @@ impl Baseliner { "Reflector {} has OWD > 5 seconds more than baseline, triggering reselection", time_data.reflector ); - let _ = self.metrics_tx.try_send(Metric::Event { - name: "reselection", - reason: "anomaly", - reflector: Some(time_data.reflector), - timestamp_ns: clock.as_nanos(), - }); + if let Some(ref tx) = self.event_metrics_tx { + let _ = tx.try_send(Metric::Event { + name: "reselection", + reason: "anomaly", + reflector: Some(time_data.reflector), + timestamp_ns: clock.as_nanos(), + }); + } // If reselection is disabled this would trigger an error // so just ignore the result let _ = self.reselect_trigger.send(true); @@ -136,14 +139,16 @@ impl Baseliner { } } - let _ = self.metrics_tx.try_send(Metric::Baseline { - reflector: time_data.reflector, - baseline_up_ewma: owd_baseline.up_ewma, - baseline_down_ewma: owd_baseline.down_ewma, - recent_up_ewma: owd_recent.up_ewma, - recent_down_ewma: owd_recent.down_ewma, - timestamp_ns: clock.as_nanos(), - }); + if let Some(ref tx) = self.baseline_metrics_tx { + let _ = tx.try_send(Metric::Baseline { + reflector: time_data.reflector, + baseline_up_ewma: owd_baseline.up_ewma, + baseline_down_ewma: owd_baseline.down_ewma, + recent_up_ewma: owd_recent.up_ewma, + recent_down_ewma: owd_recent.down_ewma, + timestamp_ns: clock.as_nanos(), + }); + } info!( "Reflector {} up baseline = {} down baseline = {}", diff --git a/src/main.rs b/src/main.rs index ab3a837..f893d76 100644 --- a/src/main.rs +++ b/src/main.rs @@ -82,19 +82,32 @@ fn main() -> anyhow::Result<()> { let (baseliner_stats_tx, baseliner_stats_rx) = channel(); let (reselect_tx, reselect_rx) = channel(); - let (metrics_tx, metrics_rx) = sync_channel(1000); + let (metrics_tx, metrics_rx) = if config.observability_enabled { + let (tx, rx) = sync_channel(1000); + (Some(tx), Some(rx)) + } else { + (None, None) + }; - let ping_tx = config - .observability_export_ping_metrics - .then(|| metrics_tx.clone()); + let ping_metrics_tx = metrics_tx + .as_ref() + .filter(|_| config.observability_export_ping_metrics) + .cloned(); - let baseline_tx = config - .observability_export_baseline_metrics - .then(|| metrics_tx.clone()); + let baseline_metrics_tx = metrics_tx + .as_ref() + .filter(|_| config.observability_export_baseline_metrics) + .cloned(); - let event_tx = config - .observability_export_events - .then(|| metrics_tx.clone()); + let event_metrics_tx = metrics_tx + .as_ref() + .filter(|_| config.observability_export_events) + .cloned(); + + let rate_metrics_tx = metrics_tx + .as_ref() + .filter(|_| config.observability_export_rate_metrics) + .cloned(); let (mut ping_listener, mut ping_sender) = match config.measurement_type { MeasurementType::Icmp => ( @@ -117,7 +130,8 @@ fn main() -> anyhow::Result<()> { reselect_trigger: reselect_tx.clone(), start_time: start_t, stats_rx: baseliner_stats_rx, - metrics_tx: metrics_tx.clone(), + baseline_metrics_tx, + event_metrics_tx: event_metrics_tx.clone(), }; let down_qdisc = Netlink::qdisc_from_ifname(config.download_interface.as_str())?; @@ -146,7 +160,6 @@ fn main() -> anyhow::Result<()> { let (error_tx, error_rx) = channel::(); let err_tx = error_tx.clone(); - let metrics_tx_clone = metrics_tx.clone(); let reflector_peers_lock_clone = reflector_peers_lock.clone(); thread::Builder::new() .name("receiver".to_string()) @@ -156,7 +169,7 @@ fn main() -> anyhow::Result<()> { config.measurement_type, reflector_peers_lock_clone, baseliner_stats_tx, - metrics_tx_clone, + ping_metrics_tx, ) { let _ = err_tx.send(e); } @@ -193,7 +206,7 @@ fn main() -> anyhow::Result<()> { reflector_peers_lock: reflector_peers_lock.clone(), reflector_pool, trigger_channel: reselect_rx, - metrics_tx: metrics_tx.clone(), + metrics_tx: event_metrics_tx, }; let err_tx = error_tx.clone(); thread::Builder::new() @@ -231,7 +244,7 @@ fn main() -> anyhow::Result<()> { reselect_tx, dl_direction, ul_direction, - metrics_tx.clone(), + rate_metrics_tx, )?; debug!( @@ -253,19 +266,20 @@ fn main() -> anyhow::Result<()> { } })?; - let metrics = Metrics { - config, - metrics_rx, - }; - - let err_tx = error_tx.clone(); - thread::Builder::new() - .name("metrics".to_string()) - .spawn(move || { - if let Err(e) = metrics.run() { - let _ = err_tx.send(e); - } - })?; + if let Some(rx) = metrics_rx { + let metrics = Metrics { + config, + metrics_rx: rx, + }; + let err_tx = error_tx.clone(); + thread::Builder::new() + .name("metrics".to_string()) + .spawn(move || { + if let Err(e) = metrics.run() { + let _ = err_tx.send(e); + } + })?; + } // Drop original sender so recv() unblocks if all threads exit cleanly drop(error_tx); diff --git a/src/pinger.rs b/src/pinger.rs index 8f8409e..745120f 100644 --- a/src/pinger.rs +++ b/src/pinger.rs @@ -55,7 +55,7 @@ pub trait PingListener { type_: MeasurementType, reflectors_lock: Arc>>, stats_tx: Sender, - metrics_tx: SyncSender, + ping_metrics_tx: Option>, ) -> anyhow::Result<()> { let socket = &mut open_socket(type_)?; @@ -84,14 +84,16 @@ pub trait PingListener { } }; - let _ = metrics_tx.try_send(Metric::Ping { - reflector: reply.reflector, - measurement_type: reply.measurement_type, - rtt: reply.rtt, - up_time: reply.up_time, - down_time: reply.down_time, - timestamp_ns: clock.as_nanos(), - }); + if let Some(ref tx) = ping_metrics_tx { + let _ = tx.try_send(Metric::Ping { + reflector: reply.reflector, + measurement_type: reply.measurement_type, + rtt: reply.rtt, + up_time: reply.up_time, + down_time: reply.down_time, + timestamp_ns: clock.as_nanos(), + }); + } debug!( "Type: {:4} | Reflector IP: {:>15} | Seq: {:5} | Current time: {:8} | Originate: {:8} | Received time: {:8} | Transmit time : {:8} | RTT: {:8} | UL time: {:5} | DL time: {:5}", diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index fed89eb..4798bbb 100644 --- a/src/ratecontroller.rs +++ b/src/ratecontroller.rs @@ -109,7 +109,7 @@ pub struct Ratecontroller { state_dl: State, state_ul: State, up_direction: StatsDirection, - metrics_tx: SyncSender, + metrics_tx: Option>, } impl Ratecontroller { @@ -248,7 +248,7 @@ impl Ratecontroller { reselect_trigger: Sender, down_direction: StatsDirection, up_direction: StatsDirection, - metrics_tx: SyncSender, + metrics_tx: Option>, ) -> anyhow::Result { let dl_qdisc = Netlink::qdisc_from_ifname(config.download_interface.as_str())?; let dl_safe_rates = diff --git a/src/reflector_selector.rs b/src/reflector_selector.rs index dba1ba0..b118ef3 100644 --- a/src/reflector_selector.rs +++ b/src/reflector_selector.rs @@ -15,7 +15,7 @@ pub struct ReflectorSelector { pub reflector_peers_lock: Arc>>, pub reflector_pool: Vec, pub trigger_channel: Receiver, - pub metrics_tx: SyncSender, + pub metrics_tx: Option>, } impl ReflectorSelector {