feat: metrics refactor

Author: Frando

Cargo.lock

@@ -40,3 +40,7 @@ unexpected_cfgs = { level = "warn", check-cfg = ["cfg(iroh_docsrs)", "cfg(iroh_l
 
 [workspace.lints.clippy]
 unused-async = "warn"
+
+[patch.crates-io]
+iroh-metrics = { path = "../iroh-metrics" }
+portmapper = { path = "../net-tools/portmapper" }

Cargo.toml

@@ -1,6 +1,6 @@
 //! Metrics support for the server
 
-use iroh_metrics::core::{Core, Counter, Metric};
+use iroh_metrics::core::{Core, Counter, Metric, MetricExt};
 use struct_iterable::Iterable;
 
 /// Metrics for iroh-dns-server
@@ -51,7 +51,7 @@ impl Default for Metrics {
 }
 
 impl Metric for Metrics {
-    fn name() -> &'static str {
+    fn name(&self) -> &'static str {
         "dns_server"
     }
 }

iroh-dns-server/src/metrics.rs

@@ -724,54 +724,6 @@ async fn build_relay_config(cfg: Config) -> Result<relay::ServerConfig<std::io::
     })
 }
 
-mod metrics {
-    use iroh_metrics::{
-        core::{Counter, Metric},
-        struct_iterable::Iterable,
-    };
-
-    /// StunMetrics tracked for the relay server
-    #[allow(missing_docs)]
-    #[derive(Debug, Clone, Iterable)]
-    pub struct StunMetrics {
-        /*
-         * Metrics about STUN requests over ipv6
-         */
-        /// Number of stun requests made
-        pub requests: Counter,
-        /// Number of successful requests over ipv4
-        pub ipv4_success: Counter,
-        /// Number of successful requests over ipv6
-        pub ipv6_success: Counter,
-
-        /// Number of bad requests, either non-stun packets or incorrect binding request
-        pub bad_requests: Counter,
-        /// Number of failures
-        pub failures: Counter,
-    }
-
-    impl Default for StunMetrics {
-        fn default() -> Self {
-            Self {
-                /*
-                 * Metrics about STUN requests
-                 */
-                requests: Counter::new("Number of STUN requests made to the server."),
-                ipv4_success: Counter::new("Number of successful ipv4 STUN requests served."),
-                ipv6_success: Counter::new("Number of successful ipv6 STUN requests served."),
-                bad_requests: Counter::new("Number of bad requests made to the STUN endpoint."),
-                failures: Counter::new("Number of STUN requests that end in failure."),
-            }
-        }
-    }
-
-    impl Metric for StunMetrics {
-        fn name() -> &'static str {
-            "stun"
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use std::num::NonZeroU32;

iroh-relay/src/main.rs

@@ -283,7 +283,7 @@ impl Server {
         #[cfg(feature = "metrics")]
         if let Some(addr) = config.metrics_addr {
             debug!("Starting metrics server");
-            use iroh_metrics::core::Metric;
+            use iroh_metrics::core::MetricExt;
 
             iroh_metrics::core::Core::init(|reg, metrics| {
                 metrics.insert(metrics::Metrics::new(reg));

iroh-relay/src/server.rs

@@ -123,7 +123,7 @@ impl Default for Metrics {
 }
 
 impl Metric for Metrics {
-    fn name() -> &'static str {
+    fn name(&self) -> &'static str {
         "relayserver"
     }
 }
@@ -163,7 +163,7 @@ impl Default for StunMetrics {
 }
 
 impl Metric for StunMetrics {
-    fn name() -> &'static str {
+    fn name(&self) -> &'static str {
         "stun"
     }
 }

iroh-relay/src/server/metrics.rs

@@ -5,6 +5,7 @@ use clap::Parser;
 #[cfg(not(any(target_os = "freebsd", target_os = "openbsd", target_os = "netbsd")))]
 use iroh_bench::quinn;
 use iroh_bench::{configure_tracing_subscriber, iroh, rt, s2n, Commands, Opt};
+use iroh_metrics::core::MetricExt;
 
 fn main() {
     let cmd = Commands::parse();
@@ -34,7 +35,6 @@ pub fn run_iroh(opt: Opt) -> Result<()> {
     if opt.metrics {
         // enable recording metrics
         iroh_metrics::core::Core::try_init(|reg, metrics| {
-            use iroh_metrics::core::Metric;
             metrics.insert(::iroh::metrics::MagicsockMetrics::new(reg));
             metrics.insert(::iroh::metrics::NetReportMetrics::new(reg));
             metrics.insert(::iroh::metrics::PortmapMetrics::new(reg));

iroh/bench/src/bin/bulk.rs

@@ -41,6 +41,7 @@ use crate::{
         DiscoveryTask, Lagged, UserData,
     },
     magicsock::{self, Handle, NodeIdMappedAddr},
+    metrics::EndpointMetrics,
     tls,
     watchable::Watcher,
     RelayProtocol,
@@ -192,6 +193,8 @@ impl Builder {
         };
         let server_config = static_config.create_server_config(self.alpn_protocols)?;
 
+        let metrics = EndpointMetrics::default();
+
         let msock_opts = magicsock::Options {
             addr_v4: self.addr_v4,
             addr_v6: self.addr_v6,
@@ -209,6 +212,7 @@ impl Builder {
             insecure_skip_relay_cert_verify: self.insecure_skip_relay_cert_verify,
             #[cfg(any(test, feature = "test-utils"))]
             path_selection: self.path_selection,
+            metrics,
         };
         Endpoint::bind(static_config, msock_opts).await
     }
@@ -620,7 +624,7 @@ impl Endpoint {
 
         let ep = Self {
             msock: msock.clone(),
-            rtt_actor: Arc::new(rtt_actor::RttHandle::new()),
+            rtt_actor: Arc::new(rtt_actor::RttHandle::new(msock.metrics.magicsock.clone())),
             static_config: Arc::new(static_config),
             session_store: Arc::new(rustls::client::ClientSessionMemoryCache::new(
                 MAX_TLS_TICKETS,
@@ -1103,6 +1107,12 @@ impl Endpoint {
         self.msock.discovery()
     }
 
+    /// Returns metrics collected for this endpoint.
+    #[cfg(feature = "metrics")]
+    pub fn metrics(&self) -> &EndpointMetrics {
+        &self.msock.metrics
+    }
+
     // # Methods for less common state updates.
 
     /// Notifies the system of potential network changes.

iroh/src/endpoint.rs

@@ -3,7 +3,6 @@
 use std::{pin::Pin, task::Poll};
 
 use iroh_base::NodeId;
-use iroh_metrics::inc;
 use n0_future::{
     task::{self, AbortOnDropHandle},
     MergeUnbounded, Stream, StreamExt,
@@ -21,9 +20,10 @@ pub(super) struct RttHandle {
 }
 
 impl RttHandle {
-    pub(super) fn new() -> Self {
+    pub(super) fn new(metrics: MagicsockMetrics) -> Self {
         let mut actor = RttActor {
             connection_events: Default::default(),
+            metrics,
         };
         let (msg_tx, msg_rx) = mpsc::channel(16);
         let handle = task::spawn(
@@ -62,6 +62,7 @@ struct RttActor {
     /// Stream of connection type changes.
     #[debug("MergeUnbounded<WatcherStream<ConnectionType>>")]
     connection_events: MergeUnbounded<MappedStream>,
+    metrics: MagicsockMetrics,
 }
 
 #[derive(Debug)]
@@ -75,8 +76,12 @@ struct MappedStream {
     was_direct_before: bool,
 }
 
+struct ConnectionEvent {
+    became_direct: bool,
+}
+
 impl Stream for MappedStream {
-    type Item = ConnectionType;
+    type Item = ConnectionEvent;
 
     /// Performs the congestion controller reset for a magic socket path change.
     ///
@@ -90,6 +95,7 @@ impl Stream for MappedStream {
     ) -> Poll<Option<Self::Item>> {
         match Pin::new(&mut self.stream).poll_next(cx) {
             Poll::Ready(Some(new_conn_type)) => {
+                let mut became_direct = false;
                 if self.connection.network_path_changed() {
                     debug!(
                         node_id = %self.node_id.fmt_short(),
@@ -99,10 +105,10 @@ impl Stream for MappedStream {
                     if !self.was_direct_before && matches!(new_conn_type, ConnectionType::Direct(_))
                     {
                         self.was_direct_before = true;
-                        inc!(MagicsockMetrics, connection_became_direct);
+                        became_direct = true
                     }
-                }
-                Poll::Ready(Some(new_conn_type))
+                };
+                Poll::Ready(Some(ConnectionEvent { became_direct }))
             }
             Poll::Ready(None) => Poll::Ready(None),
             Poll::Pending => Poll::Pending,
@@ -124,7 +130,11 @@ impl RttActor {
                         None => break,
                     }
                 }
-                _item = self.connection_events.next(), if !self.connection_events.is_empty() => {}
+                event = self.connection_events.next(), if !self.connection_events.is_empty() => {
+                    if event.map(|e| e.became_direct).unwrap_or(false) {
+                        self.metrics.connection_became_direct.inc();
+                    }
+                }
             }
         }
         debug!("rtt-actor finished");
@@ -156,6 +166,6 @@ impl RttActor {
             node_id,
             was_direct_before: false,
         });
-        inc!(MagicsockMetrics, connection_handshake_success);
+        self.metrics.connection_handshake_success.inc();
     }
 }