feat: Collect metrics for direct connections & add opt-in push metrics (#2805)

## New metrics

1. `nodes_contacted`: Number of nodes we have attempted to contact.
2. `nodes_contacted_directly`: Number of nodes we have managed to
contact directly.
3. `connection_handshake_success`: Number of connections with a
successful handshake.
4. `connection_became_direct`: Number of connections that became direct.

This sets up to compute some core metrics regarding direct connections:
1. `nodes_contacted_directly` / `nodes_contacted`: Ratio of node pairs
that manage to connect directly.
2. `connection_became_direct` / `connection_handshake_success`: Ratio of
connections that can communicate directly

Both of these metrics give us an idea on the number of network
conditions that allow us to avoid relay traffic.
Our hypothesis is that this number is around 90% & that it's similar
between these two metrics, but it's worth capturing both to see if we're
right.

We also already capture enough information to get numbers on direct vs.
relayed traffic volume:

`recv_data_relay` / (`recv_data_ipv4` + `recv_data_ipv6`): Ratio of
relay data vs. direct data.

This metric will help us explain how much bandwidth you can save by
using iroh-net.

## ~~Opt-in Push Metrics Exporter~~

Initially this PR added some config to the iroh-cli to allow configuring
a push metrics exporter.
I removed this after discussions in the RFC. It's not the ideal way to
capture metrics except for internal use.

We still retain the changes in `iroh_metrics`, to make use of a push
metrics exporter for internal use, though.

## Breaking Changes

- `MagicsockMetrics` is now `#[non_exhaustive]`. This allows us to add
more metrics without breaking backwards compatibility in the future. The
struct is not meant to be constructed outside of `iroh-net` anyways.

## Change checklist

- [x] Self-review.
- [x] Documentation updates following the [style
guide](https://rust-lang.github.io/rfcs/1574-more-api-documentation-conventions.html#appendix-a-full-conventions-text),
if relevant.
- [x] Tests if relevant.
(We tested this manually at the retreat at least. We don't have metrics
tests at the moment.)
- [x] All breaking changes documented.

---------

Co-authored-by: Asmir Avdicevic <asmir.avdicevic64@gmail.com>
Co-authored-by: Ruediger Klaehn <rklaehn@protonmail.com>
Co-authored-by: Diva M <divma@protonmail.com>
Co-authored-by: Divma <26765164+divagant-martian@users.noreply.github.com>
Co-authored-by: Floris Bruynooghe <flub@n0.computer>

Author: 5 people

iroh-metrics/src/lib.rs

@@ -76,3 +76,33 @@ pub fn parse_prometheus_metrics(data: &str) -> anyhow::Result<HashMap<String, f6
     }
     Ok(metrics)
 }
+
+/// Configuration for pushing metrics to a remote endpoint.
+#[derive(PartialEq, Eq, Debug, Default, serde::Deserialize, Clone)]
+pub struct PushMetricsConfig {
+    /// The push interval in seconds.
+    pub interval: u64,
+    /// The endpoint url for the push metrics collector.
+    pub endpoint: String,
+    /// The name of the service you're exporting metrics for.
+    ///
+    /// Generally, `metrics_exporter` is good enough for use
+    /// outside of production deployments.
+    pub service_name: String,
+    /// The name of the instance you're exporting metrics for.
+    ///
+    /// This should be device-unique. If not, this will sum up
+    /// metrics from different devices.
+    ///
+    /// E.g. `username-laptop`, `username-phone`, etc.
+    ///
+    /// Another potential scheme with good privacy would be a
+    /// keyed blake3 hash of the secret key. (This gives you
+    /// an identifier that is as unique as a `NodeID`, but
+    /// can't be correlated to `NodeID`s.)
+    pub instance_name: String,
+    /// The username for basic auth for the push metrics collector.
+    pub username: Option<String>,
+    /// The password for basic auth for the push metrics collector.
+    pub password: String,
+}

iroh-metrics/src/metrics.rs

@@ -62,3 +62,17 @@ pub async fn start_metrics_dumper(
 ) -> anyhow::Result<()> {
     crate::service::dumper(&path, interval).await
 }
+
+/// Start a metrics exporter service.
+#[cfg(feature = "metrics")]
+pub async fn start_metrics_exporter(cfg: crate::PushMetricsConfig) {
+    crate::service::exporter(
+        cfg.endpoint,
+        cfg.service_name,
+        cfg.instance_name,
+        cfg.username,
+        cfg.password,
+        std::time::Duration::from_secs(cfg.interval),
+    )
+    .await;
+}

iroh-metrics/src/service.rs

