Export network connectivity metrics for each participant

node/src/metrics.rs

@@ -166,3 +166,13 @@ lazy_static! {
         )
         .unwrap();
 }
+
+lazy_static! {
+    pub static ref NETWORK_LIVE_CONNECTIONS: prometheus::IntGaugeVec =
+        prometheus::register_int_gauge_vec!(
+            "sign_request_channel_failed",
+            "failed to send on channel in sign_request_channel",
+            &["source_participant_id", "target_participant_id"],
+        )
+        .unwrap();
+}

node/src/mpc_client.rs

@@ -75,7 +75,19 @@ impl MpcClient {
             let presignature_store = self.presignature_store.clone();
             let sign_request_store = self.sign_request_store.clone();
             let root_keyshare = self.root_keyshare.clone();
+
+            let client_metrics = client.clone();
+            let auto_abort = tracking::spawn("periodically emits metrics", async move {
+                loop {
+                    client_metrics.emit_metrics();
+
+                    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+                }
+            });
+
+            // Task for executing the main MPC loop
             tracking::spawn("monitor passive channels", async move {
+                let _auto_abort_metrics = auto_abort;
                 let mut tasks = AutoAbortTaskCollection::new();
                 loop {
                     let channel = channel_receiver.recv().await.unwrap();

node/src/network.rs

@@ -44,6 +44,8 @@ pub trait MeshNetworkTransportSender: Send + Sync + 'static {
     /// Returns the participant IDs of all nodes in the network that are currently alive.
     /// This is a subset of all_participant_ids, and includes our own participant ID.
     fn all_alive_participant_ids(&self) -> Vec<ParticipantId>;
+    /// Emits prometheus metrics regarding the state of connections to other MPC nodes
+    fn emit_metrics(&self);
 }
 
 /// The receiving side of the networking layer. It is expected that the node will run
@@ -149,6 +151,11 @@ impl MeshNetworkClient {
             }
         }
     }
+
+    /// Emit network metrics through Prometheus counters
+    pub fn emit_metrics(&self) {
+        self.transport_sender.emit_metrics();
+    }
 }
 
 enum SenderOrNewChannel {
@@ -407,6 +414,10 @@ pub mod testing {
         fn all_alive_participant_ids(&self) -> Vec<ParticipantId> {
             self.all_participant_ids()
         }
+
+        fn emit_metrics(&self) {
+            panic!("emit_metrics should not be called");
+        }
     }
 
     #[async_trait::async_trait]

node/src/p2p.rs

@@ -1,4 +1,5 @@
 use crate::config::MpcConfig;
+use crate::metrics;
 use crate::network::{MeshNetworkTransportReceiver, MeshNetworkTransportSender};
 use crate::primitives::{MpcMessage, MpcPeerMessage, ParticipantId};
 use crate::tracking::{self, AutoAbortTask, AutoAbortTaskCollection};
@@ -8,7 +9,7 @@ use borsh::BorshDeserialize;
 use rustls::pki_types::{CertificateDer, PrivatePkcs8KeyDer};
 use rustls::server::danger::ClientCertVerifier;
 use rustls::{ClientConfig, CommonState, ServerConfig};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Weak};
@@ -738,6 +739,21 @@ impl MeshNetworkTransportSender for TlsMeshSender {
         ids.sort();
         ids
     }
+
+    fn emit_metrics(&self) {
+        let my_participant_id = self.my_participant_id();
+        let live_participants: HashSet<ParticipantId> =
+            self.all_alive_participant_ids().into_iter().collect();
+
+        metrics::NETWORK_LIVE_CONNECTIONS.reset();
+
+        for id in self.all_participant_ids() {
+            let is_live_participant = live_participants.contains(&id);
+            metrics::NETWORK_LIVE_CONNECTIONS
+                .with_label_values(&[&my_participant_id.to_string(), &id.to_string()])
+                .set(is_live_participant.into());
+        }
+    }
 }
 
 #[async_trait]