feat(ipc): Prometheus and OpenTelemetry mesh telemetry exposition (closes #188)

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 <ADDR>`, 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 <peer>` and `yip ping <peer> [-n count] [-W timeout_ms] [--json]`.
 - **Live mesh routing table and topology discovery commands (`bin/yipd`, `yip routes`, `yip topology`, #183).**

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 <status|set>`) 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 <ADDR>`, `GET /metrics`, `GET /healthz`).
 
 ## Now
 

bin/yipd/src/config.rs

@@ -196,6 +196,8 @@ pub struct Config {
     pub control_socket: Option<std::path::PathBuf>,
     /// Configured TUN device MTU (`mtu=<576..=65535>`).
     pub mtu: Option<u32>,
+    /// Optional Prometheus/OpenTelemetry HTTP metrics scraping endpoint (`metrics_listen=<ip:port>`).
+    pub metrics_listen: Option<SocketAddr>,
 }
 
 // ── hex decode helper ─────────────────────────────────────────────────────────
@@ -547,6 +549,7 @@ impl Config {
         let mut shards: Option<usize> = None;
         let mut control_socket: Option<std::path::PathBuf> = None;
         let mut mtu: Option<u32> = None;
+        let mut metrics_listen: Option<SocketAddr> = 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::<SocketAddr>()
+                        .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());
+    }
 }

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<String>,
+    },
 }
 
 fn default_ping_count() -> u32 {
@@ -652,6 +657,11 @@ pub enum ControlResponse {
         stddev_rtt_ms: Option<f64>,
         samples: Vec<f64>,
     },
+    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::<String>()
+                            .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());
+    }
 }

bin/yipd/src/epoch.rs

@@ -75,6 +75,7 @@ pub struct EpochSet {
     pub(crate) previous: Option<Box<DataPlane>>,
     pub(crate) previous_retire_ms: u64,
     pub(crate) rekey: Option<RekeyInFlight>,
+    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

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;