@@ -7,7 +7,7 @@ use std::{
 use anyhow::{anyhow, Context, Result};
 use hyper::{service::service_fn, Request, Response};
 use tokio::{io::AsyncWriteExt as _, net::TcpListener};
-use tracing::{error, info};
+use tracing::{debug, error, info, warn};
 
 use crate::{core::Core, parse_prometheus_metrics};
 
@@ -116,3 +116,53 @@ async fn dump_metrics(
     }
     Ok(())
 }
+
+/// Export metrics to a push gateway.
+pub async fn exporter(
+    gateway_endpoint: String,
+    service_name: String,
+    instance_name: String,
+    username: Option<String>,
+    password: String,
+    interval: Duration,
+) {
+    let Some(core) = Core::get() else {
+        error!("metrics disabled");
+        return;
+    };
+    let push_client = reqwest::Client::new();
+    let prom_gateway_uri = format!(
+        "{}/metrics/job/{}/instance/{}",
+        gateway_endpoint, service_name, instance_name
+    );
+    loop {
+        tokio::time::sleep(interval).await;
+        let buff = core.encode();
+        match buff {
+            Err(e) => error!("Failed to encode metrics: {e:#}"),
+            Ok(buff) => {
+                let mut req = push_client.post(&prom_gateway_uri);
+                if let Some(username) = username.clone() {
+                    req = req.basic_auth(username, Some(password.clone()));
+                }
+                let res = match req.body(buff).send().await {
+                    Ok(res) => res,
+                    Err(e) => {
+                        warn!("failed to push metrics: {}", e);
+                        continue;
+                    }
+                };
+                match res.status() {
+                    reqwest::StatusCode::OK => {
+                        debug!("pushed metrics to gateway");
+                    }
+                    _ => {
+                        warn!("failed to push metrics to gateway: {:?}", res);
+                        let body = res.text().await.unwrap();
+                        warn!("error body: {}", body);
+                    }
+                }
+            }
+        }
+    }
+}

iroh-net/src/endpoint/rtt_actor.rs

@@ -5,14 +5,18 @@ use std::collections::HashMap;
 use futures_concurrency::stream::stream_group;
 use futures_lite::StreamExt;
 use iroh_base::key::NodeId;
+use iroh_metrics::inc;
 use tokio::{
     sync::{mpsc, Notify},
     task::JoinHandle,
     time::Duration,
 };
 use tracing::{debug, error, info_span, trace, warn, Instrument};
 
