fix(metrics): use AtomicU32 for drop counter to support MIPS targets

Also bumps version to 0.4.1.

Cargo.lock

@@ -190,9 +190,9 @@ checksum = "135b12329e5e3ce057a9f972339ea52bc954fe1e9358ef27f95e89716fbc5424"
 
 [[package]]
 name = "icmp-socket2"
-version = "0.4.1"
+version = "0.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e4a72abd00afc496b7dc6a0cb275c19c669da9d0c2ee5d4c2570a8a67b83b53f"
+checksum = "8e57405ef902d5506111426bb8a2498f10979c19273bcdadc852ae03d45d19d8"
 dependencies = [
  "socket2",
 ]
@@ -392,7 +392,7 @@ dependencies = [
 
 [[package]]
 name = "sqm-autorate-rust"
-version = "0.3.0"
+version = "0.4.1"
 dependencies = [
  "anyhow",
  "fastrand",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "sqm-autorate-rust"
-version = "0.3.0"
+version = "0.4.1"
 edition = "2024"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -20,11 +20,11 @@ strip = "symbols"
 
 [dependencies]
 anyhow = "1.0"
-icmp-socket2 = "0.4.1"
+icmp-socket2 = "0.4.2"
 netlink-bindings = { version = "0.2", features = ["rt-link", "tc"] }
 netlink-socket2 = "0.2"
 fastrand = "2"
-rustix = { version = "1.1", features = ["fs", "thread", "time"] }
+rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] }
 rust-uci = { version = "0.1", optional = true }
 thiserror = "2.0"
 libc = "0.2"
@@ -34,4 +34,4 @@ features = ["std"]
 version = "0.4"
 
 [features]
-uci = ["dep:rust-uci"]
+uci = ["dep:rust-uci"]

src/baseliner.rs

@@ -1,9 +1,12 @@
+use crate::SHUTDOWN;
+use crate::Config;
+use crate::metrics::{Metric, MetricsSender};
 use crate::pinger::PingReply;
 use crate::util::MutexExt;
-use crate::Config;
 use log::info;
 use std::collections::HashMap;
 use std::net::IpAddr;
+use std::sync::atomic::Ordering;
 use std::sync::mpsc::{Receiver, Sender};
 use std::sync::{Arc, Mutex};
 use std::time::Instant;
@@ -21,7 +24,9 @@ pub struct Baseliner {
     pub owd_recent: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>,
     pub reselect_trigger: Sender<bool>,
     pub start_time: Instant,
-    pub stats_receiver: Receiver<PingReply>,
+    pub stats_rx: Receiver<PingReply>,
+    pub baseline_metrics: MetricsSender,
+    pub event_metrics: MetricsSender,
 }
 
 fn ewma_factor(tick: f64, dur: f64) -> f64 {
@@ -42,7 +47,11 @@ impl Baseliner {
         let fast_factor = ewma_factor(self.config.tick_interval, 0.4);
 
         loop {
-            let time_data = self.stats_receiver.recv()?;
+            if SHUTDOWN.load(Ordering::Relaxed) {
+                info!("Baseliner shutting down");
+                return Ok(());
+            }
+            let time_data = self.stats_rx.recv()?;
 
             let mut owd_baseline_map = self.owd_baseline.lock_anyhow()?;
             let mut owd_recent_map = self.owd_recent.lock_anyhow()?;
@@ -100,6 +109,12 @@ impl Baseliner {
                     "Reflector {} has OWD > 5 seconds more than baseline, triggering reselection",
                     time_data.reflector
                 );
+                self.event_metrics.send(Metric::Event {
+                    name: "reselection",
+                    reason: "anomaly",
+                    reflector: Some(time_data.reflector),
+                    tags: &[],
+                });
                 // If reselection is disabled this would trigger an error
                 // so just ignore the result
                 let _ = self.reselect_trigger.send(true);
@@ -123,6 +138,14 @@ impl Baseliner {
                 }
             }
 
+            self.baseline_metrics.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,
+            });
+
             info!(
                 "Reflector {} up baseline = {} down baseline = {}",
                 time_data.reflector, owd_baseline.up_ewma, owd_baseline.down_ewma

src/config.rs

@@ -4,6 +4,8 @@ use log::Level;
 use log::warn;
 #[cfg(feature = "uci")]
 use rust_uci::Uci;
+use std::fmt;
+use std::fmt::Display;
 use std::fs::File;
 use std::io::BufRead;
 use std::net::IpAddr;
@@ -41,6 +43,17 @@ pub enum MeasurementType {
     TcpTimestamps,
 }
 
+impl Display for MeasurementType {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            MeasurementType::Icmp => write!(f, "icmp"),
+            MeasurementType::IcmpTimestamps => write!(f, "icmp-timestamps"),
+            MeasurementType::Ntp => write!(f, "ntp"),
+            MeasurementType::TcpTimestamps => write!(f, "tcp-timestamps"),
+        }
+    }
+}
+
 impl FromStr for MeasurementType {
     type Err = ConfigError;
 
@@ -55,6 +68,24 @@ impl FromStr for MeasurementType {
     }
 }
 
