From 6773b6ed4f3815eac9f5d3bdbbb2300cd878b25e Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 24 Sep 2026 17:01:50 +0700 Subject: [PATCH] fix(metrics): keep export failures from delaying rate restoration --- src/main.rs | 10 ++- src/metrics.rs | 236 +++++++++++++++++++++++++++++++++++++++---------- 2 files changed, 199 insertions(+), 47 deletions(-) diff --git a/src/main.rs b/src/main.rs index ad6c000..6ee7a81 100644 --- a/src/main.rs +++ b/src/main.rs @@ -450,7 +450,15 @@ fn run(settings: &Settings) -> anyhow::Result<()> { drop(metrics_tx); if let Some(handle) = metrics_thread_handle { - let _ = handle.join(); + let deadline = Instant::now() + Duration::from_secs(1); + while !handle.is_finished() && Instant::now() < deadline { + sleep(Duration::from_millis(20)); + } + if handle.is_finished() { + let _ = handle.join(); + } else { + warn!("Metrics exporter did not stop within one second"); + } } restore_shaper( diff --git a/src/metrics.rs b/src/metrics.rs index e8d59ac..a1fd40a 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -2,6 +2,7 @@ // // SPDX-License-Identifier: MPL-2.0 +use crate::SHUTDOWN; use crate::settings::Settings; use crate::settings::{MeasurementType, ObservabilityProtocol}; use crate::time::Time; @@ -12,85 +13,152 @@ use std::fmt::Write; use std::net::{IpAddr, TcpStream, UdpSocket}; use std::sync::Arc; use std::sync::atomic::{AtomicU32, Ordering}; -use std::time::Duration; +use std::time::{Duration, Instant}; const MAX_RECONNECT_BACKOFF: u64 = 60; +struct Endpoint { + host: String, + port: u16, + backoff: u64, + next_attempt: Option, +} + +impl Endpoint { + fn new(host: &str, port: u16) -> Self { + Self { + host: host.into(), + port, + backoff: 1, + next_attempt: None, + } + } + + fn failed(&mut self) -> u64 { + let delay = self.backoff; + self.next_attempt = Some(Instant::now() + Duration::from_secs(delay)); + self.backoff = (self.backoff * 2).min(MAX_RECONNECT_BACKOFF); + delay + } + + fn connected(&mut self) { + self.backoff = 1; + self.next_attempt = None; + } +} + enum Transport { - Udp(UdpSocket), + Udp { + socket: Option, + endpoint: Endpoint, + }, Tcp { stream: Option, - host: String, - port: u16, - reconnect_backoff: u64, + endpoint: Endpoint, }, } 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_udp(host: &str, port: u16) -> Self { + Transport::Udp { + socket: None, + endpoint: Endpoint::new(host, port), + } } fn new_tcp(host: &str, port: u16) -> Self { Transport::Tcp { stream: None, - host: host.to_string(), - port, - reconnect_backoff: 1, + endpoint: Endpoint::new(host, port), } } 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::Udp { socket, endpoint } => { + if socket.is_none() { + if endpoint.next_attempt.is_some_and(|at| Instant::now() < at) { + return; + } + let opened = UdpSocket::bind("0.0.0.0:0").and_then(|socket| { + socket.set_write_timeout(Some(Duration::from_millis(100)))?; + socket.connect((endpoint.host.as_str(), endpoint.port))?; + Ok(socket) + }); + match opened { + Ok(opened) => { + *socket = Some(opened); + endpoint.connected(); + } + Err(error) => { + let delay = endpoint.failed(); + warn!( + "Metrics UDP connect to {}:{} failed: {error}; retrying in {delay}s", + endpoint.host, endpoint.port + ); + return; + } + } + } + if let Some(error) = socket + .as_ref() + .and_then(|socket| socket.send(data.as_bytes()).err()) + { + *socket = None; + let delay = endpoint.failed(); + warn!("Metrics UDP send failed: {error}; retrying in {delay}s"); } } - Transport::Tcp { - stream, - host, - port, - reconnect_backoff, - } => { + Transport::Tcp { stream, endpoint } => { if stream.is_none() { - match TcpStream::connect((host.as_str(), *port)) { + if endpoint.next_attempt.is_some_and(|at| Instant::now() < at) { + return; + } + match TcpStream::connect((endpoint.host.as_str(), endpoint.port)) { Ok(s) => { s.set_write_timeout(Some(Duration::from_millis(500))).ok(); - info!("Connected to metrics collector at {}:{}", host, port); + info!( + "Connected to metrics collector at {}:{}", + endpoint.host, endpoint.port + ); *stream = Some(s); - *reconnect_backoff = 1; + endpoint.connected(); } Err(e) => { + let delay = endpoint.failed(); warn!( - "Failed to connect to {}:{} - {}, backoff {}s", - host, port, e, reconnect_backoff + "Metrics TCP connect to {}:{} failed: {e}; retrying in {delay}s", + endpoint.host, endpoint.port ); - *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 let write_error = stream .as_mut() - .and_then(|s| std::io::Write::write_all(s, full_data.as_bytes()).err()); + .and_then(|s| std::io::Write::write_all(s, data.as_bytes()).err()); if let Some(e) = write_error { warn!("TCP send failed: {}", e); *stream = None; - *reconnect_backoff = (*reconnect_backoff * 2).min(MAX_RECONNECT_BACKOFF); + endpoint.failed(); } } } } } +fn escape_tag_value(value: &str) -> String { + let mut escaped = String::with_capacity(value.len()); + for character in value.chars() { + if matches!(character, ',' | '=' | ' ') { + escaped.push('\\'); + } + escaped.push(character); + } + escaped +} + pub enum Metric { Ping { reflector: IpAddr, @@ -180,7 +248,7 @@ impl Metrics { let mut transport = match self.settings.observability.protocol { ObservabilityProtocol::Udp => { info!("Metrics exporter configured for UDP to {}:{}", host, port); - Transport::new_udp(&host, port)? + Transport::new_udp(&host, port) } ObservabilityProtocol::Tcp => { info!("Metrics exporter configured for TCP to {}:{}", host, port); @@ -189,12 +257,30 @@ impl Metrics { }; let mut batch = Vec::with_capacity(batch_size); + let mut shutdown_deadline = None; + let mut last_flush = Instant::now(); loop { - // block until first metric or timeout - match self.metrics_rx.recv_timeout(timeout) { + if SHUTDOWN.load(Ordering::Relaxed) { + let deadline = shutdown_deadline + .get_or_insert_with(|| Instant::now() + Duration::from_millis(400)); + if Instant::now() >= *deadline { + if !batch.is_empty() { + self.flush(&mut transport, &batch, host_tag); + } + break; + } + } + + match self + .metrics_rx + .recv_timeout(timeout.min(Duration::from_millis(100))) + { Ok(metric) => batch.push(metric), Err(RecvTimeoutError::Timeout) => { + if last_flush.elapsed() < timeout && shutdown_deadline.is_none() { + continue; + } let dropped = self.metrics_dropped.swap(0, Ordering::Relaxed); if dropped > 0 { let ts = Time::new(ClockId::Realtime).as_nanos(); @@ -204,9 +290,15 @@ impl Metrics { self.flush(&mut transport, &batch, host_tag); batch.clear(); } + last_flush = Instant::now(); continue; } - Err(RecvTimeoutError::Disconnected) => break, + Err(RecvTimeoutError::Disconnected) => { + if !batch.is_empty() { + self.flush(&mut transport, &batch, host_tag); + } + break; + } } // greedily drain without blocking @@ -220,6 +312,7 @@ impl Metrics { if batch.len() >= batch_size { self.flush(&mut transport, &batch, host_tag); batch.clear(); + last_flush = Instant::now(); } } @@ -227,30 +320,31 @@ impl Metrics { } fn flush(&self, transport: &mut Transport, batch: &[(Metric, u64)], host_tag: &str) { + let host_tag = escape_tag_value(host_tag); 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(_) => { + 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); + 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); + 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) { + fn write_lines(metric: &Metric, timestamp_ns: u64, host_tag: &str, out: &mut String) { match metric { Metric::Ping { reflector, @@ -262,7 +356,7 @@ impl Metrics { 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" + rtt={rtt:.3},up_time={up_time:.3},down_time={down_time:.3} {timestamp_ns}" ) .unwrap_or(()); } @@ -276,11 +370,11 @@ impl Metrics { } => { 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" + rate_kbps={dl_rate:.0},load={rx_load:.4},delta_delay={delta_delay_down:.3} {timestamp_ns}" ).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" + rate_kbps={ul_rate:.0},load={tx_load:.4},delta_delay={delta_delay_up:.3} {timestamp_ns}" ).unwrap_or(()); } Metric::Baseline { @@ -292,11 +386,11 @@ impl Metrics { } => { 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" + baseline_ewma={baseline_up_ewma:.3},recent_ewma={recent_up_ewma:.3} {timestamp_ns}" ).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" + baseline_ewma={baseline_down_ewma:.3},recent_ewma={recent_down_ewma:.3} {timestamp_ns}" ).unwrap_or(()); } Metric::Event { @@ -331,3 +425,53 @@ impl Metrics { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn line_protocol_uses_bare_timestamps_and_escaped_tags() { + let reflector = "192.0.2.1".parse().unwrap(); + let metrics = [ + Metric::Ping { + reflector, + measurement_type: MeasurementType::Icmp, + rtt: 2.0, + up_time: 1.0, + down_time: 1.0, + }, + Metric::Rate { + dl_rate: 100.0, + ul_rate: 50.0, + rx_load: 0.2, + tx_load: 0.3, + delta_delay_down: 1.0, + delta_delay_up: 2.0, + }, + Metric::Baseline { + reflector, + baseline_up_ewma: 1.0, + baseline_down_ewma: 2.0, + recent_up_ewma: 1.0, + recent_down_ewma: 2.0, + }, + Metric::Event { + name: "started", + reason: "", + reflector: None, + tags: &[], + }, + Metric::Dropped { count: 1 }, + ]; + let host = escape_tag_value("a b,c=d"); + for metric in metrics { + let mut output = String::new(); + Metrics::write_lines(&metric, 123, &host, &mut output); + for line in output.lines() { + assert!(line.contains("host=a\\ b\\,c\\=d"), "{line}"); + assert!(line.ends_with(" 123"), "{line}"); + } + } + } +}