-use crate::magicsock::{ConnectionType, ConnectionTypeStream};
+use crate::{
+    magicsock::{ConnectionType, ConnectionTypeStream},
+    metrics::MagicsockMetrics,
+};
 
 #[derive(Debug)]
 pub(super) struct RttHandle {
@@ -65,7 +69,9 @@ struct RttActor {
     ///
     /// These are weak references so not to keep the connections alive.  The key allows
     /// removing the corresponding stream from `conn_type_changes`.
-    connections: HashMap<stream_group::Key, (quinn::WeakConnectionHandle, NodeId)>,
+    /// The boolean is an indiciator of whether this connection was direct before.
+    /// This helps establish metrics on number of connections that became direct.
+    connections: HashMap<stream_group::Key, (quinn::WeakConnectionHandle, NodeId, bool)>,
     /// A way to notify the main actor loop to run over.
     ///
     /// E.g. when a new stream was added.
@@ -113,8 +119,9 @@ impl RttActor {
         node_id: NodeId,
     ) {
         let key = self.connection_events.insert(conn_type_changes);
-        self.connections.insert(key, (connection, node_id));
+        self.connections.insert(key, (connection, node_id, false));
         self.tick.notify_one();
+        inc!(MagicsockMetrics, connection_handshake_success);
     }
 
     /// Performs the congestion controller reset for a magic socket path change.
@@ -125,14 +132,19 @@ impl RttActor {
     /// happens commonly.
     fn do_reset_rtt(&mut self, item: Option<(stream_group::Key, ConnectionType)>) {
         match item {
-            Some((key, new_conn_type)) => match self.connections.get(&key) {
-                Some((handle, node_id)) => {
+            Some((key, new_conn_type)) => match self.connections.get_mut(&key) {
+                Some((handle, node_id, was_direct_before)) => {
                     if handle.reset_congestion_state() {
                         debug!(
                             node_id = %node_id.fmt_short(),
                             new_type = ?new_conn_type,
                             "Congestion controller state reset",
                         );
+                        if !*was_direct_before && matches!(new_conn_type, ConnectionType::Direct(_))
+                        {
+                            *was_direct_before = true;
+                            inc!(MagicsockMetrics, connection_became_direct);
+                        }
                     } else {
                         debug!(
                             node_id = %node_id.fmt_short(),
@@ -151,7 +163,7 @@ impl RttActor {
 
     /// Performs cleanup for closed connection.
     fn do_connections_cleanup(&mut self) {
-        for (key, (handle, node_id)) in self.connections.iter() {
+        for (key, (handle, node_id, _)) in self.connections.iter() {
             if !handle.is_alive() {
                 trace!(node_id = %node_id.fmt_short(), "removing stale connection");
                 self.connection_events.remove(*key);

iroh-net/src/magicsock/metrics.rs

@@ -6,6 +6,7 @@ use iroh_metrics::{
 /// Enum of metrics for the module
 #[allow(missing_docs)]
 #[derive(Debug, Clone, Iterable)]
+#[non_exhaustive]
 pub struct Metrics {
     pub re_stun_calls: Counter,
     pub update_direct_addrs: Counter,
@@ -68,6 +69,16 @@ pub struct Metrics {
     pub actor_tick_direct_addr_update_receiver: Counter,
     pub actor_link_change: Counter,
     pub actor_tick_other: Counter,
+
+    /// Number of nodes we have attempted to contact.
+    pub nodes_contacted: Counter,
+    /// Number of nodes we have managed to contact directly.
+    pub nodes_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,
 }
 
 impl Default for Metrics {
@@ -132,6 +143,12 @@ impl Default for Metrics {
             ),
             actor_link_change: Counter::new("actor_link_change"),
             actor_tick_other: Counter::new("actor_tick_other"),
+
+            nodes_contacted: Counter::new("nodes_contacted"),
+            nodes_contacted_directly: Counter::new("nodes_contacted_directly"),
+
+            connection_handshake_success: Counter::new("connection_handshake_success"),
+            connection_became_direct: Counter::new("connection_became_direct"),
         }
     }
 }

iroh-net/src/magicsock/node_map/node_state.rs

@@ -134,6 +134,10 @@ pub(super) struct NodeState {
     last_call_me_maybe: Option<Instant>,
     /// The type of connection we have to the node, either direct, relay, mixed, or none.
     conn_type: Watchable<ConnectionType>,
+    /// Whether the conn_type was ever observed to be `Direct` at some point.
+    ///
+    /// Used for metric reporting.
+    has_been_direct: bool,
 }
 
 /// Options for creating a new [`NodeState`].
@@ -173,6 +177,7 @@ impl NodeState {
             last_used: options.active.then(Instant::now),
             last_call_me_maybe: None,
             conn_type: Watchable::new(ConnectionType::None),
+            has_been_direct: false,
         }
     }
 
@@ -304,6 +309,10 @@ impl NodeState {
             (None, Some(relay_url)) => ConnectionType::Relay(relay_url),
             (None, None) => ConnectionType::None,
         };
+        if !self.has_been_direct && matches!(&typ, ConnectionType::Direct(_)) {
+            self.has_been_direct = true;
+            inc!(MagicsockMetrics, nodes_contacted_directly);
+        }
         if let Ok(prev_typ) = self.conn_type.update(typ.clone()) {
             // The connection type has changed.
             event!(
@@ -1132,7 +1141,11 @@ impl NodeState {
         have_ipv6: bool,
     ) -> (Option<SocketAddr>, Option<RelayUrl>, Vec<PingAction>) {
         let now = Instant::now();
-        self.last_used.replace(now);
+        let prev = self.last_used.replace(now);
+        if prev.is_none() {
+            // this is the first time we are trying to connect to this node
+            inc!(MagicsockMetrics, nodes_contacted);
+        }
         let (udp_addr, relay_url) = self.addr_for_send(&now, have_ipv6);
         let mut ping_msgs = Vec::new();
 
@@ -1485,6 +1498,7 @@ mod tests {
                     last_used: Some(now),
                     last_call_me_maybe: None,
                     conn_type: Watchable::new(ConnectionType::Direct(ip_port.into())),
+                    has_been_direct: true,
                 },
                 ip_port.into(),
             )
@@ -1504,6 +1518,7 @@ mod tests {
                 last_used: Some(now),
                 last_call_me_maybe: None,
                 conn_type: Watchable::new(ConnectionType::Relay(send_addr.clone())),
+                has_been_direct: false,
             }
         };
 
@@ -1530,6 +1545,7 @@ mod tests {
                 last_used: Some(now),
                 last_call_me_maybe: None,
                 conn_type: Watchable::new(ConnectionType::Relay(send_addr.clone())),
+                has_been_direct: false,
             }
         };
 
@@ -1569,6 +1585,7 @@ mod tests {
                         socket_addr,
                         send_addr.clone(),
                     )),
+                    has_been_direct: false,
                 },
                 socket_addr,
             )