feat: add back path congestion metrics

Author: Frando

iroh/src/magicsock/metrics.rs

@@ -1,4 +1,4 @@
-use iroh_metrics::{Counter, MetricsGroup};
+use iroh_metrics::{Counter, Histogram, MetricsGroup};
 use serde::{Deserialize, Serialize};
 
 /// Enum of metrics for the module
@@ -14,8 +14,11 @@ pub struct Metrics {
     pub send_ipv4: Counter,
     pub send_ipv6: Counter,
     pub send_relay: Counter,
+    pub send_relay_error: Counter,
 
     // Data packets (non-disco)
+    pub send_data: Counter,
+    pub send_data_network_down: Counter,
     pub recv_data_relay: Counter,
     pub recv_data_ipv4: Counter,
     pub recv_data_ipv6: Counter,
@@ -69,25 +72,27 @@ pub struct Metrics {
     pub actor_tick_direct_addr_heartbeat: Counter,
     pub actor_link_change: Counter,
     pub actor_tick_other: 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
-    // */
+
+    /// 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,
-    // /// 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,
-    // /// Histogram of RTT variance (in milliseconds) as a congestion indicator.
-    // #[default(Histogram::new(vec![0.0, 1.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0]))]
-    // pub path_rtt_variance_ms: Histogram,
-    // /// Histogram of path quality scores (0.0-1.0).
-    // #[default(Histogram::new(vec![0.0, 0.3, 0.5, 0.7, 0.85, 0.95, 1.0]))]
-    // pub path_quality_score: Histogram,
+    /// 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,
+    /// Histogram of RTT variance (in milliseconds) as a congestion indicator.
+    #[default(Histogram::new(vec![0.0, 1.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0]))]
+    pub path_rtt_variance_ms: Histogram,
+    /// Histogram of path quality scores (0.0-1.0).
+    #[default(Histogram::new(vec![0.0, 0.3, 0.5, 0.7, 0.85, 0.95, 1.0]))]
+    pub path_quality_score: Histogram,
 }

iroh/src/magicsock/remote_map/remote_state.rs

@@ -45,6 +45,9 @@ use crate::{
 mod guarded_channel;
 mod path_state;
 
+#[cfg(feature = "metrics")]
+mod metrics;
+
 // TODO: use this
 // /// Number of addresses that are not active that we keep around per endpoint.
 // ///
@@ -83,6 +86,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
@@ -240,8 +246,12 @@ impl RemoteStateActor {
     /// discipline is needed to not turn pending for a long time.
     async fn run(mut self, mut inbox: GuardedReceiver<RemoteStateMessage>) {
         trace!("actor started");
+
         let idle_timeout = time::sleep(ACTOR_MAX_IDLE_TIMEOUT);
         n0_future::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)),
@@ -296,6 +306,9 @@ impl RemoteStateActor {
                 item = self.discovery_stream.next() => {
                     self.handle_discovery_item(item);
                 }
+                _ = 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");
@@ -430,7 +443,8 @@ impl RemoteStateActor {
                     paths: Default::default(),
                     open_paths: Default::default(),
                     path_ids: Default::default(),
-                    transport_summary: TransportSummary::default(),
+                    #[cfg(feature = "metrics")]
+                    metrics: Default::default(),
                 })
                 .into_mut();
 
@@ -589,8 +603,10 @@ impl RemoteStateActor {
 
     fn handle_connection_close(&mut self, conn_id: ConnId) {
         if let Some(state) = self.connections.remove(&conn_id) {
-            self.metrics.num_conns_closed.inc();
-            state.transport_summary.record(&self.metrics);
+            #[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");
@@ -1066,6 +1082,13 @@ impl RemoteStateActor {
             }
         }
     }
+
+    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 [`RemoteStateActor`].
@@ -1174,8 +1197,11 @@ struct ConnectionState {
     open_paths: FxHashMap<PathId, transports::Addr>,
     /// Reverse map of [`Self::paths].
     path_ids: FxHashMap<transports::Addr, PathId>,
-    /// Summary over transports used in this connection, for metrics tracking.
-    transport_summary: TransportSummary,
+    /// 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 {
@@ -1187,7 +1213,8 @@ impl ConnectionState {
 
     /// Tracks an open path for the connection.
     fn add_open_path(&mut self, remote: transports::Addr, path_id: PathId) {
-        self.transport_summary.add_path(&remote);
+        #[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);
@@ -1200,11 +1227,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();
     }
@@ -1224,6 +1255,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.
@@ -1435,44 +1479,6 @@ impl Future for OnClosed {
     }
 }
 
-/// 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,
-        }
-    }
-
-    fn record(&self, metrics: &MagicsockMetrics) {
-        match self {
-            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 => {}
-        }
-    }
-}
-
 /// Converts an iterator of [`TransportAddr'] into an iterator of [`transports::Addr`].
 fn to_transports_addr(
     endpoint_id: EndpointId,