feat: multipath metrics

Frando · Frando · commit 2f4092f7eb94 · 2025-11-17T14:13:08.000+01:00
diff --git a/iroh/src/magicsock/endpoint_map/endpoint_state.rs b/iroh/src/magicsock/endpoint_map/endpoint_state.rs
@@ -37,6 +37,9 @@ use crate::{
 
 mod guarded_channel;
 
+#[cfg(feature = "metrics")]
+mod metrics;
+
 // TODO: use this
 // /// Number of addresses that are not active that we keep around per endpoint.
 // ///
@@ -75,6 +78,9 @@ const APPLICATION_ABANDON_PATH: u8 = 30;
 /// in a high frequency, and to keep data about previous path around for subsequent connections.
 const ACTOR_MAX_IDLE_TIMEOUT: Duration = Duration::from_secs(60);
 
+/// Interval in which connection and path metrics are emitted.
+const METRICS_INTERVAL: Duration = Duration::from_secs(10);
+
 /// A stream of events from all paths for all connections.
 ///
 /// The connection is identified using [`ConnId`].  The event `Err` variant happens when the
@@ -220,6 +226,7 @@ impl EndpointStateActor {
         trace!("actor started");
         let idle_timeout = MaybeFuture::None;
         tokio::pin!(idle_timeout);
+        let mut metrics_interval = time::interval(METRICS_INTERVAL);
         loop {
             let scheduled_path_open = match self.scheduled_open_path {
                 Some(when) => MaybeFuture::Some(time::sleep_until(when)),
@@ -243,11 +250,7 @@ impl EndpointStateActor {
                     self.handle_path_event(id, evt);
                 }
                 Some(conn_id) = self.connections_close.next(), if !self.connections_close.is_empty() => {
-                    self.connections.remove(&conn_id);
-                    if self.connections.is_empty() {
-                        trace!("last connection closed - clearing selected_path");
-                        self.selected_path.set(None).ok();
-                    }
+                    self.handle_connection_close(conn_id);
                 }
                 _ = self.local_addrs.updated() => {
                     trace!("local addrs updated, triggering holepunching");
@@ -266,6 +269,9 @@ impl EndpointStateActor {
                     self.scheduled_holepunch = None;
                     self.trigger_holepunching().await;
                 }
+                _ = metrics_interval.tick() => {
+                    self.record_metrics();
+                }
                 _ = &mut idle_timeout => {
                     if self.connections.is_empty() && inbox.close_if_idle() {
                         trace!("idle timeout expired and still idle: terminate actor");
@@ -369,6 +375,7 @@ impl EndpointStateActor {
     ) {
         let pub_open_paths = Watchable::default();
         if let Some(conn) = handle.upgrade() {
+            self.metrics.num_conns_opened.inc();
             // Remove any conflicting stable_ids from the local state.
             let conn_id = ConnId(conn.stable_id());
             self.connections.remove(&conn_id);
@@ -387,6 +394,8 @@ impl EndpointStateActor {
                     paths: Default::default(),
                     open_paths: Default::default(),
                     path_ids: Default::default(),
+                    #[cfg(feature = "metrics")]
+                    metrics: Default::default(),
                 })
                 .into_mut();
 
@@ -571,6 +580,19 @@ impl EndpointStateActor {
         tx.send(rtt).ok();
     }
 
+    fn handle_connection_close(&mut self, conn_id: ConnId) {
+        if let Some(state) = self.connections.remove(&conn_id) {
+            #[cfg(feature = "metrics")]
+            state.metrics.record_closed(&self.metrics);
+            #[cfg(not(feature = "metrics"))]
+            let _ = state;
+        }
+        if self.connections.is_empty() {
+            trace!("last connection closed - clearing selected_path");
+            self.selected_path.set(None).ok();
+        }
+    }
+
     /// Triggers holepunching to the remote endpoint.
     ///
     /// This will manage the entire process of holepunching with the remote endpoint.
@@ -1012,6 +1034,13 @@ impl EndpointStateActor {
             }
         }
     }
+
+    fn record_metrics(&mut self) {
+        #[cfg(feature = "metrics")]
+        for state in self.connections.values_mut() {
+            state.record_metrics_periodic(&self.metrics, self.selected_path.get());
+        }
+    }
 }
 
 /// Messages to send to the [`EndpointStateActor`].
@@ -1114,6 +1143,11 @@ struct ConnectionState {
     open_paths: FxHashMap<PathId, transports::Addr>,
     /// Reverse map of [`Self::paths].
     path_ids: FxHashMap<transports::Addr, PathId>,
+    /// Tracker for stateful metrics for this connection and its paths
+    ///
+    /// Feature-gated on the `metrics` feature because it increases memory use.
+    #[cfg(feature = "metrics")]
+    metrics: self::metrics::MetricsTracker,
 }
 
 impl ConnectionState {
@@ -1125,10 +1159,11 @@ impl ConnectionState {
 
     /// Tracks an open path for the connection.
     fn add_open_path(&mut self, remote: transports::Addr, path_id: PathId) {
+        #[cfg(feature = "metrics")]
+        self.metrics.add_path(path_id, &remote);
         self.paths.insert(path_id, remote.clone());
         self.open_paths.insert(path_id, remote.clone());
         self.path_ids.insert(remote, path_id);
-
         self.update_pub_path_info();
     }
 
@@ -1138,11 +1173,15 @@ impl ConnectionState {
             self.path_ids.remove(&addr);
         }
         self.open_paths.remove(path_id);
+        #[cfg(feature = "metrics")]
+        self.metrics.remove_path(path_id);
     }
 
     /// Removes the path from the open paths.
     fn remove_open_path(&mut self, path_id: &PathId) {
         self.open_paths.remove(path_id);
+        #[cfg(feature = "metrics")]
+        self.metrics.remove_path(path_id);
 
         self.update_pub_path_info();
     }
@@ -1162,6 +1201,19 @@ impl ConnectionState {
 
         self.pub_open_paths.set(new).ok();
     }
+
+    #[cfg(feature = "metrics")]
+    fn record_metrics_periodic(
+        &mut self,
+        metrics: &MagicsockMetrics,
+        selected_path: Option<transports::Addr>,
+    ) {
+        let Some(conn) = self.handle.upgrade() else {
+            return;
+        };
+        self.metrics
+            .record_periodic(metrics, &conn, &self.open_paths, selected_path);
+    }
 }
 
 /// Watcher for the open paths and selected transmission path in a connection.
diff --git a/iroh/src/magicsock/endpoint_map/endpoint_state/metrics.rs b/iroh/src/magicsock/endpoint_map/endpoint_state/metrics.rs
@@ -0,0 +1,158 @@
+//! Tracker for stateful metrics for connections and paths.
+
+use std::time::Duration;
+
+use quinn_proto::PathId;
+use rustc_hash::FxHashMap;
+
+use crate::{magicsock::transports, metrics::MagicsockMetrics};
+
+#[derive(Debug, Default)]
+pub(super) struct MetricsTracker {
+    transport_summary: TransportSummary,
+    path_rtt_variance: FxHashMap<PathId, RttVariance>,
+}
+
+impl MetricsTracker {
+    pub(super) fn add_path(&mut self, path_id: PathId, remote: &transports::Addr) {
+        self.transport_summary.add_path(&remote);
+        self.path_rtt_variance.insert(path_id, Default::default());
+    }
+
+    pub(super) fn remove_path(&mut self, path_id: &PathId) {
+        self.path_rtt_variance.remove(path_id);
+    }
+
+    pub(super) fn record_periodic(
+        &mut self,
+        metrics: &MagicsockMetrics,
+        conn: &quinn::Connection,
+        path_remotes: &FxHashMap<PathId, transports::Addr>,
+        selected_path: Option<transports::Addr>,
+    ) {
+        for (path_id, remote) in path_remotes.iter() {
+            let Some(stats) = conn.path_stats(*path_id) else {
+                continue;
+            };
+
+            let loss_rate = if stats.sent_packets == 0 {
+                0.0
+            } else {
+                stats.lost_packets as f64 / stats.sent_packets as f64
+            };
+            metrics.path_packet_loss_rate.observe(loss_rate);
+
+            if Some(remote) == selected_path.as_ref() {
+                metrics
+                    .connection_latency_ms
+                    .observe(stats.rtt.as_millis() as f64);
+            }
+
+            let Some(rtt_variance) = self.path_rtt_variance.get_mut(path_id) else {
+                continue;
+            };
+            rtt_variance.add_rtt_sample(stats.rtt);
+            if let Some(variance) = rtt_variance.rtt_variance() {
+                metrics
+                    .path_rtt_variance_ms
+                    .observe(variance.as_millis() as f64);
+            }
+
+            let quality = rtt_variance.quality_score(loss_rate);
+            metrics.path_quality_score.observe(quality);
+        }
+    }
+
+    pub(super) fn record_closed(&self, metrics: &MagicsockMetrics) {
+        metrics.num_conns_closed.inc();
+        match self.transport_summary {
+            TransportSummary::IpOnly => {
+                metrics.num_conns_transport_ip_only.inc();
+            }
+            TransportSummary::RelayOnly => {
+                metrics.num_conns_transport_relay_only.inc();
+            }
+            TransportSummary::IpAndRelay => {
+                metrics.num_conns_transport_ip_and_relay.inc();
+            }
+            TransportSummary::None => {}
+        }
+    }
+}
+
+/// Tracks RTT variance over time, as a congestion marker.
+#[derive(Debug, Default)]
+struct RttVariance {
+    /// Rolling window of recent latency measurements (stores up to 8 samples).
+    samples: [Option<Duration>; 8],
+    /// Index for next sample insertion (circular buffer).
+    index: usize,
+}
+
+impl RttVariance {
+    fn add_rtt_sample(&mut self, rtt: Duration) {
+        self.samples[self.index] = Some(rtt);
+        self.index = (self.index + 1) % self.samples.len();
+    }
+
+    /// Calculate RTT variance as a congestion indicator.
+    /// Higher variance suggests congestion or unstable path.
+    fn rtt_variance(&self) -> Option<Duration> {
+        let samples: Vec<Duration> = self.samples.iter().filter_map(|&s| s).collect();
+
+        if samples.len() < 2 {
+            return None;
+        }
+
+        let mean = samples.iter().sum::<Duration>() / samples.len() as u32;
+        let variance: f64 = samples
+            .iter()
+            .map(|&s| {
+                let diff = s.as_secs_f64() - mean.as_secs_f64();
+                diff * diff
+            })
+            .sum::<f64>()
+            / samples.len() as f64;
+
+        Some(Duration::from_secs_f64(variance.sqrt()))
+    }
+
+    /// Path quality score (0.0 = worst, 1.0 = best).
+    /// Factors in packet loss and RTT variance.
+    fn quality_score(&self, packet_loss: f64) -> f64 {
+        let loss_penalty = (1.0 - packet_loss).clamp(0.0, 1.0);
+
+        // Penalize high RTT variance
+        let variance_penalty = match self.rtt_variance() {
+            Some(var) if var.as_millis() > 50 => 0.7,
+            Some(var) if var.as_millis() > 20 => 0.85,
+            Some(_) => 1.0,
+            None => 1.0,
+        };
+
+        loss_penalty * variance_penalty
+    }
+}
+
+/// Used for metrics tracking.
+#[derive(Debug, Clone, Copy, Default)]
+enum TransportSummary {
+    #[default]
+    None,
+    IpOnly,
+    RelayOnly,
+    IpAndRelay,
+}
+
+impl TransportSummary {
+    fn add_path(&mut self, addr: &transports::Addr) {
+        use transports::Addr;
+        *self = match (*self, addr) {
+            (TransportSummary::None | TransportSummary::IpOnly, Addr::Ip(_)) => Self::IpOnly,
+            (TransportSummary::None | TransportSummary::RelayOnly, Addr::Relay(_, _)) => {
+                Self::RelayOnly
+            }
+            _ => Self::IpAndRelay,
+        }
+    }
+}
diff --git a/iroh/src/magicsock/metrics.rs b/iroh/src/magicsock/metrics.rs
@@ -50,15 +50,20 @@ pub struct Metrics {
 
     /*
      * Connection Metrics
+     *
+     * These all only count connections that completed the TLS handshake successfully. This means
+     * that short lived 0RTT connections are potentially not included in these counts.
      */
-    /// The number of direct connections we have made to peers.
-    pub num_direct_conns_added: Counter,
-    /// The number of direct connections we have lost to peers.
-    pub num_direct_conns_removed: Counter,
-    /// The number of connections to peers we have added over relay.
-    pub num_relay_conns_added: Counter,
-    /// The number of connections to peers we have removed over relay.
-    pub num_relay_conns_removed: Counter,
+    /// Number of connections opened (only handshaked connections are counted).
+    pub num_conns_opened: Counter,
+    /// Number of connections closed (only handshaked connections are counted).
+    pub num_conns_closed: Counter,
+    /// Number of connections that had only relay paths over their lifetime.
+    pub num_conns_transport_relay_only: Counter,
+    /// Number of connections that had only IP paths over their lifetime.
+    pub num_conns_transport_ip_only: Counter,
+    /// Number of connections that had both IP and relay paths.
+    pub num_conns_transport_ip_and_relay: Counter,
 
     pub actor_tick_main: Counter,
     pub actor_tick_msg: Counter,
@@ -68,28 +73,19 @@ pub struct Metrics {
     pub actor_link_change: Counter,
     pub actor_tick_other: Counter,
 
-    /// Number of endpoints we have attempted to contact.
-    pub endpoints_contacted: Counter,
-    /// Number of endpoints we have managed to contact directly.
-    pub endpoints_contacted_directly: Counter,
-
-    /// Number of connections with a successful handshake.
-    pub connection_handshake_success: Counter,
-    /// Number of connections with a successful handshake that became direct.
-    pub connection_became_direct: Counter,
     /// Histogram of connection latency in milliseconds across all endpoint connections.
     #[default(Histogram::new(vec![1.0, 5.0, 10.0, 25.0, 50.0, 100.0, 250.0, 500.0, 1000.0, f64::INFINITY]))]
     pub connection_latency_ms: Histogram,
 
     /*
      * Path Congestion Metrics
      */
-    /// Number of times a path was marked as outdated due to consecutive ping failures.
-    pub path_marked_outdated: Counter,
-    /// Number of ping failures recorded across all paths.
-    pub path_ping_failures: Counter,
-    /// Number of consecutive failure resets (path recovered).
-    pub path_failure_resets: Counter,
+    // /// Number of times a path was marked as outdated due to consecutive ping failures.
+    // pub path_marked_outdated: Counter,
+    // /// Number of ping failures recorded across all paths.
+    // pub path_ping_failures: Counter,
+    // /// Number of consecutive failure resets (path recovered).
+    // pub path_failure_resets: Counter,
     /// Histogram of packet loss rates (0.0-1.0) observed on UDP paths.
     #[default(Histogram::new(vec![0.0, 0.01, 0.05, 0.1, 0.2, 0.5, 1.0]))]
     pub path_packet_loss_rate: Histogram,
