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

@@ -25,7 +25,8 @@ pub struct Baseliner {
     pub reselect_trigger: Sender<bool>,
     pub start_time: Instant,
     pub stats_rx: Receiver<PingReply>,
-    pub metrics_tx: SyncSender<Metric>,
+    pub baseline_metrics_tx: Option<SyncSender<Metric>>,
+    pub event_metrics_tx: Option<SyncSender<Metric>>,
 }
 
 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 {
+                if let Some(ref tx) = self.event_metrics_tx {
-                    name: "reselection",
+                    let _ = tx.try_send(Metric::Event {
-                    reason: "anomaly",
+                        name: "reselection",
-                    reflector: Some(time_data.reflector),
+                        reason: "anomaly",
-                    timestamp_ns: clock.as_nanos(),
+                        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 {
+            if let Some(ref tx) = self.baseline_metrics_tx {
-                reflector: time_data.reflector,
+                let _ = tx.try_send(Metric::Baseline {
-                baseline_up_ewma: owd_baseline.up_ewma,
+                    reflector: time_data.reflector,
-                baseline_down_ewma: owd_baseline.down_ewma,
+                    baseline_up_ewma: owd_baseline.up_ewma,
-                recent_up_ewma: owd_recent.up_ewma,
+                    baseline_down_ewma: owd_baseline.down_ewma,
-                recent_down_ewma: owd_recent.down_ewma,
+                    recent_up_ewma: owd_recent.up_ewma,
-                timestamp_ns: clock.as_nanos(),
+                    recent_down_ewma: owd_recent.down_ewma,
-            });
+                    timestamp_ns: clock.as_nanos(),
+                });
+            }
 
             info!(
                 "Reflector {} up baseline = {} down baseline = {}",

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
+    let ping_metrics_tx = metrics_tx
-        .observability_export_ping_metrics
+        .as_ref()
-        .then(|| metrics_tx.clone());
+        .filter(|_| config.observability_export_ping_metrics)
+        .cloned();
 
-    let baseline_tx = config
+    let baseline_metrics_tx = metrics_tx
-        .observability_export_baseline_metrics
+        .as_ref()
-        .then(|| metrics_tx.clone());
+        .filter(|_| config.observability_export_baseline_metrics)
+        .cloned();
 
-    let event_tx = config
+    let event_metrics_tx = metrics_tx
-        .observability_export_events
+        .as_ref()
-        .then(|| metrics_tx.clone());
+        .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::<anyhow::Error>();
 
     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 {
+    if let Some(rx) = metrics_rx {
-        config,
+        let metrics = Metrics {
-        metrics_rx,
+            config,
-    };
+            metrics_rx: rx,
-
+        };
-    let err_tx = error_tx.clone();
+        let err_tx = error_tx.clone();
-    thread::Builder::new()
+        thread::Builder::new()
-        .name("metrics".to_string())
+            .name("metrics".to_string())
-        .spawn(move || {
+            .spawn(move || {
-            if let Err(e) = metrics.run() {
+                if let Err(e) = metrics.run() {
-                let _ = err_tx.send(e);
+                    let _ = err_tx.send(e);
-            }
+                }
-        })?;
+            })?;
+    }
 
     // Drop original sender so recv() unblocks if all threads exit cleanly
     drop(error_tx);

src/pinger.rs

@@ -55,7 +55,7 @@ pub trait PingListener {
         type_: MeasurementType,
         reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
         stats_tx: Sender<PingReply>,
-        metrics_tx: SyncSender<Metric>,
+        ping_metrics_tx: Option<SyncSender<Metric>>,
     ) -> anyhow::Result<()> {
         let socket = &mut open_socket(type_)?;
 
@@ -84,14 +84,16 @@ pub trait PingListener {
                 }
             };
 
-            let _ = metrics_tx.try_send(Metric::Ping {
+            if let Some(ref tx) = ping_metrics_tx {
-                reflector: reply.reflector,
+                let _ = tx.try_send(Metric::Ping {
-                measurement_type: reply.measurement_type,
+                    reflector: reply.reflector,
-                rtt: reply.rtt,
+                    measurement_type: reply.measurement_type,
-                up_time: reply.up_time,
+                    rtt: reply.rtt,
-                down_time: reply.down_time,
+                    up_time: reply.up_time,
-                timestamp_ns: clock.as_nanos(),
+                    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}",

src/ratecontroller.rs

@@ -109,7 +109,7 @@ pub struct Ratecontroller {
     state_dl: State,
     state_ul: State,
     up_direction: StatsDirection,
-    metrics_tx: SyncSender<Metric>,
+    metrics_tx: Option<SyncSender<Metric>>,
 }
 
 impl Ratecontroller {
@@ -248,7 +248,7 @@ impl Ratecontroller {
         reselect_trigger: Sender<bool>,
         down_direction: StatsDirection,
         up_direction: StatsDirection,
-        metrics_tx: SyncSender<Metric>,
+        metrics_tx: Option<SyncSender<Metric>>,
     ) -> anyhow::Result<Self> {
         let dl_qdisc = Netlink::qdisc_from_ifname(config.download_interface.as_str())?;
         let dl_safe_rates =

src/reflector_selector.rs

@@ -15,7 +15,7 @@ pub struct ReflectorSelector {
     pub reflector_peers_lock: Arc<RwLock<Vec<IpAddr>>>,
     pub reflector_pool: Vec<IpAddr>,
     pub trigger_channel: Receiver<bool>,
-    pub metrics_tx: SyncSender<Metric>,
+    pub metrics_tx: Option<SyncSender<Metric>>,
 }
 
 impl ReflectorSelector {