From 44f551d38cc63924fdd0032a39176eba0b2e20d5 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 13 Apr 2026 15:01:05 +0700 Subject: [PATCH] feat(observability): Add InfluxDB Line Protocol metrics module and config --- Cargo.toml | 4 +- src/config.rs | 115 ++++++++++++++++++++++ src/metrics.rs | 259 ++++++++++++++++++++++++++++++++++++++++++++++++- src/time.rs | 4 + 4 files changed, 376 insertions(+), 6 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index be11b61..fe7709a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ icmp-socket2 = "0.4.1" 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"] \ No newline at end of file +uci = ["dep:rust-uci"] diff --git a/src/config.rs b/src/config.rs index f00ea64..5628d37 100644 --- a/src/config.rs +++ b/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 { + 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, + 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, @@ -138,6 +182,59 @@ impl Config { "sqm-autorate.@output[0].suppress_statistics", Some(false), )?, + + // Observability section + observability_enabled: Self::get::( + "SQMA_OBSERVABILITY_ENABLED", + "sqm-autorate.@observability[0].enabled", + Some(false), + )?, + observability_protocol: Self::get::( + "SQMA_OBSERVABILITY_PROTOCOL", + "sqm-autorate.@observability[0].protocol", + Some(ObservabilityProtocol::Udp), + )?, + observability_host: Self::get_optional::( + "SQMA_OBSERVABILITY_HOST", + "sqm-autorate.@observability[0].host", + ), + observability_port: Self::get::( + "SQMA_OBSERVABILITY_PORT", + "sqm-autorate.@observability[0].port", + Some(8089), + )?, + observability_batch_size: Self::get::( + "SQMA_OBSERVABILITY_BATCH_SIZE", + "sqm-autorate.@observability[0].batch_size", + Some(25), + )?, + observability_batch_timeout_ms: Self::get::( + "SQMA_OBSERVABILITY_BATCH_TIMEOUT_MS", + "sqm-autorate.@observability[0].batch_timeout_ms", + Some(100), + )?, + observability_export_ping_metrics: Self::get::( + "SQMA_OBSERVABILITY_EXPORT_PING_METRICS", + "sqm-autorate.@observability[0].export_ping_metrics", + Some(false), + )?, + observability_export_rate_metrics: Self::get::( + "SQMA_OBSERVABILITY_EXPORT_RATE_METRICS", + "sqm-autorate.@observability[0].export_rate_metrics", + Some(true), + )?, + observability_export_baseline_metrics: Self::get::( + "SQMA_OBSERVABILITY_EXPORT_BASELINE_METRICS", + "sqm-autorate.@observability[0].export_baseline_metrics", + Some(false), + )?, + observability_export_events: Self::get::( + "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::( "SQMA_DOWNLOAD_DELAY_MS", @@ -235,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(env_key: &str, uci_key: &str) -> Option { + Self::get_value(env_key, uci_key).and_then(|val| val.parse::().ok()) + } + fn get_value(env_key: &str, uci_key: &str) -> Option { if let Ok(val) = env::var(env_key) { return Some(val); diff --git a/src/metrics.rs b/src/metrics.rs index 41cb9ff..aa3e782 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -1,6 +1,257 @@ -pub struct Metrics { - pub config: Config, - pub stats_receiver: Receiver, +use log::{error, info, warn}; + +use crate::Config; +use crate::config::{MeasurementType, ObservabilityProtocol}; +use std::fmt::Write; +use std::net::{IpAddr, TcpStream, UdpSocket}; +use std::sync::mpsc::{Receiver, RecvTimeoutError}; +use std::thread::sleep; +use std::time::Duration; + +const MAX_RECONNECT_BACKOFF: u64 = 60; + +enum Transport { + Udp(UdpSocket), + Tcp { + stream: Option, + host: String, + port: u16, + reconnect_backoff: u64, + }, } -impl Metrics {} +impl Transport { + fn new_udp(host: &str, port: u16) -> anyhow::Result { + 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, + timestamp_ns: u64, + }, + Rate { + dl_rate: f64, + ul_rate: f64, + rx_load: f64, + tx_load: f64, + delta_delay_down: f64, + delta_delay_up: f64, + timestamp_ns: u64, + }, + Baseline { + reflector: IpAddr, + baseline_up_ewma: f64, + baseline_down_ewma: f64, + recent_up_ewma: f64, + recent_down_ewma: f64, + timestamp_ns: u64, + }, + Event { + name: &'static str, + reason: &'static str, + reflector: Option, + timestamp_ns: u64, + }, +} + +pub struct Metrics { + pub config: Config, + pub metrics_receiver: Receiver, +} + +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_receiver.recv_timeout(timeout) { + Ok(metric) => batch.push(metric), + Err(RecvTimeoutError::Timeout) => { + 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_receiver.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], host_tag: &str) { + let lines: Vec = batch + .iter() + .map(|m| self.to_influx_line(m, host_tag)) + .collect(); + + let data = lines.join("\n"); + transport.send(&data); + } + + pub fn to_influx_line(&self, metric: &Metric, host_tag: &str) -> String { + match metric { + Metric::Ping { + reflector, + measurement_type, + rtt, + up_time, + down_time, + timestamp_ns, + } => format!( + "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" + ), + Metric::Rate { + dl_rate, + ul_rate, + rx_load, + tx_load, + delta_delay_down, + delta_delay_up, + timestamp_ns, + } => format!( + "sqm_rate,host={host_tag},direction=download \ + rate_kbps={dl:.0},load={rx_load:.4},delta_delay={delta_delay_down:.3} {timestamp_ns}i\n\ + sqm_rate,host={host_tag},direction=upload \ + rate_kbps={ul:.0},load={tx_load:.4},delta_delay={delta_delay_up:.3} {timestamp_ns}i", + dl = dl_rate, + ul = ul_rate, + ), + Metric::Baseline { + reflector, + baseline_up_ewma, + baseline_down_ewma, + recent_up_ewma, + recent_down_ewma, + timestamp_ns, + } => format!( + "sqm_baseline,host={host_tag},reflector={reflector},direction=up baseline_ewma={baseline_up_ewma:.3},recent_ewma={recent_up_ewma:.3} {timestamp_ns}i\n + sqm_baseline,host={host_tag},reflector={reflector},direction=down baseline_ewma={baseline_down_ewma:.3},recent_ewma={recent_down_ewma:.3} {timestamp_ns}i" + ), + Metric::Event { + name, + reason, + reflector, + timestamp_ns, + } => { + let mut tags = format!("host={host_tag},type={name}"); + + if let Some(reflector) = reflector { + write!(tags, ",reflector={reflector}").unwrap_or(()); + } + + if !reason.is_empty() { + write!(tags, ",reason={reason}").unwrap_or(()); + } + format!("sqm_event,{tags} count=1i {timestamp_ns}") + }, + } + } +} diff --git a/src/time.rs b/src/time.rs index 882cdf6..5d08129 100644 --- a/src/time.rs +++ b/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 }