+#[derive(Clone, Copy, Debug)]
+pub enum ObservabilityProtocol {
+    Udp,
+    Tcp,
+}
+
+impl FromStr for ObservabilityProtocol {
+    type Err = ConfigError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s.to_lowercase().as_str() {
+            "udp" => Ok(ObservabilityProtocol::Udp),
+            "tcp" => Ok(ObservabilityProtocol::Tcp),
+            _ => Err(ConfigError::InvalidMeasurementType(s.to_string())),
+        }
+    }
+}
+
 #[derive(Clone, Debug)]
 pub struct Config {
     // Network section
@@ -73,6 +104,19 @@ pub struct Config {
     pub stats_file: String,
     pub suppress_statistics: bool,
 
+    // Observability section
+    pub observability_enabled: bool,
+    pub observability_protocol: ObservabilityProtocol,
+    pub observability_host: Option<String>,
+    pub observability_port: u16,
+    pub observability_batch_size: usize,
+    pub observability_batch_timeout_ms: u64,
+    pub observability_export_ping_metrics: bool,
+    pub observability_export_rate_metrics: bool,
+    pub observability_export_baseline_metrics: bool,
+    pub observability_export_events: bool,
+    pub observability_host_tag: String,
+
     // Advanced section
     pub download_delay_ms: f64,
     pub high_load_level: f64,
@@ -84,6 +128,9 @@ pub struct Config {
     pub speed_hist_size: u32,
     pub tick_interval: f64,
     pub upload_delay_ms: f64,
+
+    // Monitoring / dry-run
+    pub dry_run: bool,
 }
 
 impl Config {
@@ -135,6 +182,59 @@ impl Config {
                 "sqm-autorate.@output[0].suppress_statistics",
                 Some(false),
             )?,
+
+            // Observability section
+            observability_enabled: Self::get::<bool>(
+                "SQMA_OBSERVABILITY_ENABLED",
+                "sqm-autorate.@observability[0].enabled",
+                Some(false),
+            )?,
+            observability_protocol: Self::get::<ObservabilityProtocol>(
+                "SQMA_OBSERVABILITY_PROTOCOL",
+                "sqm-autorate.@observability[0].protocol",
+                Some(ObservabilityProtocol::Udp),
+            )?,
+            observability_host: Self::get_optional::<String>(
+                "SQMA_OBSERVABILITY_HOST",
+                "sqm-autorate.@observability[0].host",
+            ),
+            observability_port: Self::get::<u16>(
+                "SQMA_OBSERVABILITY_PORT",
+                "sqm-autorate.@observability[0].port",
+                Some(8089),
+            )?,
+            observability_batch_size: Self::get::<usize>(
+                "SQMA_OBSERVABILITY_BATCH_SIZE",
+                "sqm-autorate.@observability[0].batch_size",
+                Some(25),
+            )?,
+            observability_batch_timeout_ms: Self::get::<u64>(
+                "SQMA_OBSERVABILITY_BATCH_TIMEOUT_MS",
+                "sqm-autorate.@observability[0].batch_timeout_ms",
+                Some(100),
+            )?,
+            observability_export_ping_metrics: Self::get::<bool>(
+                "SQMA_OBSERVABILITY_EXPORT_PING_METRICS",
+                "sqm-autorate.@observability[0].export_ping_metrics",
+                Some(false),
+            )?,
+            observability_export_rate_metrics: Self::get::<bool>(
+                "SQMA_OBSERVABILITY_EXPORT_RATE_METRICS",
+                "sqm-autorate.@observability[0].export_rate_metrics",
+                Some(true),
+            )?,
+            observability_export_baseline_metrics: Self::get::<bool>(
+                "SQMA_OBSERVABILITY_EXPORT_BASELINE_METRICS",
+                "sqm-autorate.@observability[0].export_baseline_metrics",
+                Some(false),
+            )?,
+            observability_export_events: Self::get::<bool>(
+                "SQMA_OBSERVABILITY_EXPORT_EVENTS",
+                "sqm-autorate.@observability[0].export_events",
+                Some(true),
+            )?,
+            observability_host_tag: Self::get_host_tag(),
+
             // Advanced section
             download_delay_ms: Self::get::<f64>(
                 "SQMA_DOWNLOAD_DELAY_MS",
@@ -186,6 +286,11 @@ impl Config {
                 "sqm-autorate.@advanced_settings[0].upload_delay_ms",
                 Some(15.0),
             )?,
+            dry_run: Self::get::<bool>(
+                "SQMA_DRY_RUN",
+                "sqm-autorate.@advanced_settings[0].dry_run",
+                Some(false),
+            )?,
         };
 
         let mut config = config;
@@ -227,6 +332,24 @@ impl Config {
         }
     }
 
+    fn get_host_tag() -> String {
+        if let Ok(val) = env::var("SQMA_OBSERVABILITY_HOST_TAG") {
+            if !val.is_empty() {
+                return val;
+            }
+        }
+
+        // rustix can give us the hostname without shelling out
+        rustix::system::uname()
+            .nodename()
+            .to_string_lossy()
+            .into_owned()
+    }
+
+    fn get_optional<T: FromStr>(env_key: &str, uci_key: &str) -> Option<T> {
+        Self::get_value(env_key, uci_key).and_then(|val| val.parse::<T>().ok())
+    }
+
     fn get_value(env_key: &str, uci_key: &str) -> Option<String> {
         if let Ok(val) = env::var(env_key) {
             return Some(val);

src/main.rs

@@ -3,6 +3,7 @@ extern crate core;
 mod baseliner;
 mod config;
 mod log;
+mod metrics;
 mod netlink;
 mod pinger;
 mod pinger_icmp;
@@ -13,17 +14,25 @@ mod time;
 mod util;
 
 use crate::baseliner::{Baseliner, ReflectorStats};
+use crate::metrics::{Metric, Metrics, MetricsSender};
 use ::log::{debug, info};
 use std::collections::HashMap;
 use std::net::IpAddr;
 use std::str::FromStr;
-use std::sync::mpsc::channel;
+use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
+use std::sync::mpsc::{channel, sync_channel, RecvTimeoutError};
 use std::sync::{Arc, Mutex, RwLock};
 use std::thread::sleep;
 use std::time::Duration;
 use std::time::Instant;
 use std::{process, thread};
 
+pub static SHUTDOWN: AtomicBool = AtomicBool::new(false);
+
+extern "C" fn signal_handler(_: libc::c_int) {
+    SHUTDOWN.store(true, Ordering::Relaxed);
+}
+
 use crate::config::{Config, MeasurementType};
 use crate::netlink::Netlink;
 use crate::pinger::{PingListener, PingSender};
@@ -37,8 +46,16 @@ const VERSION: &str = env!("CARGO_PKG_VERSION");
 fn main() -> anyhow::Result<()> {
     println!("Starting sqm-autorate version {}", VERSION);
 
+    unsafe {
+        libc::signal(libc::SIGINT, signal_handler as *const () as libc::sighandler_t);
+        libc::signal(libc::SIGTERM, signal_handler as *const () as libc::sighandler_t);
+    }
+
     let config = Config::new()?;
     log::init(config.log_level)?;
+    if config.dry_run {
+        info!("*** MONITORING MODE ACTIVE — qdisc rates will NOT be changed ***");
+    }
     let mut reflectors = config.load_reflectors()?;
     let start_t = Instant::now();
 
@@ -74,10 +91,53 @@ fn main() -> anyhow::Result<()> {
         }
     }
 
-    let (baseliner_stats_sender, baseliner_stats_receiver) = channel();
-    let (reselect_sender, reselect_receiver) = channel();
+    let (baseliner_stats_tx, baseliner_stats_rx) = channel();
+    let (error_tx, error_rx) = channel::<anyhow::Error>();
+    let (reselect_tx, reselect_rx) = channel();
 
-    let (mut pinger_receiver, mut pinger_sender) = match config.measurement_type {
+    let dropped = Arc::new(AtomicU32::new(0));
+
+    let (metrics_tx, metrics_thread_handle) = if config.observability_enabled {
+        let (tx, rx) = sync_channel(1000);
+        let metrics = Metrics {
+            config: config.clone(),
+            metrics_rx: rx,
+            metrics_dropped: Arc::clone(&dropped),
+        };
+        let err_tx = error_tx.clone();
+        let handle = thread::Builder::new()
+            .name("metrics".to_string())
+            .spawn(move || {
+                if let Err(e) = metrics.run() {
+                    let _ = err_tx.send(e);
+                }
+            })?;
+        (Some(tx), Some(handle))
+    } else {
+        (None, None)
+    };
+
+    let make_sender = |enabled: bool| -> MetricsSender {
+        metrics_tx
+            .as_ref()
+            .filter(|_| enabled)
+            .map(|tx| MetricsSender::new(tx.clone(), Arc::clone(&dropped)))
+            .unwrap_or_else(MetricsSender::disabled)
+    };
+
+    let ping_metrics = make_sender(config.observability_export_ping_metrics);
+    let baseline_metrics = make_sender(config.observability_export_baseline_metrics);
+    let event_metrics = make_sender(config.observability_export_events);
+    let rate_metrics = make_sender(config.observability_export_rate_metrics);
+
+    event_metrics.send(Metric::Event {
+        name: "starting",
+        reason: "",
+        reflector: None,
+        tags: if config.dry_run { &[("dry_run", "true")] } else { &[] },
+    });
+
+    let (mut ping_listener, mut ping_sender) = match config.measurement_type {
         MeasurementType::Icmp => (
             Box::new(PingerICMPEchoListener {}) as Box<dyn PingListener + Send>,
             Box::new(PingerICMPEchoSender {}) as Box<dyn PingSender + Send>,
@@ -95,9 +155,11 @@ fn main() -> anyhow::Result<()> {
         config: config.clone(),
         owd_baseline: owd_baseline.clone(),
         owd_recent: owd_recent.clone(),
-        reselect_trigger: reselect_sender.clone(),
+        reselect_trigger: reselect_tx.clone(),
         start_time: start_t,
-        stats_receiver: baseliner_stats_receiver,
+        stats_rx: baseliner_stats_rx,
+        baseline_metrics,
+        event_metrics: event_metrics.clone(),
     };
 
     let down_qdisc = Netlink::qdisc_from_ifname(config.download_interface.as_str())?;
@@ -111,8 +173,8 @@ fn main() -> anyhow::Result<()> {
         "Setting shaper rates to minimum (D/L): {} / {}",
         config.download_min_kbits, config.upload_min_kbits
     );
-    Netlink::set_qdisc_rate(down_qdisc, config.download_min_kbits as u64)?;
-    Netlink::set_qdisc_rate(up_qdisc, config.upload_min_kbits as u64)?;
+    Netlink::set_qdisc_rate(down_qdisc, config.download_min_kbits as u64, config.dry_run)?;
+    Netlink::set_qdisc_rate(up_qdisc, config.upload_min_kbits as u64, config.dry_run)?;
 
     // Sleep for a few seconds to give the shaper a chance
     // to control the queue if load is heavy
@@ -123,22 +185,21 @@ fn main() -> anyhow::Result<()> {
     );
     sleep(settle_sleep_time);
 
-    let (error_tx, error_rx) = channel::<anyhow::Error>();
-
     let err_tx = error_tx.clone();
     let reflector_peers_lock_clone = reflector_peers_lock.clone();
-    thread::Builder::new().name("receiver".to_string()).spawn(
-        move || {
-            if let Err(e) = pinger_receiver.listen(
+    thread::Builder::new()
+        .name("receiver".to_string())
+        .spawn(move || {
+            if let Err(e) = ping_listener.listen(
                 id,
                 config.measurement_type,
                 reflector_peers_lock_clone,
-                baseliner_stats_sender,
+                baseliner_stats_tx,
+                ping_metrics,
             ) {
                 let _ = err_tx.send(e);
             }
-        },
-    )?;
+        })?;
 
     let err_tx = error_tx.clone();
     thread::Builder::new()
@@ -151,9 +212,10 @@ fn main() -> anyhow::Result<()> {
 
     let err_tx = error_tx.clone();
     let reflector_peers_lock_clone = reflector_peers_lock.clone();
-    thread::Builder::new().name("sender".to_string()).spawn(
-        move || {
-            if let Err(e) = pinger_sender.send(
+    thread::Builder::new()
+        .name("sender".to_string())
+        .spawn(move || {
+            if let Err(e) = ping_sender.send(
                 id,
                 config.measurement_type,
                 reflector_peers_lock_clone,
@@ -161,8 +223,9 @@ fn main() -> anyhow::Result<()> {
             ) {
                 let _ = err_tx.send(e);
             }
-        },
-    )?;
+        })?;
+
+    let main_event_metrics = event_metrics.clone();
 
     if reflector_pool_size > config.num_reflectors as usize {
         let reflector_selector = ReflectorSelector {
@@ -170,7 +233,8 @@ fn main() -> anyhow::Result<()> {
             owd_recent: owd_recent.clone(),
             reflector_peers_lock: reflector_peers_lock.clone(),
             reflector_pool,
-            trigger_channel: reselect_receiver,
+            trigger_channel: reselect_rx,
+            metrics: event_metrics,
         };
         let err_tx = error_tx.clone();
         thread::Builder::new()
@@ -205,9 +269,10 @@ fn main() -> anyhow::Result<()> {
         owd_baseline,
         owd_recent,
         reflector_peers_lock,
-        reselect_sender,
+        reselect_tx,
         dl_direction,
         ul_direction,
+        rate_metrics,
     )?;
 
     debug!(
@@ -229,12 +294,45 @@ fn main() -> anyhow::Result<()> {
             }
         })?;
 
-    // Drop original sender so recv() unblocks if all threads exit cleanly
+    // Drop original sender so the channel disconnects if all threads exit cleanly
     drop(error_tx);
 
-    // Wait for first error
-    match error_rx.recv() {
-        Ok(e) => Err(anyhow::anyhow!("thread exited with error: {e}")),
-        Err(_) => Ok(()), // all senders dropped = all threads exited without error
+    let result = loop {
+        match error_rx.recv_timeout(Duration::from_secs(1)) {
+            Ok(e) => break Err(anyhow::anyhow!("thread exited with error: {e}")),
+            Err(RecvTimeoutError::Disconnected) => break Ok(()),
+            Err(RecvTimeoutError::Timeout) => {
+                if SHUTDOWN.load(Ordering::Relaxed) {
+                    info!("Received shutdown signal");
+                    break Ok(());
+                }
+            }
+        }
+    };
+
+    let stopping_reason = if result.is_err() { "error" } else { "signal" };
+    main_event_metrics.send(Metric::Event {
+        name: "stopping",
+        reason: stopping_reason,
+        reflector: None,
+        tags: &[],
+    });
+
+    // Drop all MetricsSender instances and the raw tx so the metrics channel
+    // disconnects once all threads also drop their copies, allowing the
+    // metrics thread to drain and exit cleanly.
+    drop(main_event_metrics);
+    drop(metrics_tx);
+    if let Some(handle) = metrics_thread_handle {
+        let _ = handle.join();
     }
+
+    info!(
+        "Restoring base shaper rates (D/L): {} / {}",
+        config.download_base_kbits, config.upload_base_kbits
+    );
+    let _ = Netlink::set_qdisc_rate(down_qdisc, config.download_base_kbits as u64, config.dry_run);
+    let _ = Netlink::set_qdisc_rate(up_qdisc, config.upload_base_kbits as u64, config.dry_run);
+
+    result
 }

src/metrics.rs

@@ -0,0 +1,331 @@
+use crate::config::{MeasurementType, ObservabilityProtocol};
+use crate::time::Time;
+use crate::Config;
+use log::{error, info, warn};
+use rustix::time::ClockId;
+use std::fmt::Write;
+use std::net::{IpAddr, TcpStream, UdpSocket};
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::mpsc::{Receiver, RecvTimeoutError, SyncSender};
+use std::sync::Arc;
+use std::time::Duration;
+
+const MAX_RECONNECT_BACKOFF: u64 = 60;
+
+enum Transport {
+    Udp(UdpSocket),
+    Tcp {
+        stream: Option<TcpStream>,
+        host: String,
+        port: u16,
+        reconnect_backoff: u64,
+    },
+}
+
+impl Transport {
+    fn new_udp(host: &str, port: u16) -> anyhow::Result<Self> {
+        let socket = UdpSocket::bind("0.0.0.0:0")?;
+        socket.set_write_timeout(Some(Duration::from_millis(100)))?;
+        socket.connect((host, port))?;
+        Ok(Transport::Udp(socket))
+    }
+
+    fn new_tcp(host: &str, port: u16) -> Self {
+        Transport::Tcp {
+            stream: None,
+            host: host.to_string(),
+            port,
+            reconnect_backoff: 1,
+        }
+    }
+
+    fn send(&mut self, data: &str) {
+        match self {
+            Transport::Udp(socket) => {
+                if let Err(e) = socket.send(data.as_bytes()) {
+                    warn!("UDP send failed: {}", e);
+                }
+            }
+            Transport::Tcp {
+                stream,
+                host,
+                port,
+                reconnect_backoff,
+            } => {
+                if stream.is_none() {
+                    match TcpStream::connect((host.as_str(), *port)) {
+                        Ok(s) => {
+                            s.set_write_timeout(Some(Duration::from_millis(500))).ok();
+                            info!("Connected to metrics collector at {}:{}", host, port);
+                            *stream = Some(s);
+                            *reconnect_backoff = 1;
+                        }
+                        Err(e) => {
+                            warn!(
+                                "Failed to connect to {}:{} - {}, backoff {}s",
+                                host, port, e, reconnect_backoff
+                            );
+                            *reconnect_backoff =
+                                (*reconnect_backoff * 2).min(MAX_RECONNECT_BACKOFF);
+                            return;
+                        }
+                    }
+                }
+
+                let full_data = format!("{}\n", data);
+                // take the stream out, write, put it back or set None on failure
+                if let Some(s) = stream {
+                    if let Err(e) = std::io::Write::write_all(s, full_data.as_bytes()) {
+                        warn!("TCP send failed: {}", e);
+                        *stream = None;
+                        *reconnect_backoff = (*reconnect_backoff * 2).min(MAX_RECONNECT_BACKOFF);
+                    }
+                }
+            }
+        }
+    }
+}
+
+pub enum Metric {
+    Ping {
+        reflector: IpAddr,
+        measurement_type: MeasurementType,
+        rtt: f64,
+        up_time: f64,
+        down_time: f64,
+    },
+    Rate {
+        dl_rate: f64,
+        ul_rate: f64,
+        rx_load: f64,
+        tx_load: f64,
+        delta_delay_down: f64,
+        delta_delay_up: f64,
+    },
+    Baseline {
+        reflector: IpAddr,
+        baseline_up_ewma: f64,
+        baseline_down_ewma: f64,
+        recent_up_ewma: f64,
+        recent_down_ewma: f64,
+    },
+    Event {
+        name: &'static str,
+        reason: &'static str,
+        reflector: Option<IpAddr>,
+        tags: &'static [(&'static str, &'static str)],
+    },
+    Dropped {
+        count: u32,
+    },
+}
+
+#[derive(Clone)]
+pub struct MetricsSender {
+    tx: Option<SyncSender<(Metric, u64)>>,
+    dropped: Arc<AtomicU32>,
+}
+
+impl MetricsSender {
+    pub fn new(tx: SyncSender<(Metric, u64)>, dropped: Arc<AtomicU32>) -> Self {
+        Self {
+            tx: Some(tx),
+            dropped,
+        }
+    }
+
+    pub fn disabled() -> Self {
+        Self {
+            tx: None,
+            dropped: Arc::new(AtomicU32::new(0)),
+        }
+    }
+
+    pub fn send(&self, metric: Metric) {
+        if let Some(ref tx) = self.tx {
+            let ts = Time::new(ClockId::Realtime).as_nanos();
+            if tx.try_send((metric, ts)).is_err() {
+                self.dropped.fetch_add(1, Ordering::Relaxed);
+            }
+        }
+    }
+}
+
+pub struct Metrics {
+    pub config: Config,
+    pub metrics_dropped: Arc<AtomicU32>,
+    pub metrics_rx: Receiver<(Metric, u64)>,
+}
+
+impl Metrics {
+    pub fn run(self) -> anyhow::Result<()> {
+        let host = match &self.config.observability_host {
+            Some(h) => h.clone(),
+            None => {
+                error!("Observability host not configured - metrics exporter disabled");
+                return Ok(());
+            }
+        };
+
+        let port = self.config.observability_port;
+        let batch_size = self.config.observability_batch_size;
+        let timeout = Duration::from_millis(self.config.observability_batch_timeout_ms);
+        let host_tag = &self.config.observability_host_tag.clone();
+
+        let mut transport = match self.config.observability_protocol {
+            ObservabilityProtocol::Udp => {
+                info!("Metrics exporter configured for UDP to {}:{}", host, port);
+                Transport::new_udp(&host, port)?
+            }
+            ObservabilityProtocol::Tcp => {
+                info!("Metrics exporter configured for TCP to {}:{}", host, port);
+                Transport::new_tcp(&host, port)
+            }
+        };
+
+        let mut batch = Vec::with_capacity(batch_size);
+
+        loop {
+            // block until first metric or timeout
+            match self.metrics_rx.recv_timeout(timeout) {
+                Ok(metric) => batch.push(metric),
+                Err(RecvTimeoutError::Timeout) => {
+                    let dropped = self.metrics_dropped.swap(0, Ordering::Relaxed);
+                    if dropped > 0 {
+                        let ts = Time::new(ClockId::Realtime).as_nanos();
+                        batch.push((Metric::Dropped { count: dropped }, ts));
+                    }
+                    if !batch.is_empty() {
+                        self.flush(&mut transport, &batch, host_tag);
+                        batch.clear();
+                    }
+                    continue;
+                }
+                Err(RecvTimeoutError::Disconnected) => break,
+            }
+
+            // greedily drain without blocking
+            loop {
+                match self.metrics_rx.try_recv() {
+                    Ok(metric) => batch.push(metric),
+                    Err(_) => break,
+                }
+                if batch.len() >= batch_size {
+                    break;
+                }
+            }
+
+            if batch.len() >= batch_size {
+                self.flush(&mut transport, &batch, host_tag);
+                batch.clear();
+            }
+        }
+
+        Ok(())
+    }
+
+    fn flush(&self, transport: &mut Transport, batch: &[(Metric, u64)], host_tag: &str) {
+        match transport {
+            /*
+             * the UDP receive buffer might be too small on the receiver to accept batched data,
+             * and it'll get cutoff in the middle of a record. so send metrics directly on UDP
+             */
+            Transport::Udp(_) => {
+                let mut data = String::with_capacity(300);
+                for (metric, timestamp_ns) in batch {
+                    data.clear();
+                    self.write_lines(metric, *timestamp_ns, host_tag, &mut data);
+                    transport.send(&data);
+                }
+            }
+            Transport::Tcp { .. } => {
+                let mut data = String::with_capacity(batch.len() * 300);
+                for (metric, timestamp_ns) in batch {
+                    self.write_lines(metric, *timestamp_ns, host_tag, &mut data);
+                }
+                transport.send(&data);
+            }
+        }
+    }
+
+    fn write_lines(&self, metric: &Metric, timestamp_ns: u64, host_tag: &str, out: &mut String) {
+        match metric {
+            Metric::Ping {
+                reflector,
+                measurement_type,
+                rtt,
+                up_time,
+                down_time,
+            } => {
+                writeln!(
+                    out,
+                    "sqm_ping,host={host_tag},reflector={reflector},type={measurement_type} \
+                    rtt={rtt:.3},up_time={up_time:.3},down_time={down_time:.3} {timestamp_ns}i"
+                )
+                .unwrap_or(());
+            }
+            Metric::Rate {
+                dl_rate,
+                ul_rate,
+                rx_load,
+                tx_load,
+                delta_delay_down,
+                delta_delay_up,
+            } => {
+                writeln!(out,
+                    "sqm_rate,host={host_tag},direction=download \
+                    rate_kbps={dl_rate:.0},load={rx_load:.4},delta_delay={delta_delay_down:.3} {timestamp_ns}i"
+                ).unwrap_or(());
+                writeln!(out,
+                    "sqm_rate,host={host_tag},direction=upload \
+                    rate_kbps={ul_rate:.0},load={tx_load:.4},delta_delay={delta_delay_up:.3} {timestamp_ns}i"
+                ).unwrap_or(());
+            }
+            Metric::Baseline {
+                reflector,
+                baseline_up_ewma,
+                baseline_down_ewma,
+                recent_up_ewma,
+                recent_down_ewma,
+            } => {
+                writeln!(out,
+                    "sqm_baseline,host={host_tag},reflector={reflector},direction=up \
+                    baseline_ewma={baseline_up_ewma:.3},recent_ewma={recent_up_ewma:.3} {timestamp_ns}i"
+                ).unwrap_or(());
+                writeln!(out,
+                    "sqm_baseline,host={host_tag},reflector={reflector},direction=down \
+                    baseline_ewma={baseline_down_ewma:.3},recent_ewma={recent_down_ewma:.3} {timestamp_ns}i"
+                ).unwrap_or(());
+            }
+            Metric::Event {
+                name,
+                reason,
+                reflector,
+                tags,
+            } => {
+                let mut tag_str = format!("host={host_tag},type={name}");
+
+                if let Some(reflector) = reflector {
+                    write!(tag_str, ",reflector={reflector}").unwrap_or(());
+                }
+
+                if !reason.is_empty() {
+                    write!(tag_str, ",reason={reason}").unwrap_or(());
+                }
+
+                for (k, v) in *tags {
+                    write!(tag_str, ",{k}={v}").unwrap_or(());
+                }
+
+                writeln!(out, "sqm_event,{tag_str} count=1i {timestamp_ns}").unwrap_or(());
+            }
+            Metric::Dropped { count } => {
+                writeln!(
+                    out,
+                    "sqm_metrics_dropped,host={host_tag} count={count}i {timestamp_ns}"
+                )
+                .unwrap_or(());
+            }
+        }
+    }
+}

src/netlink.rs

@@ -1,6 +1,7 @@
 use std::io;
 use std::str::Utf8Error;
 
+use log::info;
 use netlink_bindings::{rt_link, tc};
 use netlink_socket2::NetlinkSocket;
 use thiserror::Error;
@@ -111,7 +112,15 @@ impl Netlink {
         Netlink::qdisc_from_ifindex(ifindex, ifname)
     }
 
-    pub fn set_qdisc_rate(qdisc: Qdisc, bandwidth_kbit: u64) -> Result<(), NetlinkError> {
+    pub fn set_qdisc_rate(qdisc: Qdisc, bandwidth_kbit: u64, dry_run: bool) -> Result<(), NetlinkError> {
+        if dry_run {
+            info!(
+                "dry-run: would set qdisc ifindex={} to {} kbit/s",
+                qdisc.ifindex, bandwidth_kbit
+            );
+            return Ok(());
+        }
+
         let mut socket = NetlinkSocket::new();
         let bandwidth = bandwidth_kbit * 1000 / 8;
 

src/pinger.rs

@@ -1,9 +1,12 @@
+use crate::SHUTDOWN;
 use crate::MeasurementType;
+use crate::metrics::{Metric, MetricsSender};
+use crate::util::RwLockExt;
 use icmp_socket2::socket::IcmpSocket;
 use icmp_socket2::{IcmpSocket4, Icmpv4Packet};
-use crate::util::RwLockExt;
-use log::{debug, warn};
+use log::{debug, info, warn};
 use std::net::{IpAddr, Ipv4Addr};
+use std::sync::atomic::Ordering;
 use std::sync::mpsc::Sender;
 use std::sync::{Arc, RwLock};
 use std::time::{Duration, Instant};
@@ -24,8 +27,9 @@ pub enum PingError {
 
 pub struct PingReply {
     pub reflector: IpAddr,
+    pub measurement_type: MeasurementType,
     pub seq: u16,
-    pub rtt: i64,
+    pub rtt: f64,
     pub current_time: i64,
     pub down_time: f64,
     pub up_time: f64,
@@ -37,9 +41,7 @@ pub struct PingReply {
 
 fn open_socket(type_: MeasurementType) -> io::Result<IcmpSocket4> {
     match type_ {
-        MeasurementType::Icmp | MeasurementType::IcmpTimestamps => {
-            IcmpSocket4::new()
-        },
+        MeasurementType::Icmp | MeasurementType::IcmpTimestamps => IcmpSocket4::new(),
         _ => {
             unimplemented!()
         }
@@ -52,11 +54,16 @@ pub trait PingListener {
         id: u16,
         type_: MeasurementType,
         reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
-        stats_sender: Sender<PingReply>,
+        stats_tx: Sender<PingReply>,
+        ping_metrics: MetricsSender,
     ) -> anyhow::Result<()> {
         let socket = &mut open_socket(type_)?;
 
         loop {
+            if SHUTDOWN.load(Ordering::Relaxed) {
+                info!("Ping listener shutting down");
+                return Ok(());
+            }
             let (pkt, sender) = match socket.rcv_from() {
                 Ok(val) => val,
                 Err(_) => continue,
@@ -69,7 +76,7 @@ pub trait PingListener {
                 continue;
             }
 
-            let reply = match self.parse_packet(id, addr, pkt) {
+            let reply = match self.parse_packet(id, addr, type_, pkt) {
                 Ok(val) => val,
                 Err(_) => {
                     // parse_packet will throw an error if it's an unknown protocol etc.
@@ -78,15 +85,41 @@ pub trait PingListener {
                 }
             };
 
-            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}", "ICMP", addr.to_string(), reply.seq, reply.current_time, reply.originate_timestamp, reply.receive_timestamp, reply.transmit_timestamp, reply.rtt, reply.up_time, reply.down_time);
-            if let Err(e) = stats_sender.send(reply) {
+            ping_metrics.send(Metric::Ping {
+                reflector: reply.reflector,
+                measurement_type: reply.measurement_type,
+                rtt: reply.rtt,
+                up_time: reply.up_time,
+                down_time: reply.down_time,
+            });
+
+            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}",
+                "ICMP",
+                addr.to_string(),
+                reply.seq,
+                reply.current_time,
+                reply.originate_timestamp,
+                reply.receive_timestamp,
+                reply.transmit_timestamp,
+                reply.rtt,
+                reply.up_time,
+                reply.down_time
+            );
+            if let Err(e) = stats_tx.send(reply) {
                 warn!("Stats channel closed, stopping listener: {}", e);
                 break Ok(());
             }
         }
     }
 
-    fn parse_packet(&self, id: u16, reflector: IpAddr, pkt: Icmpv4Packet) -> Result<PingReply, PingError>;
+    fn parse_packet(
+        &self,
+        id: u16,
+        reflector: IpAddr,
+        measurement_type: MeasurementType,
+        pkt: Icmpv4Packet,
+    ) -> Result<PingReply, PingError>;
 }
 
 pub trait PingSender {
@@ -103,6 +136,10 @@ pub trait PingSender {
         let tick_duration_ms: u16 = (tick_interval * 1000.0) as u16;
 
         loop {
+            if SHUTDOWN.load(Ordering::Relaxed) {
+                info!("Ping sender shutting down");
+                return Ok(());
+            }
             // Clone the reflectors vec and drop the read lock immediately —
             // holding it for the entire tick would starve the reflector selector's write lock.
             let reflectors_unlocked = reflectors_lock.read_anyhow()?;

src/pinger_icmp.rs

@@ -1,6 +1,7 @@
 use std::net::IpAddr;
 use std::time::Instant;
 
+use crate::config::MeasurementType;
 use crate::pinger::{PingError, PingListener, PingReply, PingSender};
 use crate::time::Time;
 use icmp_socket2::Icmpv4Message;
@@ -14,7 +15,13 @@ pub struct PingerICMPEchoSender {}
 
 impl PingListener for PingerICMPEchoListener {
     // Result: RTT, down time, up time
-    fn parse_packet(&self, id: u16, reflector: IpAddr, pkt: Icmpv4Packet) -> Result<PingReply, PingError> {
+    fn parse_packet(
+        &self,
+        id: u16,
+        reflector: IpAddr,
+        measurement_type: MeasurementType,
+        pkt: Icmpv4Packet,
+    ) -> Result<PingReply, PingError> {
         match pkt.typ {
             0 => {
                 if let Icmpv4Message::EchoReply {
@@ -33,32 +40,37 @@ impl PingListener for PingerICMPEchoListener {
                     let time_sent = match payload.as_slice().try_into() {
                         Ok(bytes) => u64::from_be_bytes(bytes) as i64,
                         Err(_) => {
-                            return Err(PingError::InvalidPacket(format!("Expected 8 bytes payload, but found {}", payload.len())))
+                            return Err(PingError::InvalidPacket(format!(
+                                "Expected 8 bytes payload, but found {}",
+                                payload.len()
+                            )));
                         }
                     };
 
                     let clock = Time::new(ClockId::Monotonic);
                     let time_ms = clock.to_milliseconds() as i64;
 
-                    let rtt: i64 = time_ms - time_sent;
+                    let rtt = (time_ms - time_sent) as f64;
                     Ok(PingReply {
                         reflector,
+                        measurement_type,
                         seq: sequence,
                         rtt,
                         current_time: time_ms,
-                        down_time: (rtt / 2) as f64,
-                        up_time: (rtt / 2) as f64,
+                        down_time: (rtt / 2.0) as f64,
+                        up_time: (rtt / 2.0) as f64,
                         originate_timestamp: time_sent,
                         receive_timestamp: 0,
                         transmit_timestamp: 0,
                         last_receive_time_s: Instant::now(),
                     })
                 } else {
-                    Err(PingError::InvalidPacket(format!("Packet had type {:?}, but did not match the structure", pkt.typ)))
+                    Err(PingError::InvalidPacket(format!(
+                        "Packet had type {:?}, but did not match the structure",
+                        pkt.typ
+                    )))
                 }
-
-                
-            },
+            }
             type_ => Err(PingError::InvalidType(format!("{:?}", type_))),
         }
     }

src/pinger_icmp_ts.rs

@@ -1,8 +1,9 @@
+use crate::config::MeasurementType;
 use crate::pinger::{PingError, PingListener, PingReply, PingSender};
 use crate::time::Time;
-use icmp_socket2::packet::WithTimestampRequest;
 use icmp_socket2::Icmpv4Message;
 use icmp_socket2::Icmpv4Packet;
+use icmp_socket2::packet::WithTimestampRequest;
 use rustix::thread::ClockId;
 use std::net::IpAddr;
 use std::time::Instant;
@@ -13,7 +14,13 @@ pub struct PingerICMPTimestampSender {}
 
 impl PingListener for PingerICMPTimestampListener {
     // Result: RTT, down time, up time
-    fn parse_packet(&self, id: u16, reflector: IpAddr, pkt: Icmpv4Packet) -> Result<PingReply, PingError> {
+    fn parse_packet(
+        &self,
+        id: u16,
+        reflector: IpAddr,
+        measurement_type: MeasurementType,
+        pkt: Icmpv4Packet,
+    ) -> Result<PingReply, PingError> {
         match pkt.typ {
             // 14 = Timestamp reply
             14 => {
@@ -35,28 +42,30 @@ impl PingListener for PingerICMPTimestampListener {
                     let time_now = Time::new(ClockId::Realtime);
                     let time_since_midnight = time_now.get_time_since_midnight();
 
-                    let rtt: i64 = time_since_midnight - originate as i64;
-                    let dl_time: i64 = time_since_midnight - transmit as i64;
-                    let ul_time: i64 = receive as i64 - originate as i64;
+                    let rtt = (time_since_midnight - originate as i64) as f64;
+                    let dl_time = (time_since_midnight - transmit as i64) as f64;
+                    let ul_time = (receive as i64 - originate as i64) as f64;
 
                     Ok(PingReply {
                         reflector,
+                        measurement_type,
                         seq: sequence,
                         rtt,
                         current_time: time_since_midnight,
-                        down_time: dl_time as f64,
-                        up_time: ul_time as f64,
+                        down_time: dl_time,
+                        up_time: ul_time,
                         originate_timestamp: originate as i64,
                         receive_timestamp: receive as i64,
                         transmit_timestamp: transmit as i64,
                         last_receive_time_s: Instant::now(),
                     })
                 } else {
-                    Err(PingError::InvalidPacket(format!("Packet had type {:?}, but did not match the structure", pkt.typ)))
+                    Err(PingError::InvalidPacket(format!(
+                        "Packet had type {:?}, but did not match the structure",
+                        pkt.typ
+                    )))
                 }
-
-                
-            },
+            }
             type_ => Err(PingError::InvalidType(format!("{:?}", type_))),
         }
     }

src/ratecontroller.rs

@@ -1,3 +1,5 @@
+use crate::SHUTDOWN;
+use crate::metrics::{Metric, MetricsSender};
 use crate::netlink::{Netlink, NetlinkError, Qdisc};
 use crate::time::Time;
 use crate::util::{MutexExt, RwLockExt};
@@ -8,6 +10,7 @@ use std::collections::HashMap;
 use std::fs::File;
 use std::io::Write;
 use std::net::IpAddr;
+use std::sync::atomic::Ordering;
 use std::sync::mpsc::Sender;
 use std::sync::{Arc, Mutex, RwLock};
 use std::thread::sleep;
@@ -108,6 +111,7 @@ pub struct Ratecontroller {
     state_dl: State,
     state_ul: State,
     up_direction: StatsDirection,
+    metrics: MetricsSender,
 }
 
 impl Ratecontroller {
@@ -151,9 +155,7 @@ impl Ratecontroller {
                     / dur.as_secs_f64();
                 state.load = state.utilisation / state.current_rate;
 
-                if state.delta_stat < delay_ms
-                    && state.load > self.config.high_load_level
-                {
+                if state.delta_stat < delay_ms && state.load > self.config.high_load_level {
                     state.safe_rates[state.nrate] = (state.current_rate * state.load).floor();
                     let max_rate = state
                         .safe_rates
@@ -168,10 +170,12 @@ impl Ratecontroller {
                 }
 
                 if state.delta_stat > delay_ms {
-                    match state.safe_rates.get(fastrand::usize(..state.safe_rates.len())) {
+                    match state
+                        .safe_rates
+                        .get(fastrand::usize(..state.safe_rates.len()))
+                    {
                         Some(rnd_rate) => {
-                            state.next_rate =
-                                rnd_rate.min(0.9 * state.current_rate * state.load);
+                            state.next_rate = rnd_rate.min(0.9 * state.current_rate * state.load);
                         }
                         None => {
                             state.next_rate = 0.9 * state.current_rate * state.load;
@@ -246,6 +250,7 @@ impl Ratecontroller {
         reselect_trigger: Sender<bool>,
         down_direction: StatsDirection,
         up_direction: StatsDirection,
+        metrics: MetricsSender,
     ) -> anyhow::Result<Self> {
         let dl_qdisc = Netlink::qdisc_from_ifname(config.download_interface.as_str())?;
         let dl_safe_rates =
@@ -266,6 +271,7 @@ impl Ratecontroller {
             state_dl: State::new(dl_qdisc, cur_rx, dl_safe_rates),
             state_ul: State::new(ul_qdisc, cur_tx, ul_safe_rates),
             up_direction,
+            metrics,
         })
     }
 
@@ -282,10 +288,12 @@ impl Ratecontroller {
         Netlink::set_qdisc_rate(
             self.state_dl.qdisc,
             self.state_dl.current_rate.round() as u64,
+            self.config.dry_run,
         )?;
         Netlink::set_qdisc_rate(
             self.state_ul.qdisc,
             self.state_ul.current_rate.round() as u64,
+            self.config.dry_run,
         )?;
 
         let mut speed_hist_fd: Option<File> = None;
@@ -320,6 +328,10 @@ impl Ratecontroller {
         }
 
         loop {
+            if SHUTDOWN.load(Ordering::Relaxed) {
+                info!("Rate controller shutting down");
+                return Ok(());
+            }
             sleep(sleep_time);
             let now_t = Instant::now();
 
@@ -333,6 +345,12 @@ impl Ratecontroller {
                     warn!(
                         "One or both Netlink stats could not be read. Skipping rate control algorithm"
                     );
+                    self.metrics.send(Metric::Event {
+                        name: "netlink_read_failure",
+                        reason: "",
+                        reflector: None,
+                        tags: &[],
+                    });
                     continue;
                 }
 
@@ -340,16 +358,24 @@ impl Ratecontroller {
 
                 if self.state_dl.deltas.is_empty() || self.state_ul.deltas.is_empty() {
                     warn!("No reflector data available, dropping to minimum rates");
+                    self.metrics.send(Metric::Event {
+                        name: "reflector_unavailable",
+                        reason: "",
+                        reflector: None,
+                        tags: &[],
+                    });
                     self.state_dl.next_rate = self.config.download_min_kbits;
                     self.state_ul.next_rate = self.config.upload_min_kbits;
 
                     Netlink::set_qdisc_rate(
                         self.state_dl.qdisc,
                         self.state_dl.next_rate as u64,
+                        self.config.dry_run,
                     )?;
                     Netlink::set_qdisc_rate(
                         self.state_ul.qdisc,
                         self.state_ul.next_rate as u64,
+                        self.config.dry_run,
                     )?;
 
                     self.state_dl.current_rate = self.state_dl.next_rate;
@@ -370,11 +396,19 @@ impl Ratecontroller {
                 }
 
                 if self.state_dl.next_rate != self.state_dl.current_rate {
-                    Netlink::set_qdisc_rate(self.state_dl.qdisc, self.state_dl.next_rate as u64)?;
+                    Netlink::set_qdisc_rate(
+                        self.state_dl.qdisc,
+                        self.state_dl.next_rate as u64,
+                        self.config.dry_run,
+                    )?;
                 }
 
                 if self.state_ul.next_rate != self.state_ul.current_rate {
-                    Netlink::set_qdisc_rate(self.state_ul.qdisc, self.state_ul.next_rate as u64)?;
+                    Netlink::set_qdisc_rate(
+                        self.state_ul.qdisc,
+                        self.state_ul.next_rate as u64,
+                        self.config.dry_run,
+                    )?;
                 }
 
                 self.state_dl.current_rate = self.state_dl.next_rate;
@@ -393,6 +427,15 @@ impl Ratecontroller {
                     self.state_ul.current_rate as u64
                 );
 
+                self.metrics.send(Metric::Rate {
+                    dl_rate: self.state_dl.current_rate,
+                    ul_rate: self.state_ul.current_rate,
+                    rx_load: self.state_dl.load,
+                    tx_load: self.state_ul.load,
+                    delta_delay_down: self.state_dl.delta_stat,
+                    delta_delay_up: self.state_ul.delta_stat,
+                });
+
                 if let Some(ref mut fd) = stats_fd {
                     if let Err(e) = fd.write_all(
                         format!(

src/reflector_selector.rs

@@ -1,8 +1,11 @@
+use crate::SHUTDOWN;
+use crate::metrics::{Metric, MetricsSender};
 use crate::util::{MutexExt, RwLockExt};
 use crate::{Config, ReflectorStats};
 use log::{debug, info};
 use std::collections::HashMap;
 use std::net::IpAddr;
+use std::sync::atomic::Ordering;
 use std::sync::mpsc::Receiver;
 use std::sync::{Arc, Mutex, RwLock};
 use std::thread::sleep;
@@ -14,6 +17,7 @@ pub struct ReflectorSelector {
     pub reflector_peers_lock: Arc<RwLock<Vec<IpAddr>>>,
     pub reflector_pool: Vec<IpAddr>,
     pub trigger_channel: Receiver<bool>,
+    pub metrics: MetricsSender,
 }
 
 impl ReflectorSelector {
@@ -22,22 +26,31 @@ impl ReflectorSelector {
         let mut reselection_count = 0;
         let baseline_sleep_time =
             Duration::from_secs_f64(self.config.tick_interval * std::f64::consts::PI);
-
         // Initial wait of several seconds to allow some OWD data to build up
         sleep(baseline_sleep_time);
 
         loop {
+            if SHUTDOWN.load(Ordering::Relaxed) {
+                info!("Reflector selector shutting down");
+                return Ok(());
+            }
             /*
              * Selection is triggered either by some other thread triggering it through the channel,
              * or it passes the timeout. In any case we don't care about the result of this function,
              * so we ignore the result of it.
              */
-            let _ = self
+            let triggered = self
                 .trigger_channel
                 .recv_timeout(selector_sleep_time)
-                .unwrap_or(true);
+                .is_ok();
             reselection_count += 1;
             info!("Starting reselection [#{}]", reselection_count);
+            self.metrics.send(Metric::Event {
+                name: "reflector_selection",
+                reason: if triggered { "triggered" } else { "timeout" },
+                reflector: None,
+                tags: &[],
+            });
 
             // After 40 reselections, slow down to every 15 minutes
             if reselection_count > 40 {
@@ -54,7 +67,8 @@ impl ReflectorSelector {
             }
 
             for _ in 0..20 {
-                let next_candidate = &self.reflector_pool[fastrand::usize(..self.reflector_pool.len())];
+                let next_candidate =
+                    &self.reflector_pool[fastrand::usize(..self.reflector_pool.len())];
                 if next_peers.contains(next_candidate) {
                     continue;
                 }
@@ -117,11 +131,15 @@ impl ReflectorSelector {
 
             let mut new_peers = Vec::new();
             for i in 0..num_reflectors {
-                new_peers.push(candidates[i as usize].0);
-                info!(
-                    "New selected peer: {}",
-                    candidates[i as usize].0.to_string()
-                );
+                let peer = candidates[i as usize].0;
+                new_peers.push(peer);
+                info!("New selected peer: {}", peer);
+                self.metrics.send(Metric::Event {
+                    name: "reflector_selected",
+                    reason: "",
+                    reflector: Some(peer),
+                    tags: &[],
+                });
             }
 
             *reflectors_peers = new_peers;

src/time.rs

@@ -28,6 +28,10 @@ impl Time {
         self.time_ns
     }
 
+    pub fn as_nanos(&self) -> u64 {
+        self.time_s * 1_000_000_000 + self.time_ns
+    }
+
     pub fn as_secs_f64(&self) -> f64 {
         self.time_s as f64 + self.time_ns as f64 / 1_000_000_000.0
     }