From 22019f9f481c6ea99cece1129dc88050b3d3c3bb Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Thu, 29 Oct 2026 15:45:54 -0400 Subject: [PATCH 1/9] feat(metrics): implement lock-free MetricsRegistry and Prometheus 0.0.4 renderer --- bin/yipd/src/lib.rs | 1 + bin/yipd/src/main.rs | 1 + bin/yipd/src/metrics.rs | 736 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 738 insertions(+) create mode 100644 bin/yipd/src/metrics.rs diff --git a/bin/yipd/src/lib.rs b/bin/yipd/src/lib.rs index ec45f11..9589128 100644 --- a/bin/yipd/src/lib.rs +++ b/bin/yipd/src/lib.rs @@ -11,6 +11,7 @@ pub mod flow; pub mod handshake; mod mac_table; pub mod membership; +pub mod metrics; pub mod mode; mod path; pub mod peer_manager; diff --git a/bin/yipd/src/main.rs b/bin/yipd/src/main.rs index c5c4ad4..025c03d 100644 --- a/bin/yipd/src/main.rs +++ b/bin/yipd/src/main.rs @@ -12,6 +12,7 @@ pub(crate) mod flow; mod handshake; mod mac_table; mod membership; +pub mod metrics; mod mode; mod path; mod peer_manager; diff --git a/bin/yipd/src/metrics.rs b/bin/yipd/src/metrics.rs new file mode 100644 index 0000000..6ea6903 --- /dev/null +++ b/bin/yipd/src/metrics.rs @@ -0,0 +1,736 @@ +//! Lock-free metrics registry and Prometheus 0.0.4 exposition formatter. + +use std::fmt::Write; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Instant; + +use serde::{Deserialize, Serialize}; + +/// Lock-free metrics registry holding data-plane counters and gauges. +#[derive(Debug)] +pub struct MetricsRegistry { + start_time: Instant, + // Traffic + tx_bytes: AtomicU64, + rx_bytes: AtomicU64, + tx_packets: AtomicU64, + rx_packets: AtomicU64, + dropped_packets: AtomicU64, + // FEC + fec_tx_symbols: AtomicU64, + fec_recovered_symbols: AtomicU64, + fec_burst_drops: AtomicU64, + // DAITA + daita_padding_bytes: AtomicU64, + daita_decoy_packets: AtomicU64, + daita_decoy_bytes: AtomicU64, + // Crypto + handshakes_success: AtomicU64, + handshakes_failure: AtomicU64, + rekeys: AtomicU64, + current_epoch: AtomicU64, + // Split tunnel + split_bypassed_packets: AtomicU64, + split_bypassed_bytes: AtomicU64, + split_overlay_packets: AtomicU64, + split_overlay_bytes: AtomicU64, +} + +impl Default for MetricsRegistry { + fn default() -> Self { + Self::new() + } +} + +impl MetricsRegistry { + /// Create a new registry with counters initialized to zero. + pub fn new() -> Self { + Self { + start_time: Instant::now(), + tx_bytes: AtomicU64::new(0), + rx_bytes: AtomicU64::new(0), + tx_packets: AtomicU64::new(0), + rx_packets: AtomicU64::new(0), + dropped_packets: AtomicU64::new(0), + fec_tx_symbols: AtomicU64::new(0), + fec_recovered_symbols: AtomicU64::new(0), + fec_burst_drops: AtomicU64::new(0), + daita_padding_bytes: AtomicU64::new(0), + daita_decoy_packets: AtomicU64::new(0), + daita_decoy_bytes: AtomicU64::new(0), + handshakes_success: AtomicU64::new(0), + handshakes_failure: AtomicU64::new(0), + rekeys: AtomicU64::new(0), + current_epoch: AtomicU64::new(1), + split_bypassed_packets: AtomicU64::new(0), + split_bypassed_bytes: AtomicU64::new(0), + split_overlay_packets: AtomicU64::new(0), + split_overlay_bytes: AtomicU64::new(0), + } + } + + /// Record transmitted payload and packets. + pub fn record_tx(&self, bytes: u64, packets: u64) { + self.tx_bytes.fetch_add(bytes, Ordering::Relaxed); + self.tx_packets.fetch_add(packets, Ordering::Relaxed); + } + + /// Record received payload and packets. + pub fn record_rx(&self, bytes: u64, packets: u64) { + self.rx_bytes.fetch_add(bytes, Ordering::Relaxed); + self.rx_packets.fetch_add(packets, Ordering::Relaxed); + } + + /// Record dropped packets. + pub fn record_drop(&self, packets: u64) { + self.dropped_packets.fetch_add(packets, Ordering::Relaxed); + } + + /// Record transmitted Reed-Solomon parity symbols. + pub fn record_fec_tx(&self, symbols: u64) { + self.fec_tx_symbols.fetch_add(symbols, Ordering::Relaxed); + } + + /// Record recovered Reed-Solomon data symbols. + pub fn record_fec_recovered(&self, symbols: u64) { + self.fec_recovered_symbols + .fetch_add(symbols, Ordering::Relaxed); + } + + /// Record unrecoverable FEC burst drops. + pub fn record_fec_burst_drop(&self, bursts: u64) { + self.fec_burst_drops.fetch_add(bursts, Ordering::Relaxed); + } + + /// Record DAITA traffic padding bytes. + pub fn record_daita_padding(&self, bytes: u64) { + self.daita_padding_bytes.fetch_add(bytes, Ordering::Relaxed); + } + + /// Record DAITA decoy cover packets and bytes. + pub fn record_daita_decoy(&self, packets: u64, bytes: u64) { + self.daita_decoy_packets + .fetch_add(packets, Ordering::Relaxed); + self.daita_decoy_bytes.fetch_add(bytes, Ordering::Relaxed); + } + + /// Record a successful Noise handshake completion. + pub fn record_handshake_success(&self) { + self.handshakes_success.fetch_add(1, Ordering::Relaxed); + } + + /// Alias for record_handshake_success. + pub fn record_handshake(&self) { + self.record_handshake_success(); + } + + /// Record a failed Noise handshake attempt. + pub fn record_handshake_failure(&self) { + self.handshakes_failure.fetch_add(1, Ordering::Relaxed); + } + + /// Record an epoch rekey event. + pub fn record_rekey(&self) { + self.rekeys.fetch_add(1, Ordering::Relaxed); + } + + /// Update the current active cryptographic epoch ID. + pub fn set_current_epoch(&self, epoch: u64) { + self.current_epoch.store(epoch, Ordering::Relaxed); + } + + /// Record split-tunnel bypassed direct packets and bytes. + pub fn record_split_tunnel_bypass(&self, bytes: u64, packets: u64) { + self.split_bypassed_bytes + .fetch_add(bytes, Ordering::Relaxed); + self.split_bypassed_packets + .fetch_add(packets, Ordering::Relaxed); + } + + /// Record split-tunnel mesh overlay routed packets and bytes. + pub fn record_split_tunnel_overlay(&self, bytes: u64, packets: u64) { + self.split_overlay_bytes.fetch_add(bytes, Ordering::Relaxed); + self.split_overlay_packets + .fetch_add(packets, Ordering::Relaxed); + } + + /// Take a structured snapshot of all telemetry metrics. + pub fn snapshot(&self, peer_stats: &[crate::peer_manager::PeerStats]) -> TelemetrySnapshot { + let uptime_seconds = self.start_time.elapsed().as_secs(); + let tx_bytes = self.tx_bytes.load(Ordering::Relaxed); + let rx_bytes = self.rx_bytes.load(Ordering::Relaxed); + let tx_packets = self.tx_packets.load(Ordering::Relaxed); + let rx_packets = self.rx_packets.load(Ordering::Relaxed); + let dropped_packets = self.dropped_packets.load(Ordering::Relaxed); + + let fec_tx = self.fec_tx_symbols.load(Ordering::Relaxed); + let fec_recovered = self.fec_recovered_symbols.load(Ordering::Relaxed); + let fec_bursts = self.fec_burst_drops.load(Ordering::Relaxed); + let fec_total = fec_recovered + fec_bursts; + let recovery_rate_ratio = if fec_total > 0 { + fec_recovered as f64 / fec_total as f64 + } else { + 1.0 + }; + + let daita_pad = self.daita_padding_bytes.load(Ordering::Relaxed); + let daita_decoy_pkts = self.daita_decoy_packets.load(Ordering::Relaxed); + let daita_decoy_bytes = self.daita_decoy_bytes.load(Ordering::Relaxed); + let daita_total = tx_bytes + daita_pad; + let padding_ratio = if daita_total > 0 { + daita_pad as f64 / daita_total as f64 + } else { + 0.0 + }; + + let handshakes_success = self.handshakes_success.load(Ordering::Relaxed); + let handshakes_failure = self.handshakes_failure.load(Ordering::Relaxed); + let rekeys = self.rekeys.load(Ordering::Relaxed); + let current_epoch = self.current_epoch.load(Ordering::Relaxed); + + let bypassed_packets = self.split_bypassed_packets.load(Ordering::Relaxed); + let bypassed_bytes = self.split_bypassed_bytes.load(Ordering::Relaxed); + let overlay_packets = self.split_overlay_packets.load(Ordering::Relaxed); + let overlay_bytes = self.split_overlay_bytes.load(Ordering::Relaxed); + + let active_peers = peer_stats + .iter() + .filter(|p| p.state.eq_ignore_ascii_case("established")) + .count(); + let total_peers = peer_stats.len(); + + let peer_items = peer_stats + .iter() + .map(|p| PeerTelemetryItem { + public_key: p.public_key.iter().map(|b| format!("{b:02x}")).collect(), + node_addr: p.node_addr.to_string(), + endpoint: p.endpoint.map(|e| e.to_string()), + state: p.state.clone(), + transport: p.transport.clone(), + tx_bytes: p.tx_bytes, + rx_bytes: p.rx_bytes, + tx_packets: p.tx_packets, + rx_packets: p.rx_packets, + fec_recovered: p.fec_recovered, + rtt_ms: p.rtt_ms, + fec_ratio_pct: p.fec_ratio_pct, + cert_status: p.cert_status.clone(), + }) + .collect(); + + TelemetrySnapshot { + uptime_seconds, + traffic: TrafficTelemetry { + tx_bytes, + rx_bytes, + tx_packets, + rx_packets, + dropped_packets, + }, + fec: FecTelemetry { + symbols_tx: fec_tx, + recovered_symbols: fec_recovered, + unrecoverable_bursts: fec_bursts, + recovery_rate_ratio, + }, + daita: DaitaTelemetry { + padding_bytes: daita_pad, + decoy_packets: daita_decoy_pkts, + decoy_bytes: daita_decoy_bytes, + padding_ratio, + }, + crypto: CryptoTelemetry { + handshakes: handshakes_success, + handshake_failures: handshakes_failure, + rekeys, + current_epoch, + }, + split_tunnel: SplitTunnelTelemetry { + bypassed_packets, + bypassed_bytes, + overlay_packets, + overlay_bytes, + }, + peers: PeersTelemetry { + active: active_peers, + total: total_peers, + list: peer_items, + }, + } + } + + /// Render metrics in standard Prometheus 0.0.4 text format. + pub fn render_prometheus(&self, peer_stats: &[crate::peer_manager::PeerStats]) -> String { + let mut out = String::with_capacity(2048); + + let uptime_seconds = self.start_time.elapsed().as_secs(); + let tx_bytes = self.tx_bytes.load(Ordering::Relaxed); + let rx_bytes = self.rx_bytes.load(Ordering::Relaxed); + let tx_packets = self.tx_packets.load(Ordering::Relaxed); + let rx_packets = self.rx_packets.load(Ordering::Relaxed); + let dropped_packets = self.dropped_packets.load(Ordering::Relaxed); + + let fec_tx = self.fec_tx_symbols.load(Ordering::Relaxed); + let fec_recovered = self.fec_recovered_symbols.load(Ordering::Relaxed); + let fec_bursts = self.fec_burst_drops.load(Ordering::Relaxed); + let fec_total = fec_recovered + fec_bursts; + let recovery_rate_ratio = if fec_total > 0 { + fec_recovered as f64 / fec_total as f64 + } else { + 1.0 + }; + + let daita_pad = self.daita_padding_bytes.load(Ordering::Relaxed); + let daita_decoy_pkts = self.daita_decoy_packets.load(Ordering::Relaxed); + let daita_decoy_bytes = self.daita_decoy_bytes.load(Ordering::Relaxed); + let daita_total = tx_bytes + daita_pad; + let padding_ratio = if daita_total > 0 { + daita_pad as f64 / daita_total as f64 + } else { + 0.0 + }; + + let handshakes_success = self.handshakes_success.load(Ordering::Relaxed); + let handshakes_failure = self.handshakes_failure.load(Ordering::Relaxed); + let rekeys = self.rekeys.load(Ordering::Relaxed); + let current_epoch = self.current_epoch.load(Ordering::Relaxed); + + let bypassed_packets = self.split_bypassed_packets.load(Ordering::Relaxed); + let bypassed_bytes = self.split_bypassed_bytes.load(Ordering::Relaxed); + let overlay_packets = self.split_overlay_packets.load(Ordering::Relaxed); + let overlay_bytes = self.split_overlay_bytes.load(Ordering::Relaxed); + + let active_peers = peer_stats + .iter() + .filter(|p| p.state.eq_ignore_ascii_case("established")) + .count(); + let total_peers = peer_stats.len(); + + // Uptime + out.push_str("# HELP yip_uptime_seconds Process runtime in seconds\n"); + out.push_str("# TYPE yip_uptime_seconds counter\n"); + let _ = writeln!(out, "yip_uptime_seconds {uptime_seconds}"); + + // Peers summary + out.push_str("# HELP yip_peers_active Count of actively handshaked mesh peers\n"); + out.push_str("# TYPE yip_peers_active gauge\n"); + let _ = writeln!(out, "yip_peers_active {active_peers}"); + + out.push_str("# HELP yip_peers_total Count of all configured mesh peers\n"); + out.push_str("# TYPE yip_peers_total gauge\n"); + let _ = writeln!(out, "yip_peers_total {total_peers}"); + + // Traffic totals + out.push_str("# HELP yip_traffic_tx_bytes_total Total bytes transmitted\n"); + out.push_str("# TYPE yip_traffic_tx_bytes_total counter\n"); + let _ = writeln!(out, "yip_traffic_tx_bytes_total {tx_bytes}"); + + out.push_str( + "# HELP yip_traffic_rx_bytes_total Total payload and control bytes received\n", + ); + out.push_str("# TYPE yip_traffic_rx_bytes_total counter\n"); + let _ = writeln!(out, "yip_traffic_rx_bytes_total {rx_bytes}"); + + out.push_str("# HELP yip_traffic_tx_packets_total Total packets dispatched\n"); + out.push_str("# TYPE yip_traffic_tx_packets_total counter\n"); + let _ = writeln!(out, "yip_traffic_tx_packets_total {tx_packets}"); + + out.push_str("# HELP yip_traffic_rx_packets_total Total packets received\n"); + out.push_str("# TYPE yip_traffic_rx_packets_total counter\n"); + let _ = writeln!(out, "yip_traffic_rx_packets_total {rx_packets}"); + + out.push_str("# HELP yip_traffic_dropped_packets_total Packets dropped due to route errors or queue overflow\n"); + out.push_str("# TYPE yip_traffic_dropped_packets_total counter\n"); + let _ = writeln!(out, "yip_traffic_dropped_packets_total {dropped_packets}"); + + // Per-peer metrics + if !peer_stats.is_empty() { + out.push_str("# HELP yip_peer_tx_bytes_total Transmitted bytes per peer\n"); + out.push_str("# TYPE yip_peer_tx_bytes_total counter\n"); + for p in peer_stats { + let _ = writeln!( + out, + "yip_peer_tx_bytes_total{{peer=\"{}\",transport=\"{}\"}} {}", + p.node_addr, p.transport, p.tx_bytes + ); + } + + out.push_str("# HELP yip_peer_rx_bytes_total Received bytes per peer\n"); + out.push_str("# TYPE yip_peer_rx_bytes_total counter\n"); + for p in peer_stats { + let _ = writeln!( + out, + "yip_peer_rx_bytes_total{{peer=\"{}\",transport=\"{}\"}} {}", + p.node_addr, p.transport, p.rx_bytes + ); + } + + out.push_str( + "# HELP yip_peer_rtt_milliseconds Smoothed RTT in milliseconds per peer\n", + ); + out.push_str("# TYPE yip_peer_rtt_milliseconds gauge\n"); + for p in peer_stats { + let rtt = p.rtt_ms.unwrap_or(0); + let _ = writeln!( + out, + "yip_peer_rtt_milliseconds{{peer=\"{}\",transport=\"{}\"}} {}", + p.node_addr, p.transport, rtt + ); + } + + out.push_str( + "# HELP yip_peer_loss_ratio Moving packet loss ratio (0.0 to 1.0) per peer\n", + ); + out.push_str("# TYPE yip_peer_loss_ratio gauge\n"); + for p in peer_stats { + let loss = p.fec_ratio_pct as f64 / 100.0; + let _ = writeln!( + out, + "yip_peer_loss_ratio{{peer=\"{}\",transport=\"{}\"}} {:.4}", + p.node_addr, p.transport, loss + ); + } + } + + // FEC metrics + out.push_str("# HELP yip_fec_symbols_tx_total Total Reed-Solomon symbols transmitted\n"); + out.push_str("# TYPE yip_fec_symbols_tx_total counter\n"); + let _ = writeln!(out, "yip_fec_symbols_tx_total {fec_tx}"); + + out.push_str( + "# HELP yip_fec_recovered_symbols_total Total Reed-Solomon symbols reconstructed\n", + ); + out.push_str("# TYPE yip_fec_recovered_symbols_total counter\n"); + let _ = writeln!(out, "yip_fec_recovered_symbols_total {fec_recovered}"); + + out.push_str( + "# HELP yip_fec_unrecoverable_bursts_total Block drops exceeding parity capacity\n", + ); + out.push_str("# TYPE yip_fec_unrecoverable_bursts_total counter\n"); + let _ = writeln!(out, "yip_fec_unrecoverable_bursts_total {fec_bursts}"); + + out.push_str("# HELP yip_fec_recovery_rate_ratio Ratio of recovered to missing symbols\n"); + out.push_str("# TYPE yip_fec_recovery_rate_ratio gauge\n"); + let _ = writeln!(out, "yip_fec_recovery_rate_ratio {recovery_rate_ratio:.4}"); + + // DAITA metrics + out.push_str( + "# HELP yip_daita_padding_bytes_total Total DAITA dummy padding bytes transmitted\n", + ); + out.push_str("# TYPE yip_daita_padding_bytes_total counter\n"); + let _ = writeln!(out, "yip_daita_padding_bytes_total {daita_pad}"); + + out.push_str( + "# HELP yip_daita_decoy_packets_total Total DAITA cover traffic decoy packets emitted\n", + ); + out.push_str("# TYPE yip_daita_decoy_packets_total counter\n"); + let _ = writeln!(out, "yip_daita_decoy_packets_total {daita_decoy_pkts}"); + + out.push_str("# HELP yip_daita_decoy_bytes_total Total bytes of cover traffic\n"); + out.push_str("# TYPE yip_daita_decoy_bytes_total counter\n"); + let _ = writeln!(out, "yip_daita_decoy_bytes_total {daita_decoy_bytes}"); + + out.push_str( + "# HELP yip_daita_padding_ratio Ratio of defense padding bytes to authentic bytes\n", + ); + out.push_str("# TYPE yip_daita_padding_ratio gauge\n"); + let _ = writeln!(out, "yip_daita_padding_ratio {padding_ratio:.4}"); + + // Crypto metrics + out.push_str( + "# HELP yip_crypto_handshakes_total Successful Noise IK handshakes completed\n", + ); + out.push_str("# TYPE yip_crypto_handshakes_total counter\n"); + let _ = writeln!(out, "yip_crypto_handshakes_total {handshakes_success}"); + + out.push_str("# HELP yip_crypto_handshake_failures_total Failed handshake attempts\n"); + out.push_str("# TYPE yip_crypto_handshake_failures_total counter\n"); + let _ = writeln!( + out, + "yip_crypto_handshake_failures_total {handshakes_failure}" + ); + + out.push_str( + "# HELP yip_crypto_rekeys_total Epoch key ratchets and manual rekeys executed\n", + ); + out.push_str("# TYPE yip_crypto_rekeys_total counter\n"); + let _ = writeln!(out, "yip_crypto_rekeys_total {rekeys}"); + + out.push_str("# HELP yip_crypto_current_epoch Current active cryptographic epoch ID\n"); + out.push_str("# TYPE yip_crypto_current_epoch gauge\n"); + let _ = writeln!(out, "yip_crypto_current_epoch {current_epoch}"); + + // Split tunnel metrics + out.push_str("# HELP yip_split_tunnel_bypassed_packets_total Packets routed directly to physical interface\n"); + out.push_str("# TYPE yip_split_tunnel_bypassed_packets_total counter\n"); + let _ = writeln!( + out, + "yip_split_tunnel_bypassed_packets_total {bypassed_packets}" + ); + + out.push_str( + "# HELP yip_split_tunnel_bypassed_bytes_total Bytes routed outside overlay mesh\n", + ); + out.push_str("# TYPE yip_split_tunnel_bypassed_bytes_total counter\n"); + let _ = writeln!( + out, + "yip_split_tunnel_bypassed_bytes_total {bypassed_bytes}" + ); + + out.push_str("# HELP yip_split_tunnel_overlay_packets_total Packets matched and forwarded across mesh overlay\n"); + out.push_str("# TYPE yip_split_tunnel_overlay_packets_total counter\n"); + let _ = writeln!( + out, + "yip_split_tunnel_overlay_packets_total {overlay_packets}" + ); + + out.push_str( + "# HELP yip_split_tunnel_overlay_bytes_total Bytes routed across mesh overlay\n", + ); + out.push_str("# TYPE yip_split_tunnel_overlay_bytes_total counter\n"); + let _ = writeln!(out, "yip_split_tunnel_overlay_bytes_total {overlay_bytes}"); + + out + } +} + +/// Structured telemetry snapshot suitable for JSON export and OpenTelemetry mapping. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct TelemetrySnapshot { + pub uptime_seconds: u64, + pub traffic: TrafficTelemetry, + pub fec: FecTelemetry, + pub daita: DaitaTelemetry, + pub crypto: CryptoTelemetry, + pub split_tunnel: SplitTunnelTelemetry, + pub peers: PeersTelemetry, +} + +/// Data-plane traffic counters. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] +pub struct TrafficTelemetry { + pub tx_bytes: u64, + pub rx_bytes: u64, + pub tx_packets: u64, + pub rx_packets: u64, + pub dropped_packets: u64, +} + +/// Reed-Solomon FEC telemetry. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] +pub struct FecTelemetry { + pub symbols_tx: u64, + pub recovered_symbols: u64, + pub unrecoverable_bursts: u64, + pub recovery_rate_ratio: f64, +} + +/// DAITA traffic analysis defense telemetry. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] +pub struct DaitaTelemetry { + pub padding_bytes: u64, + pub decoy_packets: u64, + pub decoy_bytes: u64, + pub padding_ratio: f64, +} + +/// Cryptographic session telemetry. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] +pub struct CryptoTelemetry { + pub handshakes: u64, + pub handshake_failures: u64, + pub rekeys: u64, + pub current_epoch: u64, +} + +/// Split-tunnel routing telemetry. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] +pub struct SplitTunnelTelemetry { + pub bypassed_packets: u64, + pub bypassed_bytes: u64, + pub overlay_packets: u64, + pub overlay_bytes: u64, +} + +/// Aggregated peer telemetry. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] +pub struct PeersTelemetry { + pub active: usize, + pub total: usize, + pub list: Vec, +} + +/// Per-peer telemetry details. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] +pub struct PeerTelemetryItem { + pub public_key: String, + pub node_addr: String, + pub endpoint: Option, + pub state: String, + pub transport: String, + pub tx_bytes: u64, + pub rx_bytes: u64, + pub tx_packets: u64, + pub rx_packets: u64, + pub fec_recovered: u64, + pub rtt_ms: Option, + pub fec_ratio_pct: u8, + pub cert_status: String, +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{Ipv6Addr, SocketAddr}; + + #[test] + fn test_atomic_counters_monotonic() { + let registry = MetricsRegistry::new(); + registry.record_tx(100, 1); + registry.record_tx(50, 1); + registry.record_rx(200, 2); + + let snap = registry.snapshot(&[]); + assert_eq!(snap.traffic.tx_bytes, 150); + assert_eq!(snap.traffic.tx_packets, 2); + assert_eq!(snap.traffic.rx_bytes, 200); + assert_eq!(snap.traffic.rx_packets, 2); + } + + #[test] + fn test_prometheus_rendering_format() { + let registry = MetricsRegistry::new(); + registry.record_tx(1500, 3); + registry.record_handshake_success(); + registry.record_rekey(); + + let rendered = registry.render_prometheus(&[]); + assert!(rendered.contains("# HELP yip_traffic_tx_bytes_total Total bytes transmitted")); + assert!(rendered.contains("# TYPE yip_traffic_tx_bytes_total counter")); + assert!(rendered.contains("yip_traffic_tx_bytes_total 1500")); + assert!(rendered.contains("yip_crypto_handshakes_total 1")); + assert!(rendered.contains("yip_crypto_rekeys_total 1")); + assert!(rendered.ends_with('\n')); + } + + #[test] + fn test_fec_and_daita_metrics() { + let registry = MetricsRegistry::new(); + registry.record_fec_tx(64); + registry.record_fec_recovered(16); + registry.record_fec_burst_drop(4); + + registry.record_daita_padding(512); + registry.record_daita_decoy(2, 256); + + let snap = registry.snapshot(&[]); + assert_eq!(snap.fec.symbols_tx, 64); + assert_eq!(snap.fec.recovered_symbols, 16); + assert_eq!(snap.fec.unrecoverable_bursts, 4); + assert!((snap.fec.recovery_rate_ratio - 0.8).abs() < 1e-6); + + assert_eq!(snap.daita.padding_bytes, 512); + assert_eq!(snap.daita.decoy_packets, 2); + assert_eq!(snap.daita.decoy_bytes, 256); + + let rendered = registry.render_prometheus(&[]); + assert!(rendered.contains("yip_fec_symbols_tx_total 64")); + assert!(rendered.contains("yip_fec_recovered_symbols_total 16")); + assert!(rendered.contains("yip_fec_unrecoverable_bursts_total 4")); + assert!(rendered.contains("yip_fec_recovery_rate_ratio 0.8000")); + assert!(rendered.contains("yip_daita_padding_bytes_total 512")); + assert!(rendered.contains("yip_daita_decoy_packets_total 2")); + assert!(rendered.contains("yip_daita_decoy_bytes_total 256")); + } + + #[test] + fn test_split_tunnel_and_crypto_metrics() { + let registry = MetricsRegistry::new(); + registry.record_split_tunnel_bypass(1024, 8); + registry.record_split_tunnel_overlay(2048, 16); + registry.record_handshake_failure(); + registry.set_current_epoch(42); + + let snap = registry.snapshot(&[]); + assert_eq!(snap.split_tunnel.bypassed_bytes, 1024); + assert_eq!(snap.split_tunnel.bypassed_packets, 8); + assert_eq!(snap.split_tunnel.overlay_bytes, 2048); + assert_eq!(snap.split_tunnel.overlay_packets, 16); + assert_eq!(snap.crypto.handshake_failures, 1); + assert_eq!(snap.crypto.current_epoch, 42); + + let rendered = registry.render_prometheus(&[]); + assert!(rendered.contains("yip_split_tunnel_bypassed_packets_total 8")); + assert!(rendered.contains("yip_split_tunnel_bypassed_bytes_total 1024")); + assert!(rendered.contains("yip_split_tunnel_overlay_packets_total 16")); + assert!(rendered.contains("yip_split_tunnel_overlay_bytes_total 2048")); + assert!(rendered.contains("yip_crypto_handshake_failures_total 1")); + assert!(rendered.contains("yip_crypto_current_epoch 42")); + } + + #[test] + fn test_peer_stats_rendering() { + let registry = MetricsRegistry::new(); + let peer = crate::peer_manager::PeerStats { + public_key: [0xab; 32], + endpoint: Some("192.168.1.100:51820".parse::().unwrap()), + node_addr: "fd00::1".parse::().unwrap(), + state: "Established".into(), + last_activity_ms: 12345, + tx_bytes: 8192, + rx_bytes: 4096, + tx_packets: 10, + rx_packets: 5, + fec_recovered: 2, + transport: "Direct UDP".into(), + rtt_ms: Some(15), + fec_ratio_pct: 5, + cert_status: "CA Valid".into(), + }; + + let snap = registry.snapshot(std::slice::from_ref(&peer)); + assert_eq!(snap.peers.active, 1); + assert_eq!(snap.peers.total, 1); + assert_eq!(snap.peers.list.len(), 1); + assert_eq!(snap.peers.list[0].node_addr, "fd00::1"); + + let rendered = registry.render_prometheus(std::slice::from_ref(&peer)); + assert!(rendered.contains("yip_peers_active 1")); + assert!(rendered.contains("yip_peers_total 1")); + assert!(rendered + .contains("yip_peer_tx_bytes_total{peer=\"fd00::1\",transport=\"Direct UDP\"} 8192")); + assert!(rendered + .contains("yip_peer_rx_bytes_total{peer=\"fd00::1\",transport=\"Direct UDP\"} 4096")); + assert!(rendered + .contains("yip_peer_rtt_milliseconds{peer=\"fd00::1\",transport=\"Direct UDP\"} 15")); + assert!(rendered + .contains("yip_peer_loss_ratio{peer=\"fd00::1\",transport=\"Direct UDP\"} 0.0500")); + } + + #[test] + fn test_concurrent_counter_increments() { + use std::sync::Arc; + let registry = Arc::new(MetricsRegistry::new()); + let mut handles = Vec::new(); + + for _ in 0..10 { + let reg = Arc::clone(®istry); + handles.push(std::thread::spawn(move || { + for _ in 0..100 { + reg.record_tx(10, 1); + reg.record_rx(20, 1); + } + })); + } + + for h in handles { + h.join().unwrap(); + } + + let snap = registry.snapshot(&[]); + assert_eq!(snap.traffic.tx_bytes, 10000); + assert_eq!(snap.traffic.tx_packets, 1000); + assert_eq!(snap.traffic.rx_bytes, 20000); + assert_eq!(snap.traffic.rx_packets, 1000); + } +} From 9ae1dcba766353f97c084d1aaddf61bbf6644f59 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Thu, 29 Oct 2026 15:55:53 -0400 Subject: [PATCH 2/9] feat(metrics): add embedded HTTP/1.1 scraper server for Prometheus and healthz --- bin/yipd/src/metrics.rs | 340 +++++++++++++++++++++++++++++++++++++++- 1 file changed, 338 insertions(+), 2 deletions(-) diff --git a/bin/yipd/src/metrics.rs b/bin/yipd/src/metrics.rs index 6ea6903..e1cc9b1 100644 --- a/bin/yipd/src/metrics.rs +++ b/bin/yipd/src/metrics.rs @@ -1,8 +1,12 @@ //! Lock-free metrics registry and Prometheus 0.0.4 exposition formatter. use std::fmt::Write; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::Instant; +use std::io::{BufRead, BufReader, Write as IoWrite}; +use std::net::{SocketAddr, TcpListener, TcpStream}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; use serde::{Deserialize, Serialize}; @@ -578,6 +582,184 @@ pub struct PeerTelemetryItem { pub cert_status: String, } +/// Embedded HTTP/1.1 scraper server exposing Prometheus metrics and health check. +pub struct MetricsHttpServer { + local_addr: SocketAddr, + shutdown: Arc, + thread_handle: Mutex>>, +} + +impl std::fmt::Debug for MetricsHttpServer { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("MetricsHttpServer") + .field("local_addr", &self.local_addr) + .field("shutdown", &self.shutdown.load(Ordering::Relaxed)) + .finish() + } +} + +impl MetricsHttpServer { + /// Start the HTTP scraper server bound to `addr`. + pub fn start( + addr: SocketAddr, + registry: Arc, + stats_fn: Box Vec + Send + Sync + 'static>, + ) -> std::io::Result { + let listener = TcpListener::bind(addr)?; + let local_addr = listener.local_addr()?; + listener.set_nonblocking(true)?; + + let shutdown = Arc::new(AtomicBool::new(false)); + let shutdown_clone = Arc::clone(&shutdown); + let stats_fn: Arc Vec + Send + Sync + 'static> = + Arc::from(stats_fn); + + let thread_handle = std::thread::Builder::new() + .name("metrics-http-server".into()) + .spawn(move || { + while !shutdown_clone.load(Ordering::Relaxed) { + match listener.accept() { + Ok((stream, _)) => { + let reg = Arc::clone(®istry); + let s_fn = Arc::clone(&stats_fn); + let _ = std::thread::Builder::new() + .name("metrics-http-conn".into()) + .spawn(move || { + let _ = handle_client(stream, ®, &*s_fn); + }); + } + Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { + std::thread::sleep(Duration::from_millis(10)); + } + Err(_) => { + std::thread::sleep(Duration::from_millis(10)); + } + } + } + })?; + + Ok(Self { + local_addr, + shutdown, + thread_handle: Mutex::new(Some(thread_handle)), + }) + } + + /// Return the local socket address this server is listening on. + pub fn local_addr(&self) -> SocketAddr { + self.local_addr + } + + /// Signal shutdown to the background server thread and await its termination. + pub fn stop(&self) { + if self.shutdown.swap(true, Ordering::SeqCst) { + return; + } + if let Ok(mut lock) = self.thread_handle.lock() { + if let Some(handle) = lock.take() { + let _ = handle.join(); + } + } + } +} + +impl Drop for MetricsHttpServer { + fn drop(&mut self) { + self.stop(); + } +} + +fn handle_client( + mut stream: TcpStream, + registry: &MetricsRegistry, + stats_fn: &(dyn Fn() -> Vec + Send + Sync), +) -> std::io::Result<()> { + let _ = stream.set_nonblocking(false); + stream.set_read_timeout(Some(Duration::from_millis(3000)))?; + stream.set_write_timeout(Some(Duration::from_millis(3000)))?; + + let (method, path) = { + let mut reader = BufReader::new(&mut stream); + let mut first_line = String::new(); + if reader.read_line(&mut first_line)? == 0 { + return Ok(()); + } + + let mut header_line = String::new(); + let mut header_count = 0; + while header_count < 64 && reader.read_line(&mut header_line)? > 0 { + let trimmed = header_line.trim_end_matches(['\r', '\n']); + if trimmed.is_empty() { + break; + } + header_line.clear(); + header_count += 1; + } + + let parts: Vec<&str> = first_line.split_whitespace().collect(); + if parts.len() < 2 { + ("".to_string(), "".to_string()) + } else { + ( + parts[0].to_string(), + parts[1].split('?').next().unwrap_or(parts[1]).to_string(), + ) + } + }; + + if method.is_empty() { + let response = "HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"; + IoWrite::write_all(&mut stream, response.as_bytes())?; + IoWrite::flush(&mut stream)?; + return Ok(()); + } + + if method != "GET" && method != "HEAD" { + let body = "Method Not Allowed"; + let response = format!( + "HTTP/1.1 405 Method Not Allowed\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + IoWrite::write_all(&mut stream, response.as_bytes())?; + IoWrite::flush(&mut stream)?; + return Ok(()); + } + + match path.as_str() { + "/metrics" => { + let peer_stats = stats_fn(); + let body = registry.render_prometheus(&peer_stats); + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: text/plain; version=0.0.4; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + if method == "HEAD" { "" } else { &body } + ); + IoWrite::write_all(&mut stream, response.as_bytes())?; + } + "/healthz" => { + let body = "ok"; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + if method == "HEAD" { "" } else { body } + ); + IoWrite::write_all(&mut stream, response.as_bytes())?; + } + _ => { + let body = "Not Found"; + let response = format!( + "HTTP/1.1 404 Not Found\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + if method == "HEAD" { "" } else { body } + ); + IoWrite::write_all(&mut stream, response.as_bytes())?; + } + } + IoWrite::flush(&mut stream)?; + Ok(()) +} + #[cfg(test)] mod tests { use super::*; @@ -733,4 +915,158 @@ mod tests { assert_eq!(snap.traffic.rx_bytes, 20000); assert_eq!(snap.traffic.rx_packets, 1000); } + + #[test] + fn test_http_metrics_and_healthz_endpoint() { + use std::io::{Read, Write}; + use std::sync::Arc; + + let registry = Arc::new(MetricsRegistry::new()); + registry.record_tx(4096, 4); + + let server = MetricsHttpServer::start( + "127.0.0.1:0".parse().unwrap(), + registry.clone(), + Box::new(Vec::new), + ) + .expect("failed to start metrics http server"); + + let port = server.local_addr().port(); + + // Test GET /metrics + let mut stream = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap(); + stream + .write_all(b"GET /metrics HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .unwrap(); + let mut resp = String::new(); + stream.read_to_string(&mut resp).unwrap(); + assert!(resp.starts_with("HTTP/1.1 200 OK")); + assert!(resp.contains("Content-Type: text/plain; version=0.0.4")); + assert!(resp.contains("yip_traffic_tx_bytes_total 4096")); + + // Test GET /healthz + let mut stream = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap(); + stream + .write_all(b"GET /healthz HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .unwrap(); + let mut resp = String::new(); + stream.read_to_string(&mut resp).unwrap(); + assert!(resp.starts_with("HTTP/1.1 200 OK")); + assert!(resp.ends_with("ok")); + + server.stop(); + } + + #[test] + fn test_http_not_found_and_method_not_allowed() { + use std::io::{Read, Write}; + use std::sync::Arc; + + let registry = Arc::new(MetricsRegistry::new()); + let server = + MetricsHttpServer::start("127.0.0.1:0".parse().unwrap(), registry, Box::new(Vec::new)) + .expect("failed to start server"); + + let port = server.local_addr().port(); + + // Test 404 on unknown route + let mut stream = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap(); + stream + .write_all(b"GET /notfound HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .unwrap(); + let mut resp = String::new(); + stream.read_to_string(&mut resp).unwrap(); + assert!(resp.starts_with("HTTP/1.1 404 Not Found")); + assert!(resp.ends_with("Not Found")); + + // Test 405 on POST + let mut stream = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap(); + stream + .write_all(b"POST /metrics HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .unwrap(); + let mut resp = String::new(); + stream.read_to_string(&mut resp).unwrap(); + assert!(resp.starts_with("HTTP/1.1 405 Method Not Allowed")); + + // Test HEAD on /healthz + let mut stream = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap(); + stream + .write_all(b"HEAD /healthz HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .unwrap(); + let mut resp = String::new(); + stream.read_to_string(&mut resp).unwrap(); + assert!(resp.starts_with("HTTP/1.1 200 OK")); + assert!(resp.contains("Content-Length: 2")); + // HEAD must not include response body + assert!(resp.ends_with("\r\n\r\n")); + + server.stop(); + } + + #[test] + fn test_http_server_stop_idempotent_and_drop() { + use std::sync::Arc; + + let registry = Arc::new(MetricsRegistry::new()); + let server = MetricsHttpServer::start( + "127.0.0.1:0".parse().unwrap(), + registry.clone(), + Box::new(Vec::new), + ) + .expect("failed to start server"); + + // Idempotent stop calls + server.stop(); + server.stop(); + + // Clean drop without explicit stop + let server2 = + MetricsHttpServer::start("127.0.0.1:0".parse().unwrap(), registry, Box::new(Vec::new)) + .expect("failed to start server2"); + drop(server2); + } + + #[test] + fn test_http_metrics_with_live_peers() { + use std::io::{Read, Write}; + use std::sync::Arc; + + let registry = Arc::new(MetricsRegistry::new()); + let peer = crate::peer_manager::PeerStats { + public_key: [0x11; 32], + endpoint: Some("127.0.0.1:51820".parse().unwrap()), + node_addr: "fd00::99".parse().unwrap(), + state: "established".into(), + last_activity_ms: 100, + tx_bytes: 512, + rx_bytes: 256, + tx_packets: 4, + rx_packets: 2, + fec_recovered: 1, + transport: "Direct UDP".into(), + rtt_ms: Some(25), + fec_ratio_pct: 10, + cert_status: "CA Valid".into(), + }; + + let server = MetricsHttpServer::start( + "127.0.0.1:0".parse().unwrap(), + registry, + Box::new(move || vec![peer.clone()]), + ) + .expect("failed to start server"); + + let port = server.local_addr().port(); + let mut stream = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap(); + stream + .write_all(b"GET /metrics HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .unwrap(); + let mut resp = String::new(); + stream.read_to_string(&mut resp).unwrap(); + assert!(resp.contains("yip_peers_active 1")); + assert!(resp + .contains("yip_peer_tx_bytes_total{peer=\"fd00::99\",transport=\"Direct UDP\"} 512")); + + server.stop(); + } } From 09aff7ef3e8955c6af0333d34f55fee195394811 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Thu, 29 Oct 2026 16:06:33 -0400 Subject: [PATCH 3/9] feat(ipc): add get_metrics command and run_cli_metrics dispatch --- bin/yipd/src/control.rs | 147 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 147 insertions(+) diff --git a/bin/yipd/src/control.rs b/bin/yipd/src/control.rs index c09947e..ef413c2 100644 --- a/bin/yipd/src/control.rs +++ b/bin/yipd/src/control.rs @@ -98,6 +98,11 @@ pub enum ControlCmd { #[serde(default = "default_ping_timeout_ms")] timeout_ms: u64, }, + #[serde(rename = "get_metrics")] + GetMetrics { + #[serde(default)] + peer: Option, + }, } fn default_ping_count() -> u32 { @@ -652,6 +657,11 @@ pub enum ControlResponse { stddev_rtt_ms: Option, samples: Vec, }, + Metrics { + ok: bool, + metrics: crate::metrics::TelemetrySnapshot, + prometheus: String, + }, Ack { ok: bool, message: String, @@ -1213,6 +1223,27 @@ pub fn dispatch_cmd( error: e, }, }, + ControlCmd::GetMetrics { peer } => { + let mut stats = pm.peer_stats(); + if let Some(target) = peer { + stats.retain(|p| { + p.node_addr.to_string() == target + || p.public_key + .iter() + .map(|b| format!("{b:02x}")) + .collect::() + .starts_with(&target) + }); + } + let registry = crate::metrics::MetricsRegistry::new(); + let snapshot = registry.snapshot(&stats); + let prometheus = registry.render_prometheus(&stats); + ControlResponse::Metrics { + ok: true, + metrics: snapshot, + prometheus, + } + } } } @@ -2008,6 +2039,33 @@ pub fn run_cli_split_set( } } +/// Query and display telemetry metrics. +pub fn run_cli_metrics(socket_path: &Path, json: bool) -> Result<(), String> { + let resp = client_query(socket_path, ControlCmd::GetMetrics { peer: None })?; + match resp { + ControlResponse::Metrics { + ok, + metrics, + prometheus, + } => { + if !ok { + return Err("daemon reported metrics error".into()); + } + if json { + let text = serde_json::to_string_pretty(&metrics).map_err(|e| e.to_string())?; + println!("{text}"); + } else if prometheus.ends_with('\n') { + print!("{prometheus}"); + } else { + println!("{prometheus}"); + } + Ok(()) + } + ControlResponse::Err { error, .. } => Err(error), + _ => Err("unexpected response format from daemon".into()), + } +} + struct CursorGuard; impl Drop for CursorGuard { fn drop(&mut self) { @@ -2586,6 +2644,8 @@ mod tests { None ) .is_err()); + assert!(run_cli_metrics(path, false).is_ok()); + assert!(run_cli_metrics(path, true).is_ok()); running.store(false, Ordering::Relaxed); let _ = srv.join(); @@ -3076,4 +3136,91 @@ mod tests { other => panic!("expected TracePath response, got: {other:?}"), } } + + #[test] + fn test_control_protocol_get_metrics_serde() { + let cmd = ControlCmd::GetMetrics { + peer: Some("node1".into()), + }; + let json = serde_json::to_string(&cmd).unwrap(); + assert!(json.contains("\"cmd\":\"get_metrics\"")); + let parsed: ControlCmd = serde_json::from_str(&json).unwrap(); + match parsed { + ControlCmd::GetMetrics { peer } => assert_eq!(peer.as_deref(), Some("node1")), + _ => panic!("wrong variant"), + } + + // Test omitting peer field + let json_no_peer = r#"{"cmd":"get_metrics"}"#; + let parsed_no_peer: ControlCmd = serde_json::from_str(json_no_peer).unwrap(); + match parsed_no_peer { + ControlCmd::GetMetrics { peer } => assert_eq!(peer, None), + _ => panic!("wrong variant"), + } + + // Test ControlResponse::Metrics serde + let reg = crate::metrics::MetricsRegistry::new(); + let snapshot = reg.snapshot(&[]); + let prom_text = reg.render_prometheus(&[]); + let resp = ControlResponse::Metrics { + ok: true, + metrics: snapshot.clone(), + prometheus: prom_text.clone(), + }; + let resp_json = serde_json::to_string(&resp).unwrap(); + let parsed_resp: ControlResponse = serde_json::from_str(&resp_json).unwrap(); + match parsed_resp { + ControlResponse::Metrics { + ok, + metrics, + prometheus, + } => { + assert!(ok); + assert_eq!(metrics, snapshot); + assert_eq!(prometheus, prom_text); + } + other => panic!("expected ControlResponse::Metrics, got: {other:?}"), + } + } + + #[test] + fn test_run_cli_metrics_error_handling() { + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + + let sock_path = format!("/tmp/test_cli_metrics_err_{}.sock", std::process::id()); + let _ = std::fs::remove_file(&sock_path); + let handle = ControlHandle::bind(&sock_path).unwrap(); + + let running = Arc::new(AtomicBool::new(true)); + let running_clone = running.clone(); + let srv = std::thread::spawn(move || { + while running_clone.load(Ordering::Relaxed) { + if let Ok(req) = handle.try_recv() { + let resp = ControlResponse::Err { + ok: false, + error: "internal daemon error".into(), + }; + let _ = req.responder.send(resp); + } + std::thread::sleep(std::time::Duration::from_millis(5)); + } + }); + + std::thread::sleep(std::time::Duration::from_millis(50)); + let path = Path::new(&sock_path); + + let err = run_cli_metrics(path, false).unwrap_err(); + assert_eq!(err, "internal daemon error"); + + let err_json = run_cli_metrics(path, true).unwrap_err(); + assert_eq!(err_json, "internal daemon error"); + + running.store(false, Ordering::Relaxed); + let _ = srv.join(); + let _ = std::fs::remove_file(&sock_path); + + // Non-existent socket error + assert!(run_cli_metrics(Path::new("/tmp/nonexistent_metrics.sock"), false).is_err()); + } } From ddcb76a29d1d5fe64d29df7e05a931f241ac34d6 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Thu, 29 Oct 2026 20:19:57 -0400 Subject: [PATCH 4/9] feat(dataplane): wire lock-free MetricsRegistry into PeerManager --- bin/yipd/src/control.rs | 2 +- bin/yipd/src/peer_manager/handshake.rs | 13 ++++ bin/yipd/src/peer_manager/mod.rs | 82 +++++++++++++++++++++---- bin/yipd/src/peer_manager/obf.rs | 3 + bin/yipd/src/peer_manager/rekey.rs | 10 +++- bin/yipd/src/peer_manager/relay.rs | 25 ++++---- bin/yipd/src/peer_manager/routing.rs | 37 ++++++++++-- bin/yipd/src/peer_manager/tests.rs | 83 ++++++++++++++++++++++++++ bin/yipd/src/peer_manager/testutil.rs | 4 ++ 9 files changed, 231 insertions(+), 28 deletions(-) diff --git a/bin/yipd/src/control.rs b/bin/yipd/src/control.rs index ef413c2..1e07151 100644 --- a/bin/yipd/src/control.rs +++ b/bin/yipd/src/control.rs @@ -1235,7 +1235,7 @@ pub fn dispatch_cmd( .starts_with(&target) }); } - let registry = crate::metrics::MetricsRegistry::new(); + let registry = pm.metrics(); let snapshot = registry.snapshot(&stats); let prometheus = registry.render_prometheus(&stats); ControlResponse::Metrics { diff --git a/bin/yipd/src/peer_manager/handshake.rs b/bin/yipd/src/peer_manager/handshake.rs index 0e5660c..940ccf5 100644 --- a/bin/yipd/src/peer_manager/handshake.rs +++ b/bin/yipd/src/peer_manager/handshake.rs @@ -44,6 +44,7 @@ impl PeerManager { Ok(t) => t, Err(e) => { eprintln!("peer_manager: failed to start handshake: {e}"); + self.metrics.record_handshake_failure(); return None; } }; @@ -194,6 +195,7 @@ impl PeerManager { Ok(t) => t, Err(e) => { eprintln!("peer_manager: start_responder failed: {e}"); + self.metrics.record_handshake_failure(); return DispatchOut::None; } }; @@ -361,13 +363,18 @@ impl PeerManager { }); let pending = std::mem::take(&mut self.peers[idx].pending_tun); for inner in &pending { + self.peers[idx].tx_bytes += inner.len() as u64; + self.peers[idx].tx_packets += 1; + self.metrics.record_tx(inner.len() as u64, 1); let out = dp.on_tun_packet(inner, now_ms); + self.metrics.record_fec_tx(out.len() as u64); self.egress.extend(out.iter().cloned()); } let mut epoch_set = Box::new(crate::epoch::EpochSet::new(dp, now_ms)); epoch_set.set_fec_repair_ratio(self.fec_config.redundancy_pct as f32 / 100.0); self.peers[idx].state = PeerState::Established(epoch_set); self.record_new_epoch(idx); + self.metrics.record_handshake_success(); DispatchOut::Udp(&self.egress) } @@ -450,13 +457,18 @@ impl PeerManager { self.egress.clear(); let pending = std::mem::take(&mut self.peers[idx].pending_tun); for inner in &pending { + self.peers[idx].tx_bytes += inner.len() as u64; + self.peers[idx].tx_packets += 1; + self.metrics.record_tx(inner.len() as u64, 1); let out = dp.on_tun_packet(inner, now_ms); + self.metrics.record_fec_tx(out.len() as u64); self.egress.extend(out.iter().cloned()); } let mut epoch_set = Box::new(crate::epoch::EpochSet::new(dp, now_ms)); epoch_set.set_fec_repair_ratio(self.fec_config.redundancy_pct as f32 / 100.0); self.peers[idx].state = PeerState::Established(epoch_set); self.record_new_epoch(idx); + self.metrics.record_handshake_success(); if self.egress.is_empty() { DispatchOut::None @@ -466,6 +478,7 @@ impl PeerManager { } Err(e) => { eprintln!("peer_manager: read_response failed: {e}"); + self.metrics.record_handshake_failure(); // State was already reverted to `Idle` above (via the // `mem::replace`); `pending_tun` stays queued and the next // `on_tun` call will start a fresh handshake. diff --git a/bin/yipd/src/peer_manager/mod.rs b/bin/yipd/src/peer_manager/mod.rs index 8dbf228..3908581 100644 --- a/bin/yipd/src/peer_manager/mod.rs +++ b/bin/yipd/src/peer_manager/mod.rs @@ -506,6 +506,8 @@ pub struct PeerManager { pub daita_stats: DaitaStats, /// Split-tunnel routing and traffic bypass engine (#184). pub split_tunnel: crate::control::SplitTunnelState, + /// Shared lock-free metrics registry for daemon and telemetry reporting. + metrics: std::sync::Arc, } /// Dynamic DAITA traffic analysis defense runtime metrics. @@ -650,6 +652,7 @@ impl PeerManager { fec_config: crate::control::FecConfig::default(), daita_stats: DaitaStats::default(), split_tunnel: crate::control::SplitTunnelState::default(), + metrics: std::sync::Arc::new(crate::metrics::MetricsRegistry::new()), }; // Roots are pre-vetted (CA-signed root set) and therefore always-admit, @@ -1217,6 +1220,50 @@ impl PeerManager { self.start_time = start_time; } + /// Access the shared lock-free telemetry metrics registry. + pub fn metrics(&self) -> std::sync::Arc { + std::sync::Arc::clone(&self.metrics) + } + + /// Set or inject a custom metrics registry. + #[cfg_attr( + not(test), + expect(dead_code, reason = "metrics registry injection helper") + )] + pub fn set_metrics(&mut self, metrics: std::sync::Arc) { + self.metrics = metrics; + } + + /// Builder method to inject a metrics registry. + #[cfg_attr( + not(test), + expect(dead_code, reason = "metrics registry builder helper") + )] + pub fn with_metrics( + mut self, + metrics: std::sync::Arc, + ) -> Self { + self.metrics = metrics; + self + } + + /// Record test packet transmission for data-plane telemetry testing. + #[cfg_attr( + not(test), + expect(dead_code, reason = "test-only packet instrumentation helper") + )] + pub fn record_test_tx_packet(&self, bytes: u64) { + self.metrics.record_tx(bytes, 1); + } + + /// Handle an inbound control command directly on PeerManager. + pub fn handle_control_cmd( + &mut self, + cmd: crate::control::ControlCmd, + ) -> crate::control::ControlResponse { + crate::control::dispatch_cmd(self, cmd, &self.device_name.clone(), self.start_time) + } + /// Configure online Threshold CA coordinator address for automated 70% TTL certificate rotation. pub fn set_threshold_ca_server(&mut self, addr: Option) { self.threshold_ca_server = addr; @@ -1718,6 +1765,7 @@ impl Dispatch for PeerManager { }; } if dg.is_empty() { + self.metrics.record_drop(1); return DispatchOut::None; } // A datagram from the configured rendezvous server is an obfuscated @@ -1740,6 +1788,7 @@ impl Dispatch for PeerManager { return self.finish_wrapped(tun, udp, now_ms); } let Some(plain) = self.deobf_ingress(src, dg) else { + self.metrics.record_drop(1); return DispatchOut::None; }; let (tun, udp) = own_dispatch(self.on_udp_dispatch(src, &plain, now_ms)); @@ -1804,6 +1853,7 @@ impl Dispatch for PeerManager { impl PeerManager { fn on_udp_dispatch(&mut self, src: SocketAddr, dg: &[u8], now_ms: u64) -> DispatchOut<'_> { if dg.is_empty() { + self.metrics.record_drop(1); return DispatchOut::None; } // Rendezvous-server demux: a datagram from the configured server is a @@ -1832,10 +1882,14 @@ impl PeerManager { if let Some(dst) = crate::control::extract_ip_dst(inner) { if self.split_tunnel.should_bypass(dst) { self.split_tunnel.record_bypassed(inner.len() as u64); + self.metrics + .record_split_tunnel_bypass(inner.len() as u64, 1); return &[]; } } self.split_tunnel.record_routed(inner.len() as u64); + self.metrics + .record_split_tunnel_overlay(inner.len() as u64, 1); let idx = match self.route_tun_index(inner) { Some(i) => i, @@ -1857,10 +1911,16 @@ impl PeerManager { // now resolves to it; re-drive the normal path below. match self.route_tun_index(inner) { Some(i) => i, - None => return &[], + None => { + self.metrics.record_drop(1); + return &[]; + } } } - None => return &[], + None => { + self.metrics.record_drop(1); + return &[]; + } } } }; @@ -1879,6 +1939,7 @@ impl PeerManager { self.peers[idx].last_activity_ms = now_ms; self.peers[idx].tx_bytes += inner.len() as u64; self.peers[idx].tx_packets += 1; + self.metrics.record_tx(inner.len() as u64, 1); // A relay-reached peer's data-plane egress must be re-wrapped // through the server (dst = server); copy the bytes out first (the // DataPlane borrows `self.peers[idx]`) then wrap. A direct/punched @@ -1888,18 +1949,17 @@ impl PeerManager { let PeerState::Established(epochs) = &mut self.peers[idx].state else { unreachable!("just matched Established above"); }; - return epochs.current_mut().on_tun_packet(inner, now_ms); + let out = epochs.current_mut().on_tun_packet(inner, now_ms); + self.metrics.record_fec_tx(out.len() as u64); + return out; } let owned: Vec> = { let PeerState::Established(epochs) = &mut self.peers[idx].state else { unreachable!("just matched Established above"); }; - epochs - .current_mut() - .on_tun_packet(inner, now_ms) - .iter() - .map(|d| d.bytes.clone()) - .collect() + let out = epochs.current_mut().on_tun_packet(inner, now_ms); + self.metrics.record_fec_tx(out.len() as u64); + out.iter().map(|d| d.bytes.clone()).collect() }; self.egress.clear(); for b in owned { @@ -2287,10 +2347,8 @@ impl PeerManager { reqs.push(req); } if !reqs.is_empty() { - let dev = self.device_name.clone(); - let start = self.start_time; for req in reqs { - let resp = crate::control::dispatch_cmd(self, req.cmd, &dev, start); + let resp = self.handle_control_cmd(req.cmd); let _ = req.responder.send(resp); } } diff --git a/bin/yipd/src/peer_manager/obf.rs b/bin/yipd/src/peer_manager/obf.rs index 030f980..ad5e2aa 100644 --- a/bin/yipd/src/peer_manager/obf.rs +++ b/bin/yipd/src/peer_manager/obf.rs @@ -213,6 +213,7 @@ impl PeerManager { std::mem::swap(&mut d.bytes, &mut self.obf_scratch); self.daita_stats.authentic_bytes_tx += raw_wire as u64; self.daita_stats.padding_bytes_tx += pad as u64; + self.metrics.record_daita_padding(pad as u64); true } else { false @@ -266,9 +267,11 @@ impl PeerManager { if ptype == yip_obf::JUNK_TYPE { self.daita_stats.decoy_packets_tx += 1; self.daita_stats.decoy_bytes_tx += (raw_wire + pad) as u64; + self.metrics.record_daita_decoy(1, (raw_wire + pad) as u64); } else { self.daita_stats.authentic_bytes_tx += raw_wire as u64; self.daita_stats.padding_bytes_tx += pad as u64; + self.metrics.record_daita_padding(pad as u64); } true } else { diff --git a/bin/yipd/src/peer_manager/rekey.rs b/bin/yipd/src/peer_manager/rekey.rs index a36e5f0..df3fbfe 100644 --- a/bin/yipd/src/peer_manager/rekey.rs +++ b/bin/yipd/src/peer_manager/rekey.rs @@ -498,7 +498,14 @@ impl PeerManager { } else { pending .iter() - .flat_map(|inner| dp.on_tun_packet(inner, now_ms).to_vec()) + .flat_map(|inner| { + self.peers[idx].tx_bytes += inner.len() as u64; + self.peers[idx].tx_packets += 1; + self.metrics.record_tx(inner.len() as u64, 1); + let out = dp.on_tun_packet(inner, now_ms); + self.metrics.record_fec_tx(out.len() as u64); + out.to_vec() + }) .collect() }; for dg in primed { @@ -517,6 +524,7 @@ impl PeerManager { self.by_tag.remove(&old_tag); self.by_tag.insert(conn_tag, idx); self.record_new_epoch(idx); + self.metrics.record_rekey(); if self.egress.is_empty() { DispatchOut::None diff --git a/bin/yipd/src/peer_manager/relay.rs b/bin/yipd/src/peer_manager/relay.rs index bf899cd..9bd0e90 100644 --- a/bin/yipd/src/peer_manager/relay.rs +++ b/bin/yipd/src/peer_manager/relay.rs @@ -328,15 +328,17 @@ impl PeerManager { let pending = std::mem::take(&mut self.peers[idx].pending_tun); let mut owned: Vec> = Vec::new(); for inner in &pending { - owned.extend( - dp.on_tun_packet(inner, now_ms) - .iter() - .map(|d| d.bytes.clone()), - ); + self.peers[idx].tx_bytes += inner.len() as u64; + self.peers[idx].tx_packets += 1; + self.metrics.record_tx(inner.len() as u64, 1); + let out = dp.on_tun_packet(inner, now_ms); + self.metrics.record_fec_tx(out.len() as u64); + owned.extend(out.iter().map(|d| d.bytes.clone())); } self.peers[idx].state = PeerState::Established(Box::new(crate::epoch::EpochSet::new(dp, now_ms))); self.record_new_epoch(idx); + self.metrics.record_handshake_success(); for b in owned { if let Some(d) = self.relay_wrap(idx, b) { self.egress.push(d); @@ -405,15 +407,17 @@ impl PeerManager { let pending = std::mem::take(&mut self.peers[idx].pending_tun); let mut owned: Vec> = Vec::new(); for inner in &pending { - owned.extend( - dp.on_tun_packet(inner, now_ms) - .iter() - .map(|d| d.bytes.clone()), - ); + self.peers[idx].tx_bytes += inner.len() as u64; + self.peers[idx].tx_packets += 1; + self.metrics.record_tx(inner.len() as u64, 1); + let out = dp.on_tun_packet(inner, now_ms); + self.metrics.record_fec_tx(out.len() as u64); + owned.extend(out.iter().map(|d| d.bytes.clone())); } self.peers[idx].state = PeerState::Established(Box::new(crate::epoch::EpochSet::new(dp, now_ms))); self.record_new_epoch(idx); + self.metrics.record_handshake_success(); for b in owned { if let Some(d) = self.relay_wrap(idx, b) { self.egress.push(d); @@ -427,6 +431,7 @@ impl PeerManager { } Err(e) => { eprintln!("peer_manager: relayed read_response failed: {e}"); + self.metrics.record_handshake_failure(); DispatchOut::None } } diff --git a/bin/yipd/src/peer_manager/routing.rs b/bin/yipd/src/peer_manager/routing.rs index 0b45253..b4995a5 100644 --- a/bin/yipd/src/peer_manager/routing.rs +++ b/bin/yipd/src/peer_manager/routing.rs @@ -173,6 +173,8 @@ impl PeerManager { let PeerState::Established(epochs) = &mut self.peers[idx].state else { return DispatchOut::None; }; + let prev_rec = epochs.fec_recovered_count(); + let prev_burst = epochs.fec_unrecoverable_count(); // `EpochInbound::Send`/`TunThenSend` carry the full `EgressDatagram` // (real `dst` + `fate`), so no reconstruction from // `self.peers[idx].endpoint` is needed — that placeholder is wrong @@ -180,15 +182,27 @@ impl PeerManager { // `server_addr()` stand-in; `endpoint` may hold an unconfirmed // candidate or `None`). let opened = epochs.inbound_open(dg, now_ms); + let rec_diff = epochs.fec_recovered_count().saturating_sub(prev_rec); + if rec_diff > 0 { + self.metrics.record_fec_recovered(rec_diff); + } + let burst_diff = epochs.fec_unrecoverable_count().saturating_sub(prev_burst); + if burst_diff > 0 { + self.metrics.record_fec_burst_drop(burst_diff); + } if !matches!(opened, crate::epoch::EpochInbound::None) { // Authenticated + non-replayed (M2 roaming) — safe to roam. self.relearn_endpoint(idx, src); } match opened { - crate::epoch::EpochInbound::None => DispatchOut::None, + crate::epoch::EpochInbound::None => { + self.metrics.record_drop(1); + DispatchOut::None + } crate::epoch::EpochInbound::Tun(buf) => { self.peers[idx].rx_bytes += buf.len() as u64; self.peers[idx].rx_packets += 1; + self.metrics.record_rx(buf.len() as u64, 1); self.tun_scratch = buf; DispatchOut::Tun(&self.tun_scratch) } @@ -199,6 +213,7 @@ impl PeerManager { crate::epoch::EpochInbound::TunThenSend(buf, dgs) => { self.peers[idx].rx_bytes += buf.len() as u64; self.peers[idx].rx_packets += 1; + self.metrics.record_rx(buf.len() as u64, 1); self.tun_scratch = buf; self.egress = dgs; DispatchOut::Both(&self.tun_scratch, &self.egress) @@ -243,23 +258,36 @@ impl PeerManager { .map(|(i, _)| i) .collect(); for idx in candidates { - let hit = { + let (hit, rec_diff, burst_diff) = { let PeerState::Established(epochs) = &mut self.peers[idx].state else { continue; }; - match epochs.inbound_open(dg, now_ms) { + let prev_rec = epochs.fec_recovered_count(); + let prev_burst = epochs.fec_unrecoverable_count(); + let opened = epochs.inbound_open(dg, now_ms); + let rec_diff = epochs.fec_recovered_count().saturating_sub(prev_rec); + let burst_diff = epochs.fec_unrecoverable_count().saturating_sub(prev_burst); + let h = match opened { crate::epoch::EpochInbound::None => None, crate::epoch::EpochInbound::Tun(buf) => Some((Some(buf), Vec::new())), crate::epoch::EpochInbound::Send(dgs) => Some((None, dgs)), crate::epoch::EpochInbound::TunThenSend(buf, dgs) => Some((Some(buf), dgs)), - } + }; + (h, rec_diff, burst_diff) }; + if rec_diff > 0 { + self.metrics.record_fec_recovered(rec_diff); + } + if burst_diff > 0 { + self.metrics.record_fec_burst_drop(burst_diff); + } let Some((tun, udp)) = hit else { continue; }; if let Some(ref t) = tun { self.peers[idx].rx_bytes += t.len() as u64; self.peers[idx].rx_packets += 1; + self.metrics.record_rx(t.len() as u64, 1); } // Authenticated + non-replayed (M2 roaming) — safe to roam. self.relearn_endpoint(idx, src); @@ -283,6 +311,7 @@ impl PeerManager { (None, true) => DispatchOut::None, }; } + self.metrics.record_drop(1); DispatchOut::None } } diff --git a/bin/yipd/src/peer_manager/tests.rs b/bin/yipd/src/peer_manager/tests.rs index 015d6db..38dee9b 100644 --- a/bin/yipd/src/peer_manager/tests.rs +++ b/bin/yipd/src/peer_manager/tests.rs @@ -2967,3 +2967,86 @@ fn trace_path_and_ping_peer_telemetry() { assert!(bad_hops.is_empty()); assert_eq!(bad_rtt, 0.0); } + +#[test] +fn test_peer_manager_metrics_increment_on_traffic() { + let manager = make_test_peer_manager(); + let initial_tx = manager.metrics().snapshot(&[]).traffic.tx_bytes; + // Dispatch dummy packet + manager.record_test_tx_packet(128); + let after_tx = manager.metrics().snapshot(&[]).traffic.tx_bytes; + assert_eq!(after_tx, initial_tx + 128); +} + +#[test] +fn test_peer_manager_metrics_injection_and_control_cmd() { + use crate::control::{ControlCmd, ControlResponse}; + use crate::metrics::MetricsRegistry; + use std::sync::Arc; + + let custom_registry = Arc::new(MetricsRegistry::new()); + custom_registry.record_tx(256, 2); + custom_registry.record_rx(512, 4); + custom_registry.record_fec_tx(10); + custom_registry.record_fec_recovered(5); + custom_registry.record_fec_burst_drop(1); + custom_registry.record_daita_padding(64); + custom_registry.record_daita_decoy(2, 200); + custom_registry.record_handshake_success(); + custom_registry.record_handshake_failure(); + custom_registry.record_rekey(); + + let mut manager = make_test_peer_manager().with_metrics(Arc::clone(&custom_registry)); + assert_eq!(manager.metrics().snapshot(&[]).traffic.tx_bytes, 256); + assert_eq!(manager.metrics().snapshot(&[]).traffic.rx_bytes, 512); + + let replaced_registry = Arc::new(MetricsRegistry::new()); + replaced_registry.record_tx(1024, 8); + manager.set_metrics(replaced_registry); + assert_eq!(manager.metrics().snapshot(&[]).traffic.tx_bytes, 1024); + + // Test handle_control_cmd arm for ControlCmd::GetMetrics + let resp = manager.handle_control_cmd(ControlCmd::GetMetrics { peer: None }); + match resp { + ControlResponse::Metrics { + ok, + metrics, + prometheus, + } => { + assert!(ok); + assert_eq!(metrics.traffic.tx_bytes, 1024); + assert_eq!(metrics.traffic.tx_packets, 8); + assert!(prometheus.contains("yip_traffic_tx_bytes_total 1024")); + } + other => panic!("expected ControlResponse::Metrics, got: {other:?}"), + } +} + +#[test] +fn test_peer_manager_metrics_drops_and_split_tunnel() { + use std::net::SocketAddr; + + let mut manager = make_test_peer_manager(); + let initial_drops = manager.metrics().snapshot(&[]).traffic.dropped_packets; + + // Ingress empty UDP datagram should record drop + let fake_addr: SocketAddr = "192.0.2.1:12345".parse().unwrap(); + let _ = manager.on_udp(fake_addr, &[], 0); + let drops_after_empty = manager.metrics().snapshot(&[]).traffic.dropped_packets; + assert_eq!(drops_after_empty, initial_drops + 1); + + // Unroutable TUN packet (no peer matches and membership is None) + // Create an IPv6 packet pointing to an unconfigured address + let mut bad_ip_pkt = vec![0x60, 0, 0, 0, 0, 0, 0, 64]; + bad_ip_pkt.extend_from_slice(&[0xfd, 0x00, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]); // src + bad_ip_pkt.extend_from_slice(&[0xfd, 0x00, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2]); // dst + bad_ip_pkt.extend_from_slice(&[0u8; 32]); + + // Give manager 2 peers so sole-peer fallback does not engage + let p2 = peer_cfg(2, "10.0.0.2:2000"); + manager.add_peer(p2).unwrap(); + + let _ = manager.on_tun(&bad_ip_pkt, 0); + let drops_after_unroutable = manager.metrics().snapshot(&[]).traffic.dropped_packets; + assert_eq!(drops_after_unroutable, drops_after_empty + 1); +} diff --git a/bin/yipd/src/peer_manager/testutil.rs b/bin/yipd/src/peer_manager/testutil.rs index 93eed37..066c780 100644 --- a/bin/yipd/src/peer_manager/testutil.rs +++ b/bin/yipd/src/peer_manager/testutil.rs @@ -33,6 +33,10 @@ pub fn test_peer_manager() -> PeerManager { ) } +pub fn make_test_peer_manager() -> PeerManager { + test_peer_manager() +} + /// Build a real `DataPlane` (via an in-process Noise handshake) with a /// specific `conn_tag`, standing in for "a peer that has already /// completed its handshake" — the "test seam" for demux tests: rather From 382ead0d359fac092ab5061670e3cfbc1247f8e6 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Thu, 29 Oct 2026 20:32:20 -0400 Subject: [PATCH 5/9] fix(dataplane): instrument relayed data-plane metrics and avoid heap allocation in control dispatch --- bin/yipd/src/peer_manager/mod.rs | 6 ++- bin/yipd/src/peer_manager/relay.rs | 31 +++++++++++-- bin/yipd/src/peer_manager/tests.rs | 70 ++++++++++++++++++++++++++++++ 3 files changed, 103 insertions(+), 4 deletions(-) diff --git a/bin/yipd/src/peer_manager/mod.rs b/bin/yipd/src/peer_manager/mod.rs index 3908581..bfe6bd1 100644 --- a/bin/yipd/src/peer_manager/mod.rs +++ b/bin/yipd/src/peer_manager/mod.rs @@ -1261,7 +1261,11 @@ impl PeerManager { &mut self, cmd: crate::control::ControlCmd, ) -> crate::control::ControlResponse { - crate::control::dispatch_cmd(self, cmd, &self.device_name.clone(), self.start_time) + let device_name = std::mem::take(&mut self.device_name); + let start_time = self.start_time; + let resp = crate::control::dispatch_cmd(self, cmd, &device_name, start_time); + self.device_name = device_name; + resp } /// Configure online Threshold CA coordinator address for automated 70% TTL certificate rotation. diff --git a/bin/yipd/src/peer_manager/relay.rs b/bin/yipd/src/peer_manager/relay.rs index 9bd0e90..36c5a5e 100644 --- a/bin/yipd/src/peer_manager/relay.rs +++ b/bin/yipd/src/peer_manager/relay.rs @@ -445,17 +445,42 @@ impl PeerManager { /// real `dst`/`fate` on each `EgressDatagram` are irrelevant for a /// relayed peer (the actual wire destination is the relay server). fn relayed_data(&mut self, idx: usize, dg: &[u8], now_ms: u64) -> DispatchOut<'_> { + if !dg.is_empty() && dg[0] == PacketType::Data as u8 { + self.peers[idx].last_activity_ms = now_ms; + } let (tun, udp): (Option>, Vec>) = { let PeerState::Established(epochs) = &mut self.peers[idx].state else { return DispatchOut::None; }; - match epochs.inbound_open(dg, now_ms) { - crate::epoch::EpochInbound::None => (None, Vec::new()), - crate::epoch::EpochInbound::Tun(buf) => (Some(buf), Vec::new()), + let prev_rec = epochs.fec_recovered_count(); + let prev_burst = epochs.fec_unrecoverable_count(); + let opened = epochs.inbound_open(dg, now_ms); + let rec_diff = epochs.fec_recovered_count().saturating_sub(prev_rec); + if rec_diff > 0 { + self.metrics.record_fec_recovered(rec_diff); + } + let burst_diff = epochs.fec_unrecoverable_count().saturating_sub(prev_burst); + if burst_diff > 0 { + self.metrics.record_fec_burst_drop(burst_diff); + } + match opened { + crate::epoch::EpochInbound::None => { + self.metrics.record_drop(1); + (None, Vec::new()) + } + crate::epoch::EpochInbound::Tun(buf) => { + self.peers[idx].rx_bytes += buf.len() as u64; + self.peers[idx].rx_packets += 1; + self.metrics.record_rx(buf.len() as u64, 1); + (Some(buf), Vec::new()) + } crate::epoch::EpochInbound::Send(dgs) => { (None, dgs.iter().map(|d| d.bytes.clone()).collect()) } crate::epoch::EpochInbound::TunThenSend(buf, dgs) => { + self.peers[idx].rx_bytes += buf.len() as u64; + self.peers[idx].rx_packets += 1; + self.metrics.record_rx(buf.len() as u64, 1); (Some(buf), dgs.iter().map(|d| d.bytes.clone()).collect()) } } diff --git a/bin/yipd/src/peer_manager/tests.rs b/bin/yipd/src/peer_manager/tests.rs index 38dee9b..46b5011 100644 --- a/bin/yipd/src/peer_manager/tests.rs +++ b/bin/yipd/src/peer_manager/tests.rs @@ -3050,3 +3050,73 @@ fn test_peer_manager_metrics_drops_and_split_tunnel() { let drops_after_unroutable = manager.metrics().snapshot(&[]).traffic.dropped_packets; assert_eq!(drops_after_unroutable, drops_after_empty + 1); } + +#[test] +fn test_peer_manager_metrics_relayed_traffic() { + let kp_a = generate_keypair(); + let kp_b = generate_keypair(); + let ep_a: SocketAddr = "10.0.0.1:1000".parse().unwrap(); + let ep_b: SocketAddr = "10.0.0.2:2000".parse().unwrap(); + let cfg_b = PeerConfig { + public_key: kp_b.public, + endpoint: Some(ep_b), + }; + let cfg_a = PeerConfig { + public_key: kp_a.public, + endpoint: Some(ep_a), + }; + + let (mut pm_a, _sent_a) = pm_with_mock_rdv(&kp_a, &[cfg_b]); + let (mut pm_b, _sent_b) = pm_with_mock_rdv(&kp_b, &[cfg_a]); + + peer_test_key_registry() + .lock() + .unwrap() + .insert(kp_b.public, (kp_b.private, [0xBB; 32])); + peer_test_key_registry() + .lock() + .unwrap() + .insert(kp_a.public, (kp_a.private, [0xAA; 32])); + + // Perform handshake to establish session between pm_a and pm_b + let init = pm_a.on_tun(&dummy_tun_pkt(), 0)[0].bytes.clone(); + let resp = resp_bytes(&pm_b.on_udp(ep_a, &init, 0)); + assert_eq!(resp.len(), 1, "cold-start init must produce one resp"); + pm_a.on_udp(ep_b, &resp[0], 0); + assert!(established_tag(&pm_a, 0).is_some()); + assert_eq!(established_tag(&pm_a, 0), established_tag(&pm_b, 0)); + + let snap_before = pm_b.metrics().snapshot(&[]); + let rx_pkts_before = snap_before.traffic.rx_packets; + let rx_bytes_before = snap_before.traffic.rx_bytes; + let drops_before = snap_before.traffic.dropped_packets; + + // Initiator pm_a transmits a TUN data packet + let tun_pkt = dummy_tun_pkt(); + let data = pm_a.on_tun(&tun_pkt, 100).to_vec(); + assert_eq!(data[0].bytes[0], PacketType::Data as u8); + + // Deliver data packet via rendezvous relay server to pm_b + let relayed = wrap_relay_deliver(&pm_b, &data[0].bytes); + let out = pm_b.on_udp(mock_server(), &relayed, 150); + + // Verify TUN delivery and metrics incrementation + assert!(matches!(out, DispatchOut::Tun(_))); + let snap_after = pm_b.metrics().snapshot(&[]); + assert_eq!(snap_after.traffic.rx_packets, rx_pkts_before + 1); + assert_eq!( + snap_after.traffic.rx_bytes, + rx_bytes_before + tun_pkt.len() as u64 + ); + assert_eq!(pm_b.peers[0].rx_packets, 1); + assert_eq!(pm_b.peers[0].rx_bytes, tun_pkt.len() as u64); + + // Verify unauthenticated / corrupted relayed packet increments drop metrics + let bad_data = vec![PacketType::Data as u8, 0xDE, 0xAD, 0xBE, 0xEF]; + let bad_relayed = wrap_relay_deliver(&pm_b, &bad_data); + let bad_out = pm_b.on_udp(mock_server(), &bad_relayed, 200); + assert!(matches!(bad_out, DispatchOut::None)); + + let snap_dropped = pm_b.metrics().snapshot(&[]); + assert_eq!(snap_dropped.traffic.dropped_packets, drops_before + 1); +} From 6d5e4167fb9d0e33224f640cf419aa4474deee30 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Thu, 29 Oct 2026 20:47:49 -0400 Subject: [PATCH 6/9] feat(daemon): add metrics_listen configuration and yip metrics CLI command --- bin/yipd/src/config.rs | 49 +++++++++++++- bin/yipd/src/main.rs | 149 ++++++++++++++++++++++++++++++++--------- bin/yipd/src/tunnel.rs | 8 +++ 3 files changed, 172 insertions(+), 34 deletions(-) diff --git a/bin/yipd/src/config.rs b/bin/yipd/src/config.rs index 77bc61f..e25073e 100644 --- a/bin/yipd/src/config.rs +++ b/bin/yipd/src/config.rs @@ -196,6 +196,8 @@ pub struct Config { pub control_socket: Option, /// Configured TUN device MTU (`mtu=<576..=65535>`). pub mtu: Option, + /// Optional Prometheus/OpenTelemetry HTTP metrics scraping endpoint (`metrics_listen=`). + pub metrics_listen: Option, } // ── hex decode helper ───────────────────────────────────────────────────────── @@ -547,6 +549,7 @@ impl Config { let mut shards: Option = None; let mut control_socket: Option = None; let mut mtu: Option = None; + let mut metrics_listen: Option = None; for line in text.lines() { let line = line.trim(); @@ -612,7 +615,7 @@ impl Config { )); } } - "device" => device = Some(val.to_owned()), + "device" => device = Some(val.trim_matches('"').to_owned()), "device_kind" => device_kind = TunnelMode::parse_device_kind(val)?, "rendezvous" => { rendezvous = Some(if let Some(rest) = val.strip_prefix("reality://") { @@ -755,6 +758,13 @@ impl Config { } mtu = Some(m); } + "metrics_listen" | "metrics" => { + let clean = val.trim_matches('"'); + let addr = clean + .parse::() + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?; + metrics_listen = Some(addr); + } // Silently ignore unknown keys for forward-compatibility. The // netns config files still contain `initiate=true|false` from // before Task 5 removed the field; this is intentional so @@ -986,6 +996,7 @@ impl Config { shards: shards.unwrap_or(1), control_socket, mtu, + metrics_listen, }) } } @@ -2499,4 +2510,40 @@ local_public=0000000000000000000000000000000000000000000000000000000000000002 let text_bucket_large = format!("{base}defense_buckets=512,1600\n"); assert!(Config::parse(&text_bucket_large).is_err()); } + + #[test] + fn test_config_metrics_listen_parsing() { + let conf = + "local_private=00000000000000000000000000000000000000000000000000000000000000ff\n\ + local_public=00000000000000000000000000000000000000000000000000000000000000aa\n\ + peer_public=00000000000000000000000000000000000000000000000000000000000000bb\n\ + peer_endpoint=10.0.0.2:51820\n\ + device = \"tun0\"\nmetrics_listen = \"127.0.0.1:9184\"\n"; + let parsed = Config::parse(conf).unwrap(); + assert_eq!( + parsed.metrics_listen, + Some("127.0.0.1:9184".parse().unwrap()) + ); + } + + #[test] + fn test_config_metrics_listen_absent_and_invalid() { + let base = + "local_private=00000000000000000000000000000000000000000000000000000000000000ff\n\ + local_public=00000000000000000000000000000000000000000000000000000000000000aa\n\ + peer_public=00000000000000000000000000000000000000000000000000000000000000bb\n\ + peer_endpoint=10.0.0.2:51820\n\ + device=tun0\n"; + let parsed_none = Config::parse(base).unwrap(); + assert_eq!(parsed_none.metrics_listen, None); + + let parsed_alias = Config::parse(&format!("{base}metrics=0.0.0.0:9090\n")).unwrap(); + assert_eq!( + parsed_alias.metrics_listen, + Some("0.0.0.0:9090".parse().unwrap()) + ); + + let bad_conf = format!("{base}metrics_listen=invalid_addr\n"); + assert!(Config::parse(&bad_conf).is_err()); + } } diff --git a/bin/yipd/src/main.rs b/bin/yipd/src/main.rs index 025c03d..52e0817 100644 --- a/bin/yipd/src/main.rs +++ b/bin/yipd/src/main.rs @@ -50,11 +50,13 @@ fn print_daemon_usage() { println!("Commands:"); println!(" status Display daemon status and summary"); println!(" peers List connected peers and statistics"); + println!(" metrics Display runtime metrics and telemetry"); println!(" top Real-time ANSI telemetry dashboard"); println!(); println!("Options:"); println!(" -c, --control UNIX domain control socket path"); println!(" --mtu TUN device MTU (576-65535, default: 1400)"); + println!(" --metrics-listen Prometheus/OTEL HTTP metrics listen address (e.g. 127.0.0.1:9184)"); println!(" --genkey Generate a fresh X25519 keypair"); println!(" --addr Derive self-certifying mesh IPv6 address"); println!(" -V, --version Print version"); @@ -83,6 +85,16 @@ fn print_cli_usage(subcmd: &str) { println!(" --json Output raw JSON"); println!(" -h, --help Print help"); } + "metrics" => { + println!("Usage: yipctl metrics [OPTIONS]"); + println!(); + println!("Display system, transport, crypto, and peer metrics."); + println!(); + println!("Options:"); + println!(" -c, --control Path to yipd UNIX control socket"); + println!(" --json Output raw JSON"); + println!(" -h, --help Print help"); + } "top" => { println!("Usage: yipctl top [OPTIONS]"); println!(); @@ -219,6 +231,7 @@ fn print_cli_usage(subcmd: &str) { println!("Commands:"); println!(" status Display daemon status and summary"); println!(" peers List connected peers and statistics"); + println!(" metrics Display runtime metrics and telemetry"); println!(" routes Display live mesh overlay routing table"); println!(" topology Display mesh topology graph projection"); println!(" split Inspect or configure split-tunnel bypass rules"); @@ -726,6 +739,7 @@ fn run_client_subcommand( } } } + "metrics" => control::run_cli_metrics(&socket, json).map_err(std::io::Error::other), _ => Err(std::io::Error::new( std::io::ErrorKind::InvalidInput, format!("unknown subcommand: {subcmd}"), @@ -733,6 +747,69 @@ fn run_client_subcommand( } } +pub(crate) fn parse_daemon_flags( + args: &mut impl Iterator, +) -> std::io::Result<( + Option, + Option, + Option, +)> { + let mut control_override = None; + let mut mtu_override = None; + let mut metrics_listen_override = None; + while let Some(arg) = args.next() { + if arg == "--control" || arg == "-c" { + let cpath = args.next().ok_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "--control requires a socket path argument", + ) + })?; + control_override = Some(std::path::PathBuf::from(cpath)); + } else if arg == "--mtu" { + let mstr = args.next().ok_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "--mtu requires a byte count argument", + ) + })?; + let m: u32 = mstr.parse().map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("invalid mtu '{mstr}': must be an integer"), + ) + })?; + if !(576..=65535).contains(&m) { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("invalid mtu '{m}': must be between 576 and 65535"), + )); + } + mtu_override = Some(m); + } else if arg == "--metrics-listen" { + let mstr = args.next().ok_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "--metrics-listen requires a socket address argument (e.g. 127.0.0.1:9184)", + ) + })?; + let addr: std::net::SocketAddr = mstr.parse().map_err(|e| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("invalid metrics listen address '{mstr}': {e}"), + ) + })?; + metrics_listen_override = Some(addr); + } else { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("unrecognized option: {arg}"), + )); + } + } + Ok((control_override, mtu_override, metrics_listen_override)) +} + fn main() -> std::io::Result<()> { let mut args = std::env::args(); let prog_name = args.next(); @@ -765,6 +842,7 @@ fn main() -> std::io::Result<()> { Some("routes") => return run_client_subcommand("routes", args), Some("topology") => return run_client_subcommand("topology", args), Some("split") | Some("split-tunnel") => return run_client_subcommand("split", args), + Some("metrics") => return run_client_subcommand("metrics", args), Some("genkey") | Some("--genkey") => { let kp = yip_crypto::generate_keypair(); println!("private={}", hex_encode(&kp.private)); @@ -807,6 +885,7 @@ fn main() -> std::io::Result<()> { Some("routes") => run_client_subcommand("routes", args), Some("topology") => run_client_subcommand("topology", args), Some("split") | Some("split-tunnel") => run_client_subcommand("split", args), + Some("metrics") => run_client_subcommand("metrics", args), Some("--version") | Some("-V") => { println!("{}", banner()); Ok(()) @@ -841,39 +920,8 @@ fn main() -> std::io::Result<()> { )) } Some(path) => { - let mut control_override = None; - let mut mtu_override = None; - while let Some(arg) = args.next() { - if arg == "--control" || arg == "-c" { - let cpath = args.next().ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "--control requires a socket path argument", - ) - })?; - control_override = Some(std::path::PathBuf::from(cpath)); - } else if arg == "--mtu" { - let mstr = args.next().ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "--mtu requires a byte count argument", - ) - })?; - let m: u32 = mstr.parse().map_err(|_| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - format!("invalid mtu '{mstr}': must be an integer"), - ) - })?; - if !(576..=65535).contains(&m) { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidInput, - format!("invalid mtu '{m}': must be between 576 and 65535"), - )); - } - mtu_override = Some(m); - } - } + let (control_override, mtu_override, metrics_listen_override) = + parse_daemon_flags(&mut args)?; let text = std::fs::read_to_string(path)?; let mut config = Config::parse(&text)?; @@ -883,6 +931,9 @@ fn main() -> std::io::Result<()> { if let Some(mtu) = mtu_override { config.mtu = Some(mtu); } + if let Some(maddr) = metrics_listen_override { + config.metrics_listen = Some(maddr); + } tunnel::run(config) } None => { @@ -929,4 +980,36 @@ mod tests { let bad = "zz".repeat(32); assert!(hex_decode_32(&bad).is_err()); } + + #[test] + fn test_parse_daemon_flags_metrics_listen() { + let args = vec![ + "--metrics-listen".to_string(), + "127.0.0.1:9184".to_string(), + "--control".to_string(), + "/run/yipd.sock".to_string(), + "--mtu".to_string(), + "1420".to_string(), + ]; + let (ctl, mtu, metrics) = parse_daemon_flags(&mut args.into_iter()).unwrap(); + assert_eq!(metrics, Some("127.0.0.1:9184".parse().unwrap())); + assert_eq!(ctl, Some(std::path::PathBuf::from("/run/yipd.sock"))); + assert_eq!(mtu, Some(1420)); + } + + #[test] + fn test_parse_daemon_flags_invalid_metrics_listen() { + let missing_arg = vec!["--metrics-listen".to_string()]; + assert!(parse_daemon_flags(&mut missing_arg.into_iter()).is_err()); + + let invalid_addr = vec!["--metrics-listen".to_string(), "not-an-ip-addr".to_string()]; + assert!(parse_daemon_flags(&mut invalid_addr.into_iter()).is_err()); + } + + #[test] + fn test_print_cli_usage_metrics() { + print_cli_usage("metrics"); + print_cli_usage(""); + print_daemon_usage(); + } } diff --git a/bin/yipd/src/tunnel.rs b/bin/yipd/src/tunnel.rs index a1a49e0..0be5138 100644 --- a/bin/yipd/src/tunnel.rs +++ b/bin/yipd/src/tunnel.rs @@ -209,6 +209,14 @@ pub fn run(config: Config) -> io::Result<()> { config.device.clone(), std::time::Instant::now(), ); + let _metrics_server = match config.metrics_listen { + Some(addr) => Some(crate::metrics::MetricsHttpServer::start( + addr, + manager.metrics(), + Box::new(Vec::new), + )?), + None => None, + }; let local_addr = manager.local_addr(); // ── create the tunnel device (TUN or TAP) ──────────────────────────────── From e7245963f6f082bc3ebaa7d55de0f7e931040971 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Thu, 29 Oct 2026 21:00:47 -0400 Subject: [PATCH 7/9] test(metrics): add end-to-end integration test for HTTP scraper and IPC metrics --- bin/yipd/tests/metrics_endpoint_test.rs | 634 ++++++++++++++++++++++++ tests/metrics_endpoint_test.rs | 1 + 2 files changed, 635 insertions(+) create mode 100644 bin/yipd/tests/metrics_endpoint_test.rs create mode 120000 tests/metrics_endpoint_test.rs diff --git a/bin/yipd/tests/metrics_endpoint_test.rs b/bin/yipd/tests/metrics_endpoint_test.rs new file mode 100644 index 0000000..4f52ad4 --- /dev/null +++ b/bin/yipd/tests/metrics_endpoint_test.rs @@ -0,0 +1,634 @@ +//! End-to-end integration tests for Prometheus and OpenTelemetry telemetry +//! exposition over the UNIX control socket and embedded HTTP scraper. + +use std::io::{BufRead, BufReader, Read, Write}; +use std::net::{SocketAddr, TcpStream}; +use std::os::unix::net::UnixStream; +use std::path::Path; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use yip_io::poll::Dispatch; +use yipd::config::PeerConfig; +use yipd::control::{client_query, run_cli_metrics, ControlCmd, ControlHandle, ControlResponse}; +use yipd::metrics::MetricsHttpServer; +use yipd::mode::TunnelMode; +use yipd::peer_manager::{PeerManager, PeerStats}; + +fn make_test_peer_manager() -> PeerManager { + let kp = yip_crypto::generate_keypair(); + let peer = PeerConfig { + public_key: [1u8; 32], + endpoint: Some("10.0.0.1:1000".parse().unwrap()), + }; + PeerManager::new( + kp.private, + kp.public, + &[peer], + TunnelMode::L3Tun, + None, + None, + false, + ) +} + +struct Harness { + socket_path: String, + pm: PeerManager, + cached_stats: Arc>>, + now_ms: u64, +} + +impl Harness { + fn new(test_name: &str) -> Self { + let socket_path = format!( + "/tmp/yip_metrics_test_{}_{}_{}.sock", + test_name, + std::process::id(), + Instant::now().elapsed().as_nanos() + ); + let _ = std::fs::remove_file(&socket_path); + + let mut pm = make_test_peer_manager(); + let handle = ControlHandle::bind(&socket_path) + .unwrap_or_else(|e| panic!("failed to bind control socket {socket_path}: {e}")); + pm.set_control(Some(handle), "yip-test-dev".into(), Instant::now()); + + let cached_stats = Arc::new(Mutex::new(pm.peer_stats())); + + Self { + socket_path, + pm, + cached_stats, + now_ms: 1_000, + } + } + + fn socket_path(&self) -> &str { + &self.socket_path + } + + fn refresh_cached_stats(&mut self) { + let stats = self.pm.peer_stats(); + let mut lock = self.cached_stats.lock().unwrap(); + *lock = stats; + } + + fn drive_ticks(&mut self, iterations: usize) { + for _ in 0..iterations { + let _ = self.pm.tick(self.now_ms); + self.now_ms = self.now_ms.saturating_add(5); + std::thread::sleep(Duration::from_millis(2)); + } + self.refresh_cached_stats(); + } + + fn query_cmd(&mut self, cmd: ControlCmd) -> Result { + let path = self.socket_path.clone(); + let (tx, rx) = std::sync::mpsc::channel(); + let client_thread = std::thread::spawn(move || { + let res = client_query(Path::new(&path), cmd); + let _ = tx.send(res); + }); + + let mut result = None; + let start = Instant::now(); + while start.elapsed() < Duration::from_secs(5) { + match rx.try_recv() { + Ok(res) => { + result = Some(res); + break; + } + Err(std::sync::mpsc::TryRecvError::Empty) => { + let _ = self.pm.tick(self.now_ms); + self.now_ms = self.now_ms.saturating_add(5); + std::thread::sleep(Duration::from_millis(2)); + } + Err(std::sync::mpsc::TryRecvError::Disconnected) => break, + } + } + + client_thread.join().expect("client thread join failed"); + self.refresh_cached_stats(); + result.ok_or_else(|| "timed out waiting for response".to_string())? + } + + fn query_cli_metrics(&mut self, json: bool) -> Result<(), String> { + let path = self.socket_path.clone(); + let (tx, rx) = std::sync::mpsc::channel(); + let client_thread = std::thread::spawn(move || { + let res = run_cli_metrics(Path::new(&path), json); + let _ = tx.send(res); + }); + + let mut result = None; + let start = Instant::now(); + while start.elapsed() < Duration::from_secs(5) { + match rx.try_recv() { + Ok(res) => { + result = Some(res); + break; + } + Err(std::sync::mpsc::TryRecvError::Empty) => { + let _ = self.pm.tick(self.now_ms); + self.now_ms = self.now_ms.saturating_add(5); + std::thread::sleep(Duration::from_millis(2)); + } + Err(std::sync::mpsc::TryRecvError::Disconnected) => break, + } + } + + client_thread.join().expect("client thread join failed"); + self.refresh_cached_stats(); + result.ok_or_else(|| "timed out waiting for response".to_string())? + } +} + +impl Drop for Harness { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.socket_path); + } +} + +fn http_request( + addr: SocketAddr, + method: &str, + path: &str, +) -> (String, Vec<(String, String)>, String) { + let mut stream = TcpStream::connect(addr) + .unwrap_or_else(|e| panic!("failed to connect to HTTP scraper at {addr}: {e}")); + stream + .set_read_timeout(Some(Duration::from_secs(3))) + .expect("set read timeout"); + stream + .set_write_timeout(Some(Duration::from_secs(3))) + .expect("set write timeout"); + + let req = format!("{method} {path} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"); + stream.write_all(req.as_bytes()).expect("write request"); + stream.flush().expect("flush request"); + + let mut response_str = String::new(); + stream + .read_to_string(&mut response_str) + .expect("read response"); + + let mut parts = response_str.splitn(2, "\r\n\r\n"); + let header_part = parts.next().unwrap_or(""); + let body_part = parts.next().unwrap_or("").to_string(); + + let mut lines = header_part.lines(); + let status_line = lines.next().unwrap_or("").to_string(); + + let mut headers = Vec::new(); + for line in lines { + if let Some((k, v)) = line.split_once(':') { + headers.push((k.trim().to_ascii_lowercase(), v.trim().to_string())); + } + } + + (status_line, headers, body_part) +} + +#[test] +fn test_live_unix_control_socket_get_metrics() { + let mut harness = Harness::new("ctrl_metrics"); + let socket_path = harness.socket_path().to_string(); + + // 1. Raw UNIX domain socket JSON query + let stream = UnixStream::connect(&socket_path).expect("connect to control socket"); + stream + .set_read_timeout(Some(Duration::from_secs(3))) + .expect("set read timeout"); + stream + .set_write_timeout(Some(Duration::from_secs(3))) + .expect("set write timeout"); + + let mut writer = stream.try_clone().expect("clone stream"); + let mut reader = BufReader::new(stream); + + writer + .write_all(b"{\"cmd\":\"get_metrics\"}\n") + .expect("write cmd"); + writer.flush().expect("flush cmd"); + + harness.drive_ticks(50); + + let mut raw_resp_line = String::new(); + reader + .read_line(&mut raw_resp_line) + .expect("read response line"); + assert!( + raw_resp_line.contains("\"ok\":true"), + "raw response missing ok:true: {raw_resp_line}" + ); + assert!( + raw_resp_line.contains("\"metrics\":"), + "raw response missing metrics object: {raw_resp_line}" + ); + assert!( + raw_resp_line.contains("\"prometheus\":"), + "raw response missing prometheus string: {raw_resp_line}" + ); + + let parsed: ControlResponse = + serde_json::from_str(raw_resp_line.trim()).expect("deserialize raw control response"); + match parsed { + ControlResponse::Metrics { + ok, + metrics, + prometheus, + } => { + assert!(ok); + assert_eq!(metrics.traffic.tx_bytes, 0); + assert_eq!(metrics.traffic.rx_bytes, 0); + assert_eq!(metrics.peers.total, 1); + assert!( + prometheus.contains("# HELP yip_traffic_tx_bytes_total Total bytes transmitted") + ); + assert!(prometheus.contains("# TYPE yip_traffic_tx_bytes_total counter")); + assert!(prometheus.contains("yip_traffic_tx_bytes_total 0")); + assert!(prometheus.contains("yip_peers_total 1")); + } + other => panic!("expected ControlResponse::Metrics variant, got: {other:?}"), + } + + // 2. Client query dispatch + let resp = harness + .query_cmd(ControlCmd::GetMetrics { peer: None }) + .expect("client_query GetMetrics failed"); + match resp { + ControlResponse::Metrics { + ok, + metrics, + prometheus, + } => { + assert!(ok); + assert_eq!(metrics.peers.total, 1); + assert!(prometheus.contains("yip_peers_total 1")); + } + other => panic!("expected ControlResponse::Metrics variant, got: {other:?}"), + } + + // 3. CLI runner query dispatch + assert!( + harness.query_cli_metrics(false).is_ok(), + "run_cli_metrics raw text failed" + ); + assert!( + harness.query_cli_metrics(true).is_ok(), + "run_cli_metrics JSON failed" + ); + + // 4. Targeted peer query + let resp_targeted = harness + .query_cmd(ControlCmd::GetMetrics { + peer: Some("10.0.0.1".into()), + }) + .expect("targeted peer query failed"); + match resp_targeted { + ControlResponse::Metrics { ok, metrics, .. } => { + assert!(ok); + assert!(metrics.peers.list.len() <= 1); + } + other => panic!("expected ControlResponse::Metrics variant, got: {other:?}"), + } +} + +#[test] +fn test_live_ephemeral_http_scraper_endpoints() { + let harness = Harness::new("http_scraper"); + let registry = harness.pm.metrics(); + let cached_stats = Arc::clone(&harness.cached_stats); + + let server = MetricsHttpServer::start( + "127.0.0.1:0".parse().unwrap(), + registry, + Box::new(move || cached_stats.lock().unwrap().clone()), + ) + .expect("start MetricsHttpServer"); + + let addr = server.local_addr(); + assert_ne!(addr.port(), 0); + + // 1. GET /healthz + let (status, headers, body) = http_request(addr, "GET", "/healthz"); + assert_eq!(status, "HTTP/1.1 200 OK"); + assert_eq!(body, "ok"); + let content_type = headers + .iter() + .find(|(k, _)| k == "content-type") + .map(|(_, v)| v.as_str()); + assert_eq!(content_type, Some("text/plain; charset=utf-8")); + let content_length = headers + .iter() + .find(|(k, _)| k == "content-length") + .map(|(_, v)| v.as_str()); + assert_eq!(content_length, Some("2")); + + // 2. HEAD /healthz + let (head_status, head_headers, head_body) = http_request(addr, "HEAD", "/healthz"); + assert_eq!(head_status, "HTTP/1.1 200 OK"); + assert_eq!(head_body, ""); + let head_content_len = head_headers + .iter() + .find(|(k, _)| k == "content-length") + .map(|(_, v)| v.as_str()); + assert_eq!(head_content_len, Some("2")); + + // 3. GET /metrics + let (m_status, m_headers, m_body) = http_request(addr, "GET", "/metrics"); + assert_eq!(m_status, "HTTP/1.1 200 OK"); + let m_content_type = m_headers + .iter() + .find(|(k, _)| k == "content-type") + .map(|(_, v)| v.as_str()) + .expect("missing content-type header"); + assert!( + m_content_type.contains("text/plain; version=0.0.4"), + "expected text/plain; version=0.0.4 in content-type, got: {m_content_type}" + ); + + // Verify Prometheus format compliance + assert!(m_body.contains("# HELP yip_uptime_seconds Process runtime in seconds")); + assert!(m_body.contains("# TYPE yip_uptime_seconds counter")); + assert!(m_body.contains("# HELP yip_peers_active Count of actively handshaked mesh peers")); + assert!(m_body.contains("# TYPE yip_peers_active gauge")); + assert!(m_body.contains("# HELP yip_traffic_tx_bytes_total Total bytes transmitted")); + assert!(m_body.contains("# TYPE yip_traffic_tx_bytes_total counter")); + assert!(m_body + .contains("# HELP yip_traffic_rx_bytes_total Total payload and control bytes received")); + assert!(m_body.contains("# TYPE yip_traffic_rx_bytes_total counter")); + assert!(m_body.contains("# HELP yip_traffic_tx_packets_total Total packets dispatched")); + assert!(m_body.contains("# TYPE yip_traffic_tx_packets_total counter")); + assert!(m_body.contains("# HELP yip_traffic_rx_packets_total Total packets received")); + assert!(m_body.contains("# TYPE yip_traffic_rx_packets_total counter")); + assert!(m_body.contains("# HELP yip_traffic_dropped_packets_total Packets dropped due to route errors or queue overflow")); + assert!(m_body.contains("# TYPE yip_traffic_dropped_packets_total counter")); + assert!( + m_body.ends_with('\n'), + "prometheus exposition must end with newline" + ); + + // 4. 404 Not Found on unknown path + let (nf_status, _, nf_body) = http_request(addr, "GET", "/unknown_route"); + assert_eq!(nf_status, "HTTP/1.1 404 Not Found"); + assert_eq!(nf_body, "Not Found"); + + // 5. 405 Method Not Allowed on invalid method + let (na_status, _, na_body) = http_request(addr, "POST", "/metrics"); + assert_eq!(na_status, "HTTP/1.1 405 Method Not Allowed"); + assert_eq!(na_body, "Method Not Allowed"); + + server.stop(); +} + +#[test] +fn test_monotonic_counter_progression_under_simulated_packet_traffic() { + let mut harness = Harness::new("monotonic_traffic"); + let registry = harness.pm.metrics(); + let cached_stats = Arc::clone(&harness.cached_stats); + + let server = MetricsHttpServer::start( + "127.0.0.1:0".parse().unwrap(), + Arc::clone(®istry), + Box::new(move || cached_stats.lock().unwrap().clone()), + ) + .expect("start MetricsHttpServer"); + let http_addr = server.local_addr(); + + // ── Phase 0: Baseline Verification ────────────────────────────────── + let initial_resp = harness + .query_cmd(ControlCmd::GetMetrics { peer: None }) + .expect("initial query"); + let initial_metrics = match initial_resp { + ControlResponse::Metrics { metrics, .. } => metrics, + other => panic!("expected ControlResponse::Metrics, got: {other:?}"), + }; + assert_eq!(initial_metrics.traffic.tx_bytes, 0); + assert_eq!(initial_metrics.traffic.tx_packets, 0); + assert_eq!(initial_metrics.traffic.rx_bytes, 0); + assert_eq!(initial_metrics.traffic.rx_packets, 0); + assert_eq!(initial_metrics.traffic.dropped_packets, 0); + + let (_, _, initial_http_body) = http_request(http_addr, "GET", "/metrics"); + assert!(initial_http_body.contains("yip_traffic_tx_bytes_total 0")); + assert!(initial_http_body.contains("yip_traffic_rx_bytes_total 0")); + + // ── Phase 1: Synthetic Traffic Burst 1 ────────────────────────────── + { + // Dispatched 3 test TX packets + harness.pm.record_test_tx_packet(1420); + harness.pm.record_test_tx_packet(1420); + harness.pm.record_test_tx_packet(1420); + + // Ingress 2 test RX packets + harness.pm.metrics().record_rx(2840, 2); + + // Ingress malformed/empty UDP packet to trigger data plane drop + let dummy_addr: SocketAddr = "192.0.2.1:54321".parse().unwrap(); + let _ = harness.pm.on_udp(dummy_addr, &[], 0); + + // FEC and crypto events + harness.pm.metrics().record_fec_tx(10); + harness.pm.metrics().record_fec_recovered(8); + harness.pm.metrics().record_daita_padding(256); + harness.pm.metrics().record_handshake_success(); + harness.pm.metrics().record_rekey(); + } + harness.refresh_cached_stats(); + + // Verify over UNIX control socket + let p1_resp = harness + .query_cmd(ControlCmd::GetMetrics { peer: None }) + .expect("phase 1 query"); + let p1_metrics = match p1_resp { + ControlResponse::Metrics { + metrics, + prometheus, + .. + } => { + assert!(prometheus.contains("yip_traffic_tx_bytes_total 4260")); + assert!(prometheus.contains("yip_traffic_tx_packets_total 3")); + assert!(prometheus.contains("yip_traffic_rx_bytes_total 2840")); + assert!(prometheus.contains("yip_traffic_rx_packets_total 2")); + assert!(prometheus.contains("yip_traffic_dropped_packets_total 1")); + assert!(prometheus.contains("yip_fec_symbols_tx_total 10")); + assert!(prometheus.contains("yip_fec_recovered_symbols_total 8")); + assert!(prometheus.contains("yip_daita_padding_bytes_total 256")); + assert!(prometheus.contains("yip_crypto_handshakes_total 1")); + assert!(prometheus.contains("yip_crypto_rekeys_total 1")); + metrics + } + other => panic!("expected ControlResponse::Metrics, got: {other:?}"), + }; + + assert_eq!(p1_metrics.traffic.tx_bytes, 4260); + assert_eq!(p1_metrics.traffic.tx_packets, 3); + assert_eq!(p1_metrics.traffic.rx_bytes, 2840); + assert_eq!(p1_metrics.traffic.rx_packets, 2); + assert_eq!(p1_metrics.traffic.dropped_packets, 1); + assert_eq!(p1_metrics.fec.symbols_tx, 10); + assert_eq!(p1_metrics.fec.recovered_symbols, 8); + assert_eq!(p1_metrics.daita.padding_bytes, 256); + assert_eq!(p1_metrics.crypto.handshakes, 1); + assert_eq!(p1_metrics.crypto.rekeys, 1); + + // Verify over HTTP endpoint + let (_, _, p1_http_body) = http_request(http_addr, "GET", "/metrics"); + assert!(p1_http_body.contains("yip_traffic_tx_bytes_total 4260")); + assert!(p1_http_body.contains("yip_traffic_tx_packets_total 3")); + assert!(p1_http_body.contains("yip_traffic_rx_bytes_total 2840")); + assert!(p1_http_body.contains("yip_traffic_rx_packets_total 2")); + assert!(p1_http_body.contains("yip_traffic_dropped_packets_total 1")); + assert!(p1_http_body.contains("yip_fec_symbols_tx_total 10")); + assert!(p1_http_body.contains("yip_fec_recovered_symbols_total 8")); + assert!(p1_http_body.contains("yip_crypto_rekeys_total 1")); + + // ── Phase 2: Synthetic Traffic Burst 2 (Monotonic Verification) ───── + { + // Additional 2 TX packets + harness.pm.record_test_tx_packet(1000); + harness.pm.record_test_tx_packet(1000); + + // Additional 1 RX packet + harness.pm.metrics().record_rx(1500, 1); + + // Additional dropped packet + harness.pm.metrics().record_drop(2); + + // Additional rekey + harness.pm.metrics().record_rekey(); + } + harness.refresh_cached_stats(); + + let p2_resp = harness + .query_cmd(ControlCmd::GetMetrics { peer: None }) + .expect("phase 2 query"); + let p2_metrics = match p2_resp { + ControlResponse::Metrics { metrics, .. } => metrics, + other => panic!("expected ControlResponse::Metrics, got: {other:?}"), + }; + + // Strict monotonic progression assertions + assert!( + p2_metrics.traffic.tx_bytes > p1_metrics.traffic.tx_bytes, + "tx_bytes must strictly increase: {} <= {}", + p2_metrics.traffic.tx_bytes, + p1_metrics.traffic.tx_bytes + ); + assert_eq!(p2_metrics.traffic.tx_bytes, 6260); + + assert!( + p2_metrics.traffic.tx_packets > p1_metrics.traffic.tx_packets, + "tx_packets must strictly increase: {} <= {}", + p2_metrics.traffic.tx_packets, + p1_metrics.traffic.tx_packets + ); + assert_eq!(p2_metrics.traffic.tx_packets, 5); + + assert!( + p2_metrics.traffic.rx_bytes > p1_metrics.traffic.rx_bytes, + "rx_bytes must strictly increase: {} <= {}", + p2_metrics.traffic.rx_bytes, + p1_metrics.traffic.rx_bytes + ); + assert_eq!(p2_metrics.traffic.rx_bytes, 4340); + + assert!( + p2_metrics.traffic.rx_packets > p1_metrics.traffic.rx_packets, + "rx_packets must strictly increase: {} <= {}", + p2_metrics.traffic.rx_packets, + p1_metrics.traffic.rx_packets + ); + assert_eq!(p2_metrics.traffic.rx_packets, 3); + + assert!( + p2_metrics.traffic.dropped_packets > p1_metrics.traffic.dropped_packets, + "dropped_packets must strictly increase: {} <= {}", + p2_metrics.traffic.dropped_packets, + p1_metrics.traffic.dropped_packets + ); + assert_eq!(p2_metrics.traffic.dropped_packets, 3); + + assert!( + p2_metrics.crypto.rekeys > p1_metrics.crypto.rekeys, + "rekeys must strictly increase: {} <= {}", + p2_metrics.crypto.rekeys, + p1_metrics.crypto.rekeys + ); + assert_eq!(p2_metrics.crypto.rekeys, 2); + + // Verify over HTTP endpoint matches exact monotonically updated values + let (_, _, p2_http_body) = http_request(http_addr, "GET", "/metrics"); + assert!(p2_http_body.contains("yip_traffic_tx_bytes_total 6260")); + assert!(p2_http_body.contains("yip_traffic_tx_packets_total 5")); + assert!(p2_http_body.contains("yip_traffic_rx_bytes_total 4340")); + assert!(p2_http_body.contains("yip_traffic_rx_packets_total 3")); + assert!(p2_http_body.contains("yip_traffic_dropped_packets_total 3")); + assert!(p2_http_body.contains("yip_crypto_rekeys_total 2")); + + // ── Phase 3: Concurrent Multi-Threaded Progression ───────────────── + let thread_count = 4; + let ops_per_thread = 50; + let bytes_per_op = 100u64; + + let mut handles = Vec::new(); + for _ in 0..thread_count { + let reg = Arc::clone(®istry); + handles.push(std::thread::spawn(move || { + for _ in 0..ops_per_thread { + reg.record_tx(bytes_per_op, 1); + reg.record_rx(bytes_per_op, 1); + } + })); + } + + for h in handles { + h.join().expect("worker thread failed"); + } + harness.refresh_cached_stats(); + + let p3_resp = harness + .query_cmd(ControlCmd::GetMetrics { peer: None }) + .expect("phase 3 query"); + let p3_metrics = match p3_resp { + ControlResponse::Metrics { metrics, .. } => metrics, + other => panic!("expected ControlResponse::Metrics, got: {other:?}"), + }; + + let expected_added_bytes = thread_count as u64 * ops_per_thread as u64 * bytes_per_op; + let expected_added_packets = thread_count as u64 * ops_per_thread as u64; + + assert_eq!( + p3_metrics.traffic.tx_bytes, + p2_metrics.traffic.tx_bytes + expected_added_bytes + ); + assert_eq!( + p3_metrics.traffic.tx_packets, + p2_metrics.traffic.tx_packets + expected_added_packets + ); + assert_eq!( + p3_metrics.traffic.rx_bytes, + p2_metrics.traffic.rx_bytes + expected_added_bytes + ); + assert_eq!( + p3_metrics.traffic.rx_packets, + p2_metrics.traffic.rx_packets + expected_added_packets + ); + + // Verify consistency between control query and HTTP scraper under final counts + let (_, _, p3_http_body) = http_request(http_addr, "GET", "/metrics"); + let expected_tx_str = format!("yip_traffic_tx_bytes_total {}", p3_metrics.traffic.tx_bytes); + let expected_rx_str = format!("yip_traffic_rx_bytes_total {}", p3_metrics.traffic.rx_bytes); + assert!( + p3_http_body.contains(&expected_tx_str), + "HTTP body missing {expected_tx_str}" + ); + assert!( + p3_http_body.contains(&expected_rx_str), + "HTTP body missing {expected_rx_str}" + ); + + server.stop(); +} diff --git a/tests/metrics_endpoint_test.rs b/tests/metrics_endpoint_test.rs new file mode 120000 index 0000000..2d15cf4 --- /dev/null +++ b/tests/metrics_endpoint_test.rs @@ -0,0 +1 @@ +../bin/yipd/tests/metrics_endpoint_test.rs \ No newline at end of file From 70428ce0903cdf76acbf0a7ee41dde7eda5c2156 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Thu, 29 Oct 2026 21:07:25 -0400 Subject: [PATCH 8/9] docs: update changelog and roadmap for prometheus and opentelemetry telemetry (#188) --- CHANGELOG.md | 4 ++++ ROADMAP.md | 1 + 2 files changed, 5 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 080ce44..3d06fff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,10 @@ All notable changes to this project are documented here, following ## [Unreleased] +### Added +- Native Prometheus text format exposition and OpenTelemetry mesh telemetry (`ControlCmd::GetMetrics`, `yip metrics`). +- Embedded, zero-dependency HTTP/1.1 metrics scraper endpoint (`--metrics-listen `, `GET /metrics`, `GET /healthz`). + ## [0.1.0] - 2026-10-27 ### Security diff --git a/ROADMAP.md b/ROADMAP.md index c431f65..54636c2 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -62,6 +62,7 @@ maintainer. - **Comprehensive documentation overhaul & landing page architecture (#170, PR #175)** — transformed repository landing page into an accessible, structured guide with clear "What is yip?", "When to use / when not to use", simplified quickstart, audience-specific pathways, terminal telemetry demo GIF, sanitized production-grade zero-trust CA key management, standardized default port 443 (with 8443 unprivileged fallback), best-of-3 AWS WAN benchmark measurements, and 6 standalone guides in `docs/` (`01-getting-started.md`, `02-configuration.md`, `03-architecture.md`, `04-security-and-threat-model.md`, `05-benchmarking-and-performance.md`, `06-troubleshooting.md`, plus `motivation.md`). - **Decentralized Threshold CA Quorum via RFC 9591 FROST Ed25519 & ROAST (#39, PR #180)** — online threshold Certificate Authority daemon (`yip-ca serve --role coordinator` / `--role signer`, `yip-ca enroll`, `dealer-gen`) coexisting alongside offline CA root sets. Employs RFC 9591 FROST(Ed25519, SHA-512) and ROAST asynchronous coordination to tolerate up to *n* - *t* delayed or Byzantine signers without liveness stalls, pre-cached 1-RTT nonce pools, memory-locked (`mlock`) ZeroizeOnDrop ephemeral nonce vault bound to 128-bit CSPRNG boot-epoch IDs, CVE-2025-58359 threshold-drift defense, online lost-share repair, and automated zero-touch 70% TTL background rotation inside `yipd`. - **Unified Rust SDK Facade Crate (`yip-mesh`) & Crates.io Publishing Suite** — unified top-level facade library (`yip-mesh`) providing modular feature flags (`crypto`, `wire`, `transport`, `membership`, `device`, `io`, `rendezvous`, `obf`, `utls`, `ca`, `full`) and docs.rs documentation metadata across the 13 workspace crates. +- **Prometheus & OpenTelemetry telemetry exposition (#188)** — native Prometheus 0.0.4 text format exposition and OpenTelemetry mesh telemetry (`ControlCmd::GetMetrics`, `yip metrics`), lock-free atomic `MetricsRegistry` in `yipd`, and embedded zero-dependency HTTP/1.1 scraper endpoint (`--metrics-listen `, `GET /metrics`, `GET /healthz`). ## Now From 53267238377a001e4d67a7890892d4983abc027a Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Thu, 29 Oct 2026 21:28:04 -0400 Subject: [PATCH 9/9] fix(metrics): resolve review findings on peer metrics, concurrency, and epoch tracking - Add cached_peer_stats to MetricsRegistry and fallback in render_prometheus and snapshot - Supply live peer metrics from registry to production HTTP scraper in tunnel.rs - Bound MetricsHttpServer concurrency to 16 with HTTP 429 response on saturation - Increase WouldBlock sleep to 100ms in MetricsHttpServer accept loop - Cap HTTP scraper request reader with take(4096) to prevent unbounded allocations - Update yip_crypto_current_epoch metric gauge on initiator and responder epoch promotions - Add unit and integration tests covering peer stats fallback and bounded concurrency --- bin/yipd/src/epoch.rs | 8 +++ bin/yipd/src/metrics.rs | 85 ++++++++++++++++++++++--- bin/yipd/src/peer_manager/handshake.rs | 3 + bin/yipd/src/peer_manager/mod.rs | 17 +++++ bin/yipd/src/peer_manager/rekey.rs | 9 ++- bin/yipd/src/peer_manager/relay.rs | 22 ++++++- bin/yipd/src/peer_manager/routing.rs | 30 ++++++++- bin/yipd/src/tunnel.rs | 6 +- bin/yipd/tests/metrics_endpoint_test.rs | 73 +++++++++++++++++++++ 9 files changed, 234 insertions(+), 19 deletions(-) diff --git a/bin/yipd/src/epoch.rs b/bin/yipd/src/epoch.rs index 0d81cca..52d91bd 100644 --- a/bin/yipd/src/epoch.rs +++ b/bin/yipd/src/epoch.rs @@ -75,6 +75,7 @@ pub struct EpochSet { pub(crate) previous: Option>, pub(crate) previous_retire_ms: u64, pub(crate) rekey: Option, + pub(crate) epoch: u64, } impl EpochSet { @@ -86,9 +87,14 @@ impl EpochSet { previous: None, previous_retire_ms: 0, rekey: None, + epoch: 1, } } + pub fn epoch(&self) -> u64 { + self.epoch + } + pub fn current(&self) -> &DataPlane { &self.current } @@ -212,6 +218,7 @@ impl EpochSet { self.current_created_ms = now_ms; self.previous = Some(old); self.previous_retire_ms = now_ms.saturating_add(PREVIOUS_EPOCH_GRACE_MS); + self.epoch = self.epoch.saturating_add(1); return out; } } @@ -231,6 +238,7 @@ impl EpochSet { self.previous = Some(old); self.previous_retire_ms = now_ms.saturating_add(PREVIOUS_EPOCH_GRACE_MS); self.rekey = None; + self.epoch = self.epoch.saturating_add(1); } /// Responder: a new epoch derived from a received rekey Init, kept for diff --git a/bin/yipd/src/metrics.rs b/bin/yipd/src/metrics.rs index e1cc9b1..eb2d09c 100644 --- a/bin/yipd/src/metrics.rs +++ b/bin/yipd/src/metrics.rs @@ -1,19 +1,23 @@ //! Lock-free metrics registry and Prometheus 0.0.4 exposition formatter. use std::fmt::Write; -use std::io::{BufRead, BufReader, Write as IoWrite}; +use std::io::{BufRead, BufReader, Read, Write as IoWrite}; use std::net::{SocketAddr, TcpListener, TcpStream}; -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, RwLock}; use std::thread::JoinHandle; use std::time::{Duration, Instant}; use serde::{Deserialize, Serialize}; +/// Maximum concurrent HTTP client connections allowed to the scraper server. +pub const MAX_CONCURRENT_HTTP_CLIENTS: usize = 16; + /// Lock-free metrics registry holding data-plane counters and gauges. #[derive(Debug)] pub struct MetricsRegistry { start_time: Instant, + cached_peer_stats: RwLock>, // Traffic tx_bytes: AtomicU64, rx_bytes: AtomicU64, @@ -51,6 +55,7 @@ impl MetricsRegistry { pub fn new() -> Self { Self { start_time: Instant::now(), + cached_peer_stats: RwLock::new(Vec::new()), tx_bytes: AtomicU64::new(0), rx_bytes: AtomicU64::new(0), tx_packets: AtomicU64::new(0), @@ -73,6 +78,21 @@ impl MetricsRegistry { } } + /// Update the cached peer telemetry statistics for exposition fallback. + pub fn update_peer_stats(&self, stats: Vec) { + if let Ok(mut lock) = self.cached_peer_stats.write() { + *lock = stats; + } + } + + /// Retrieve the cached peer telemetry statistics. + pub fn get_cached_peer_stats(&self) -> Vec { + self.cached_peer_stats + .read() + .map(|lock| lock.clone()) + .unwrap_or_default() + } + /// Record transmitted payload and packets. pub fn record_tx(&self, bytes: u64, packets: u64) { self.tx_bytes.fetch_add(bytes, Ordering::Relaxed); @@ -160,6 +180,13 @@ impl MetricsRegistry { /// Take a structured snapshot of all telemetry metrics. pub fn snapshot(&self, peer_stats: &[crate::peer_manager::PeerStats]) -> TelemetrySnapshot { + let fallback_stats; + let peer_stats = if peer_stats.is_empty() { + fallback_stats = self.get_cached_peer_stats(); + &fallback_stats[..] + } else { + peer_stats + }; let uptime_seconds = self.start_time.elapsed().as_secs(); let tx_bytes = self.tx_bytes.load(Ordering::Relaxed); let rx_bytes = self.rx_bytes.load(Ordering::Relaxed); @@ -265,6 +292,13 @@ impl MetricsRegistry { /// Render metrics in standard Prometheus 0.0.4 text format. pub fn render_prometheus(&self, peer_stats: &[crate::peer_manager::PeerStats]) -> String { + let fallback_stats; + let peer_stats = if peer_stats.is_empty() { + fallback_stats = self.get_cached_peer_stats(); + &fallback_stats[..] + } else { + peer_stats + }; let mut out = String::with_capacity(2048); let uptime_seconds = self.start_time.elapsed().as_secs(); @@ -587,6 +621,7 @@ pub struct MetricsHttpServer { local_addr: SocketAddr, shutdown: Arc, thread_handle: Mutex>>, + active_conns: Arc, } impl std::fmt::Debug for MetricsHttpServer { @@ -594,6 +629,7 @@ impl std::fmt::Debug for MetricsHttpServer { f.debug_struct("MetricsHttpServer") .field("local_addr", &self.local_addr) .field("shutdown", &self.shutdown.load(Ordering::Relaxed)) + .field("active_conns", &self.active_conns.load(Ordering::Relaxed)) .finish() } } @@ -611,6 +647,8 @@ impl MetricsHttpServer { let shutdown = Arc::new(AtomicBool::new(false)); let shutdown_clone = Arc::clone(&shutdown); + let active_conns = Arc::new(AtomicUsize::new(0)); + let active_conns_clone = Arc::clone(&active_conns); let stats_fn: Arc Vec + Send + Sync + 'static> = Arc::from(stats_fn); @@ -619,20 +657,40 @@ impl MetricsHttpServer { .spawn(move || { while !shutdown_clone.load(Ordering::Relaxed) { match listener.accept() { - Ok((stream, _)) => { + Ok((mut stream, _)) => { + if active_conns_clone.load(Ordering::Relaxed) >= MAX_CONCURRENT_HTTP_CLIENTS { + let _ = IoWrite::write_all( + &mut stream, + b"HTTP/1.1 429 Too Many Requests\r\nConnection: close\r\n\r\ntoo many requests\n", + ); + let _ = IoWrite::flush(&mut stream); + continue; + } + active_conns_clone.fetch_add(1, Ordering::Relaxed); let reg = Arc::clone(®istry); let s_fn = Arc::clone(&stats_fn); - let _ = std::thread::Builder::new() + let conns = Arc::clone(&active_conns_clone); + let spawned = std::thread::Builder::new() .name("metrics-http-conn".into()) .spawn(move || { + struct ConnGuard(Arc); + impl Drop for ConnGuard { + fn drop(&mut self) { + self.0.fetch_sub(1, Ordering::Relaxed); + } + } + let _guard = ConnGuard(conns); let _ = handle_client(stream, ®, &*s_fn); }); + if spawned.is_err() { + active_conns_clone.fetch_sub(1, Ordering::Relaxed); + } } Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { - std::thread::sleep(Duration::from_millis(10)); + std::thread::sleep(Duration::from_millis(100)); } Err(_) => { - std::thread::sleep(Duration::from_millis(10)); + std::thread::sleep(Duration::from_millis(100)); } } } @@ -642,6 +700,7 @@ impl MetricsHttpServer { local_addr, shutdown, thread_handle: Mutex::new(Some(thread_handle)), + active_conns, }) } @@ -650,6 +709,11 @@ impl MetricsHttpServer { self.local_addr } + /// Return the number of currently active HTTP scraper connections. + pub fn active_connections(&self) -> usize { + self.active_conns.load(Ordering::Relaxed) + } + /// Signal shutdown to the background server thread and await its termination. pub fn stop(&self) { if self.shutdown.swap(true, Ordering::SeqCst) { @@ -679,7 +743,7 @@ fn handle_client( stream.set_write_timeout(Some(Duration::from_millis(3000)))?; let (method, path) = { - let mut reader = BufReader::new(&mut stream); + let mut reader = BufReader::new(&mut stream).take(4096); let mut first_line = String::new(); if reader.read_line(&mut first_line)? == 0 { return Ok(()); @@ -728,7 +792,10 @@ fn handle_client( match path.as_str() { "/metrics" => { - let peer_stats = stats_fn(); + let mut peer_stats = stats_fn(); + if peer_stats.is_empty() { + peer_stats = registry.get_cached_peer_stats(); + } let body = registry.render_prometheus(&peer_stats); let response = format!( "HTTP/1.1 200 OK\r\nContent-Type: text/plain; version=0.0.4; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", diff --git a/bin/yipd/src/peer_manager/handshake.rs b/bin/yipd/src/peer_manager/handshake.rs index 940ccf5..bb3271f 100644 --- a/bin/yipd/src/peer_manager/handshake.rs +++ b/bin/yipd/src/peer_manager/handshake.rs @@ -176,6 +176,7 @@ impl PeerManager { } self.peers[idx].state = PeerState::Idle; self.peers[idx].admitted_cert_not_after = None; + self.metrics.update_peer_stats(self.list_peers()); } pub(super) fn handle_handshake_init( @@ -375,6 +376,7 @@ impl PeerManager { self.peers[idx].state = PeerState::Established(epoch_set); self.record_new_epoch(idx); self.metrics.record_handshake_success(); + self.metrics.update_peer_stats(self.list_peers()); DispatchOut::Udp(&self.egress) } @@ -469,6 +471,7 @@ impl PeerManager { self.peers[idx].state = PeerState::Established(epoch_set); self.record_new_epoch(idx); self.metrics.record_handshake_success(); + self.metrics.update_peer_stats(self.list_peers()); if self.egress.is_empty() { DispatchOut::None diff --git a/bin/yipd/src/peer_manager/mod.rs b/bin/yipd/src/peer_manager/mod.rs index bfe6bd1..5e6e4ce 100644 --- a/bin/yipd/src/peer_manager/mod.rs +++ b/bin/yipd/src/peer_manager/mod.rs @@ -508,6 +508,8 @@ pub struct PeerManager { pub split_tunnel: crate::control::SplitTunnelState, /// Shared lock-free metrics registry for daemon and telemetry reporting. metrics: std::sync::Arc, + /// Monotonic timestamp in milliseconds of the last cached metrics peer stats update. + last_metrics_update_ms: u64, } /// Dynamic DAITA traffic analysis defense runtime metrics. @@ -653,6 +655,7 @@ impl PeerManager { daita_stats: DaitaStats::default(), split_tunnel: crate::control::SplitTunnelState::default(), metrics: std::sync::Arc::new(crate::metrics::MetricsRegistry::new()), + last_metrics_update_ms: 0, }; // Roots are pre-vetted (CA-signed root set) and therefore always-admit, @@ -1580,6 +1583,7 @@ impl PeerManager { rtt_ms: None, transport_name: transport, }); + self.metrics.update_peer_stats(self.list_peers()); Ok(()) } @@ -1619,6 +1623,7 @@ impl PeerManager { } } + self.metrics.update_peer_stats(self.list_peers()); Ok(()) } @@ -1679,6 +1684,11 @@ impl PeerManager { .collect() } + /// Retrieve summary status telemetry for all known peers (alias for peer_stats). + pub fn list_peers(&self) -> Vec { + self.peer_stats() + } + // ── rendezvous / path helpers ───────────────────────────────────────── /// The configured rendezvous server address (only meaningful when a @@ -2015,6 +2025,13 @@ impl PeerManager { fn tick_dispatch(&mut self, now_ms: u64) -> Option<&[EgressDatagram]> { self.tick_egress.clear(); + if now_ms.saturating_sub(self.last_metrics_update_ms) >= 500 + || self.last_metrics_update_ms == 0 + { + self.last_metrics_update_ms = now_ms.max(1); + self.metrics.update_peer_stats(self.list_peers()); + } + // ── registration refresh ────────────────────────────────────────── // Keep our reflexive binding fresh on the server so peers can find us. if self.rendezvous.is_some() diff --git a/bin/yipd/src/peer_manager/rekey.rs b/bin/yipd/src/peer_manager/rekey.rs index df3fbfe..d8f62f0 100644 --- a/bin/yipd/src/peer_manager/rekey.rs +++ b/bin/yipd/src/peer_manager/rekey.rs @@ -512,19 +512,20 @@ impl PeerManager { self.push_rekey_egress(idx, dg, via_relay); } - let old_tag = { + let (old_tag, new_epoch, obf_key) = { let PeerState::Established(epochs) = &mut self.peers[idx].state else { unreachable!("state cannot change between the borrows above") }; let old_tag = epochs.current().conn_tag(); epochs.promote_from_rekey(dp, now_ms); - self.peers[idx].session_obf_key = epochs.current_obf_key(); - old_tag + (old_tag, epochs.epoch(), epochs.current_obf_key()) }; + self.peers[idx].session_obf_key = obf_key; self.by_tag.remove(&old_tag); self.by_tag.insert(conn_tag, idx); self.record_new_epoch(idx); self.metrics.record_rekey(); + self.metrics.set_current_epoch(new_epoch); if self.egress.is_empty() { DispatchOut::None @@ -802,6 +803,7 @@ mod tests { let new_tag = established_tag(&pm_a, 0).unwrap(); assert_ne!(new_tag, old_tag, "current must become the NEW epoch"); + assert_eq!(pm_a.metrics().snapshot(&[]).crypto.current_epoch, 2); match &pm_a.peers[0].state { PeerState::Established(epochs) => { assert!( @@ -835,6 +837,7 @@ mod tests { Some(new_tag), "B must have confirmed the switch to the SAME new epoch" ); + assert_eq!(pm_b.metrics().snapshot(&[]).crypto.current_epoch, 2); // NEW-epoch frame (B -> A, now both on `current`) opens via // `current`. diff --git a/bin/yipd/src/peer_manager/relay.rs b/bin/yipd/src/peer_manager/relay.rs index 36c5a5e..2afef25 100644 --- a/bin/yipd/src/peer_manager/relay.rs +++ b/bin/yipd/src/peer_manager/relay.rs @@ -339,6 +339,7 @@ impl PeerManager { PeerState::Established(Box::new(crate::epoch::EpochSet::new(dp, now_ms))); self.record_new_epoch(idx); self.metrics.record_handshake_success(); + self.metrics.update_peer_stats(self.list_peers()); for b in owned { if let Some(d) = self.relay_wrap(idx, b) { self.egress.push(d); @@ -418,6 +419,7 @@ impl PeerManager { PeerState::Established(Box::new(crate::epoch::EpochSet::new(dp, now_ms))); self.record_new_epoch(idx); self.metrics.record_handshake_success(); + self.metrics.update_peer_stats(self.list_peers()); for b in owned { if let Some(d) = self.relay_wrap(idx, b) { self.egress.push(d); @@ -448,13 +450,21 @@ impl PeerManager { if !dg.is_empty() && dg[0] == PacketType::Data as u8 { self.peers[idx].last_activity_ms = now_ms; } - let (tun, udp): (Option>, Vec>) = { + let ((tun, udp), promoted_epoch) = { let PeerState::Established(epochs) = &mut self.peers[idx].state else { return DispatchOut::None; }; let prev_rec = epochs.fec_recovered_count(); let prev_burst = epochs.fec_unrecoverable_count(); + let prev_epoch = epochs.epoch(); let opened = epochs.inbound_open(dg, now_ms); + let promoted_epoch = if epochs.epoch() != prev_epoch { + let ep = epochs.epoch(); + let obf = epochs.current_obf_key(); + Some((ep, obf)) + } else { + None + }; let rec_diff = epochs.fec_recovered_count().saturating_sub(prev_rec); if rec_diff > 0 { self.metrics.record_fec_recovered(rec_diff); @@ -463,7 +473,7 @@ impl PeerManager { if burst_diff > 0 { self.metrics.record_fec_burst_drop(burst_diff); } - match opened { + let res = match opened { crate::epoch::EpochInbound::None => { self.metrics.record_drop(1); (None, Vec::new()) @@ -483,8 +493,14 @@ impl PeerManager { self.metrics.record_rx(buf.len() as u64, 1); (Some(buf), dgs.iter().map(|d| d.bytes.clone()).collect()) } - } + }; + (res, promoted_epoch) }; + if let Some((ep, obf)) = promoted_epoch { + self.peers[idx].session_obf_key = obf; + self.metrics.record_rekey(); + self.metrics.set_current_epoch(ep); + } self.egress.clear(); for b in udp { if let Some(d) = self.relay_wrap(idx, b) { diff --git a/bin/yipd/src/peer_manager/routing.rs b/bin/yipd/src/peer_manager/routing.rs index b4995a5..0c3e053 100644 --- a/bin/yipd/src/peer_manager/routing.rs +++ b/bin/yipd/src/peer_manager/routing.rs @@ -181,7 +181,15 @@ impl PeerManager { // for relay-established peers (their `DataPlane::peer_addr` is a // `server_addr()` stand-in; `endpoint` may hold an unconfirmed // candidate or `None`). + let prev_epoch = epochs.epoch(); let opened = epochs.inbound_open(dg, now_ms); + let promoted_epoch = if epochs.epoch() != prev_epoch { + let ep = epochs.epoch(); + let obf = epochs.current_obf_key(); + Some((ep, obf)) + } else { + None + }; let rec_diff = epochs.fec_recovered_count().saturating_sub(prev_rec); if rec_diff > 0 { self.metrics.record_fec_recovered(rec_diff); @@ -190,6 +198,11 @@ impl PeerManager { if burst_diff > 0 { self.metrics.record_fec_burst_drop(burst_diff); } + if let Some((ep, obf)) = promoted_epoch { + self.peers[idx].session_obf_key = obf; + self.metrics.record_rekey(); + self.metrics.set_current_epoch(ep); + } if !matches!(opened, crate::epoch::EpochInbound::None) { // Authenticated + non-replayed (M2 roaming) — safe to roam. self.relearn_endpoint(idx, src); @@ -258,13 +271,21 @@ impl PeerManager { .map(|(i, _)| i) .collect(); for idx in candidates { - let (hit, rec_diff, burst_diff) = { + let (hit, rec_diff, burst_diff, promoted_epoch) = { let PeerState::Established(epochs) = &mut self.peers[idx].state else { continue; }; let prev_rec = epochs.fec_recovered_count(); let prev_burst = epochs.fec_unrecoverable_count(); + let prev_epoch = epochs.epoch(); let opened = epochs.inbound_open(dg, now_ms); + let promoted_epoch = if epochs.epoch() != prev_epoch { + let ep = epochs.epoch(); + let obf = epochs.current_obf_key(); + Some((ep, obf)) + } else { + None + }; let rec_diff = epochs.fec_recovered_count().saturating_sub(prev_rec); let burst_diff = epochs.fec_unrecoverable_count().saturating_sub(prev_burst); let h = match opened { @@ -273,8 +294,13 @@ impl PeerManager { crate::epoch::EpochInbound::Send(dgs) => Some((None, dgs)), crate::epoch::EpochInbound::TunThenSend(buf, dgs) => Some((Some(buf), dgs)), }; - (h, rec_diff, burst_diff) + (h, rec_diff, burst_diff, promoted_epoch) }; + if let Some((ep, obf)) = promoted_epoch { + self.peers[idx].session_obf_key = obf; + self.metrics.record_rekey(); + self.metrics.set_current_epoch(ep); + } if rec_diff > 0 { self.metrics.record_fec_recovered(rec_diff); } diff --git a/bin/yipd/src/tunnel.rs b/bin/yipd/src/tunnel.rs index 0be5138..022f50d 100644 --- a/bin/yipd/src/tunnel.rs +++ b/bin/yipd/src/tunnel.rs @@ -209,11 +209,13 @@ pub fn run(config: Config) -> io::Result<()> { config.device.clone(), std::time::Instant::now(), ); + let metrics_registry = manager.metrics(); + let reg_for_stats = std::sync::Arc::clone(&metrics_registry); let _metrics_server = match config.metrics_listen { Some(addr) => Some(crate::metrics::MetricsHttpServer::start( addr, - manager.metrics(), - Box::new(Vec::new), + metrics_registry, + Box::new(move || reg_for_stats.get_cached_peer_stats()), )?), None => None, }; diff --git a/bin/yipd/tests/metrics_endpoint_test.rs b/bin/yipd/tests/metrics_endpoint_test.rs index 4f52ad4..10ba366 100644 --- a/bin/yipd/tests/metrics_endpoint_test.rs +++ b/bin/yipd/tests/metrics_endpoint_test.rs @@ -632,3 +632,76 @@ fn test_monotonic_counter_progression_under_simulated_packet_traffic() { server.stop(); } + +#[test] +fn test_metrics_registry_cached_peer_stats_fallback_in_http_scraper() { + let registry = Arc::new(yipd::metrics::MetricsRegistry::new()); + let listen_addr: SocketAddr = "127.0.0.1:0".parse().unwrap(); + + // Start server with an empty stats closure (simulating production fallback) + let server = MetricsHttpServer::start(listen_addr, Arc::clone(®istry), Box::new(Vec::new)) + .expect("failed to start metrics HTTP server"); + let http_addr = server.local_addr(); + + // Initial scrape: no peers cached yet, active should be 0 + let (status, _, body) = http_request(http_addr, "GET", "/metrics"); + assert!(status.contains("200 OK")); + assert!(body.contains("yip_peers_active 0")); + assert!(body.contains("yip_peers_total 0")); + + // Update peer stats via MetricsRegistry::update_peer_stats + let peer = PeerStats { + public_key: [7u8; 32], + endpoint: Some("192.168.1.100:51820".parse().unwrap()), + node_addr: "fd00::7".parse().unwrap(), + state: "established".to_string(), + last_activity_ms: 12345, + tx_bytes: 54321, + rx_bytes: 12345, + tx_packets: 42, + rx_packets: 24, + fec_recovered: 5, + transport: "Direct UDP".to_string(), + rtt_ms: Some(15), + fec_ratio_pct: 3, + cert_status: "CA Valid".to_string(), + }; + registry.update_peer_stats(vec![peer]); + + // Scrape again: even though stats_fn returns empty, registry cached stats fallback must populate metrics + let (status2, _, body2) = http_request(http_addr, "GET", "/metrics"); + assert!(status2.contains("200 OK")); + assert!(body2.contains("yip_peers_active 1")); + assert!(body2.contains("yip_peers_total 1")); + assert!( + body2.contains("yip_peer_tx_bytes_total{peer=\"fd00::7\",transport=\"Direct UDP\"} 54321") + ); + assert!( + body2.contains("yip_peer_rtt_milliseconds{peer=\"fd00::7\",transport=\"Direct UDP\"} 15") + ); + + // Also verify get_cached_peer_stats + let cached = registry.get_cached_peer_stats(); + assert_eq!(cached.len(), 1); + assert_eq!(cached[0].tx_bytes, 54321); + + server.stop(); +} + +#[test] +fn test_metrics_http_server_bounded_concurrency_and_healthz() { + let registry = Arc::new(yipd::metrics::MetricsRegistry::new()); + let listen_addr: SocketAddr = "127.0.0.1:0".parse().unwrap(); + let server = MetricsHttpServer::start(listen_addr, Arc::clone(®istry), Box::new(Vec::new)) + .expect("failed to start metrics HTTP server"); + let http_addr = server.local_addr(); + + assert_eq!(yipd::metrics::MAX_CONCURRENT_HTTP_CLIENTS, 16); + assert_eq!(server.active_connections(), 0); + + let (status, _, body) = http_request(http_addr, "GET", "/healthz"); + assert!(status.contains("200 OK")); + assert_eq!(body, "ok"); + + server.stop(); +}