diff --git a/CHANGELOG.md b/CHANGELOG.md index 94df9bda..bb9f5d3d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ All notable changes to this project are documented here, following ## [Unreleased] ### Added +- **Prometheus and OpenTelemetry mesh telemetry exposition (`bin/yipd`, `yip metrics`, `--metrics-listen`, #188).** + Added native Prometheus 0.0.4 text-format telemetry exposition and structured OpenTelemetry-aligned snapshots (`ControlCmd::GetMetrics`, `yip metrics [--json]`). Built lock-free `MetricsRegistry` in `bin/yipd` backed by atomic counters for packet I/O, FEC symbol recovery, DAITA padding/decoys, cryptographic ratchets, and split-tunnel bypass. Implemented embedded, zero-dependency HTTP/1.1 scraper server (`--metrics-listen `, config `metrics_listen`) with bounded concurrency (16 max connections), 3000ms socket timeout guards, and `/metrics` and `/healthz` endpoints. - **Multi-hop circuit tracing and path diagnostic probes (`bin/yipd`, `yip trace`, `yip ping`, #187).** Added live circuit tracing (`ControlCmd::TracePath`) and latency diagnostic probes (`ControlCmd::PingPeer`) to the daemon IPC protocol. Supports flexible peer resolution via 64-character public key, short prefix hex, IPv6 node address, socket endpoint, or numeric index. Projects multi-hop path segments across direct P2P and TLS-relayed overlay routes, returning hop identifiers, next-hops, and segment latencies. Added ping sampling with deterministic jitter variance, packet loss accounting, and min/avg/max/stddev RTT calculations, exposed via CLI subcommands `yip trace ` and `yip ping [-n count] [-W timeout_ms] [--json]`. - **Live mesh routing table and topology discovery commands (`bin/yipd`, `yip routes`, `yip topology`, #183).** diff --git a/ROADMAP.md b/ROADMAP.md index 2f6d35f2..9a388f5f 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -65,6 +65,7 @@ maintainer. - **Live mesh routing table & interactive topology discovery (#183)** — live routing table queries (`ControlCmd::GetRoutes`) and interactive topology graph projection (`ControlCmd::GetTopology`) over UNIX domain control socket and CLI (`yip routes`, `yip topology`), mapping peers into latency categories (`Optimal`, `Normal`, `Elevated`) and computing next-hop metrics and MTU budgets. - **Split-tunnel CIDR, domain, and application route bypass control (#184)** — dynamic traffic bypass rule management (`ControlCmd::GetSplitTunnelStatus`, `ControlCmd::SetSplitTunnelConfig`, `yip split `) with zero-allocation bitmask subnet filtering in `PeerManager::on_tun_dispatch` for IPv4 and IPv6 traffic. - **Multi-hop circuit tracing and path diagnostic probes (#187)** — hop-by-hop mesh circuit tracing (`ControlCmd::TracePath`, `yip trace`) and active latency diagnostic probes (`ControlCmd::PingPeer`, `yip ping`) with min/avg/max/stddev RTT calculations, jitter sampling, and timeout-induced loss detection across direct and relayed circuits. +- **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 diff --git a/bin/yipd/src/config.rs b/bin/yipd/src/config.rs index 77bc61f0..e25073e0 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/control.rs b/bin/yipd/src/control.rs index c09947e2..1e07151f 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 = pm.metrics(); + 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()); + } } diff --git a/bin/yipd/src/epoch.rs b/bin/yipd/src/epoch.rs index 0d81cca6..52d91bd4 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/lib.rs b/bin/yipd/src/lib.rs index ec45f116..95891283 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 c5c4ad44..52e08175 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; @@ -49,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"); @@ -82,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!(); @@ -218,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"); @@ -725,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}"), @@ -732,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(); @@ -764,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)); @@ -806,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(()) @@ -840,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)?; @@ -882,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 => { @@ -928,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/metrics.rs b/bin/yipd/src/metrics.rs new file mode 100644 index 00000000..eb2d09c6 --- /dev/null +++ b/bin/yipd/src/metrics.rs @@ -0,0 +1,1139 @@ +//! Lock-free metrics registry and Prometheus 0.0.4 exposition formatter. + +use std::fmt::Write; +use std::io::{BufRead, BufReader, Read, Write as IoWrite}; +use std::net::{SocketAddr, TcpListener, TcpStream}; +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, + 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(), + cached_peer_stats: RwLock::new(Vec::new()), + 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), + } + } + + /// 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); + 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 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); + 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 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(); + 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, +} + +/// Embedded HTTP/1.1 scraper server exposing Prometheus metrics and health check. +pub struct MetricsHttpServer { + local_addr: SocketAddr, + shutdown: Arc, + thread_handle: Mutex>>, + active_conns: Arc, +} + +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)) + .field("active_conns", &self.active_conns.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 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); + + let thread_handle = std::thread::Builder::new() + .name("metrics-http-server".into()) + .spawn(move || { + while !shutdown_clone.load(Ordering::Relaxed) { + match listener.accept() { + 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 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(100)); + } + Err(_) => { + std::thread::sleep(Duration::from_millis(100)); + } + } + } + })?; + + Ok(Self { + local_addr, + shutdown, + thread_handle: Mutex::new(Some(thread_handle)), + active_conns, + }) + } + + /// Return the local socket address this server is listening on. + pub fn local_addr(&self) -> SocketAddr { + 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) { + 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).take(4096); + 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 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{}", + 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::*; + 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); + } + + #[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(); + } +} diff --git a/bin/yipd/src/peer_manager/handshake.rs b/bin/yipd/src/peer_manager/handshake.rs index 0e5660c0..bb3271f4 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; } }; @@ -175,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( @@ -194,6 +196,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 +364,19 @@ 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(); + self.metrics.update_peer_stats(self.list_peers()); DispatchOut::Udp(&self.egress) } @@ -450,13 +459,19 @@ 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(); + self.metrics.update_peer_stats(self.list_peers()); if self.egress.is_empty() { DispatchOut::None @@ -466,6 +481,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 8dbf2285..5e6e4ce2 100644 --- a/bin/yipd/src/peer_manager/mod.rs +++ b/bin/yipd/src/peer_manager/mod.rs @@ -506,6 +506,10 @@ 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, + /// Monotonic timestamp in milliseconds of the last cached metrics peer stats update. + last_metrics_update_ms: u64, } /// Dynamic DAITA traffic analysis defense runtime metrics. @@ -650,6 +654,8 @@ 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()), + last_metrics_update_ms: 0, }; // Roots are pre-vetted (CA-signed root set) and therefore always-admit, @@ -1217,6 +1223,54 @@ 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 { + 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. pub fn set_threshold_ca_server(&mut self, addr: Option) { self.threshold_ca_server = addr; @@ -1529,6 +1583,7 @@ impl PeerManager { rtt_ms: None, transport_name: transport, }); + self.metrics.update_peer_stats(self.list_peers()); Ok(()) } @@ -1568,6 +1623,7 @@ impl PeerManager { } } + self.metrics.update_peer_stats(self.list_peers()); Ok(()) } @@ -1628,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 @@ -1718,6 +1779,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 +1802,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 +1867,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 +1896,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 +1925,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 +1953,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 +1963,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 { @@ -1951,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() @@ -2287,10 +2368,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 030f9800..ad5e2aa9 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 a36e5f0c..d8f62f0a 100644 --- a/bin/yipd/src/peer_manager/rekey.rs +++ b/bin/yipd/src/peer_manager/rekey.rs @@ -498,25 +498,34 @@ 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 { 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 @@ -794,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!( @@ -827,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 bf899cdd..2afef257 100644 --- a/bin/yipd/src/peer_manager/relay.rs +++ b/bin/yipd/src/peer_manager/relay.rs @@ -328,15 +328,18 @@ 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(); + 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); @@ -405,15 +408,18 @@ 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(); + 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); @@ -427,6 +433,7 @@ impl PeerManager { } Err(e) => { eprintln!("peer_manager: relayed read_response failed: {e}"); + self.metrics.record_handshake_failure(); DispatchOut::None } } @@ -440,21 +447,60 @@ 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<'_> { - let (tun, udp): (Option>, Vec>) = { + if !dg.is_empty() && dg[0] == PacketType::Data as u8 { + self.peers[idx].last_activity_ms = now_ms; + } + let ((tun, udp), promoted_epoch) = { 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 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); + } + let burst_diff = epochs.fec_unrecoverable_count().saturating_sub(prev_burst); + if burst_diff > 0 { + self.metrics.record_fec_burst_drop(burst_diff); + } + let res = 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()) } - } + }; + (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 0b452534..0c3e053b 100644 --- a/bin/yipd/src/peer_manager/routing.rs +++ b/bin/yipd/src/peer_manager/routing.rs @@ -173,22 +173,49 @@ 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 // 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); + } + let burst_diff = epochs.fec_unrecoverable_count().saturating_sub(prev_burst); + 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); } 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 +226,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 +271,49 @@ impl PeerManager { .map(|(i, _)| i) .collect(); for idx in candidates { - let hit = { + let (hit, rec_diff, burst_diff, promoted_epoch) = { 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 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 { 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, 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); + } + 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 +337,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 015d6db9..46b50116 100644 --- a/bin/yipd/src/peer_manager/tests.rs +++ b/bin/yipd/src/peer_manager/tests.rs @@ -2967,3 +2967,156 @@ 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); +} + +#[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); +} diff --git a/bin/yipd/src/peer_manager/testutil.rs b/bin/yipd/src/peer_manager/testutil.rs index 93eed371..066c7805 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 diff --git a/bin/yipd/src/tunnel.rs b/bin/yipd/src/tunnel.rs index a1a49e08..022f50df 100644 --- a/bin/yipd/src/tunnel.rs +++ b/bin/yipd/src/tunnel.rs @@ -209,6 +209,16 @@ 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, + metrics_registry, + Box::new(move || reg_for_stats.get_cached_peer_stats()), + )?), + None => None, + }; let local_addr = manager.local_addr(); // ── create the tunnel device (TUN or TAP) ──────────────────────────────── diff --git a/bin/yipd/tests/metrics_endpoint_test.rs b/bin/yipd/tests/metrics_endpoint_test.rs new file mode 100644 index 00000000..10ba3661 --- /dev/null +++ b/bin/yipd/tests/metrics_endpoint_test.rs @@ -0,0 +1,707 @@ +//! 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(); +} + +#[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(); +} diff --git a/tests/metrics_endpoint_test.rs b/tests/metrics_endpoint_test.rs new file mode 120000 index 00000000..2d15cf45 --- /dev/null +++ b/tests/metrics_endpoint_test.rs @@ -0,0 +1 @@ +../bin/yipd/tests/metrics_endpoint_test.rs \ No newline at end of file