feat(threshold-signer): add metrics & extra libp2p logs (#27)

* feat(dcipher-agents/libp2p): add connection closed & topic unsubscribe logs

* feat(dcipher-agents/threshold_signer): add metrics for connections

* feat(dcipher-agents/threshold_signer): add metrics for partials

* feat(blocklock-agent & randomness-agent): expose threshold signer metrics

* feat(threshold-signer): add missing metrics module

Author: azixus

blocklock-agent/examples/blocklock/healthcheck.rs

@@ -1,5 +1,6 @@
 use anyhow::anyhow;
-use dcipher_agents::agents::blocklock::metrics::Metrics;
+use dcipher_agents::agents::blocklock::metrics::Metrics as BlocklockMetrics;
+use dcipher_agents::signer::threshold_signer::metrics::Metrics as ThresholdSignerMetrics;
 use prometheus::{Encoder, TextEncoder};
 use std::net::IpAddr;
 use warp::Filter;
@@ -12,7 +13,7 @@ pub async fn start_api(listen_addr: IpAddr, port: u16) -> anyhow::Result<()> {
 
     let metrics = warp::path!("metrics").map(|| {
         let encoder = TextEncoder::new();
-        let metrics = Metrics::gather();
+        let metrics = [BlocklockMetrics::gather(), ThresholdSignerMetrics::gather()].concat();
         let mut buffer = Vec::new();
 
         match encoder.encode(&metrics, &mut buffer) {

dcipher-agents/src/signer/threshold_signer.rs

@@ -3,11 +3,13 @@
 
 mod aggregation;
 mod libp2p;
+pub mod metrics;
 
 pub use aggregation::lagrange_points_interpolate_at;
 
 use crate::ser::EvmSerialize;
 use crate::signer::threshold_signer::libp2p::LibP2PNode;
+use crate::signer::threshold_signer::metrics::Metrics;
 use crate::signer::{AsynchronousSigner, BlsSigner, BlsVerifier};
 use ark_ec::{AffineRepr, CurveGroup};
 use itertools::Either;
@@ -355,6 +357,7 @@ where
 
                             let m = serde_cbor::to_vec(&partial)
                                 .expect("serialization should always work");
+                            Metrics::report_partials_sent(1);
                             if tx_to_libp2p.send(m).is_err() {
                                 tracing::error!("Failed to send message to signer: channel closed");
                             }
@@ -398,13 +401,16 @@ where
                         }
                     };
 
+                Metrics::report_partials_received(1);
+
                 // Verify the validity of the partial signature against its pk
                 let Some(pk) = self.pks.get(usize::from(party_id) - 1) else {
                     tracing::error!(sender_id = party_id, "Invalid party_id / pks vector");
                     continue;
                 };
                 if !self.signer.verify(&partial.m, partial.sig, *pk) {
                     tracing::error!(sender_id = party_id, "Received invalid partial signature");
+                    Metrics::report_invalid_partials(1);
                     continue;
                 }
 

dcipher-agents/src/signer/threshold_signer/libp2p.rs

@@ -1,6 +1,7 @@
 //! Libp2p node that can be used to broadcast and receive arbitrary messages using floodsub and a
 //! peer whitelist.
 
+use crate::signer::threshold_signer::metrics::Metrics;
 use futures_util::StreamExt;
 use libp2p::allow_block_list::AllowedPeers;
 use libp2p::floodsub::{FloodsubEvent, FloodsubMessage};
@@ -10,6 +11,7 @@ use libp2p::{
     Multiaddr, PeerId, Swarm, allow_block_list, floodsub, noise, swarm::SwarmEvent, tcp, yamux,
 };
 use std::collections::HashMap;
+use std::num::NonZeroU32;
 use std::time::Duration;
 use thiserror::Error;
 use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
@@ -223,14 +225,40 @@ impl LibP2PNode {
                 peer_id,
                 topic,
             })) => {
-                tracing::debug!(%peer_id, ?topic, "Peer listening to topic");
+                let short_id = peer_id_to_short_id.get(&peer_id).or_else(|| {
+                    tracing::error!(
+                        incoming_peer_id = %peer_id,
+                        "Failed to convert peer_id to short_id"
+                    );
+                    None
+                });
+
+                tracing::info!(%peer_id, ?short_id, ?topic, "Peer subscribed to topic");
                 // Once we've received at least one topic subscription from a remote peer, we should
                 // be able to send messages.
                 *ready_send_messages = true;
             }
 
+            SwarmEvent::Behaviour(BehaviourEvent::Floodsub(FloodsubEvent::Unsubscribed {
+                peer_id,
+                topic,
+            })) => {
+                let short_id = peer_id_to_short_id.get(&peer_id).or_else(|| {
+                    tracing::error!(
+                        incoming_peer_id = %peer_id,
+                        "Failed to convert peer_id to short_id"
+                    );
+                    None
+                });
+
+                tracing::info!(%peer_id, ?short_id, ?topic, "Peer unsubscribed to topic");
+            }
+
             SwarmEvent::ConnectionEstablished {
-                peer_id, endpoint, ..
+                peer_id,
+                endpoint,
+                num_established,
+                ..
             } => {
                 let short_id = peer_id_to_short_id.get(&peer_id).or_else(|| {
                     tracing::error!(
@@ -241,14 +269,51 @@ impl LibP2PNode {
                     None
                 });
 
+                if num_established == const { NonZeroU32::new(1).unwrap() } {
+                    // First connection established, report new peer connected
+                    Metrics::report_peer_connected();
+                }
+
                 tracing::info!(
                     incoming_peer_id = %peer_id,
                     incoming_short_id = ?short_id,
                     incoming_remote_addr = %endpoint.get_remote_address(),
+                    num_established,
                     "Libp2p node established connection with peer"
                 );
             }
 
+            SwarmEvent::ConnectionClosed {
+                peer_id,
+                endpoint,
+                num_established,
+                cause,
+                ..
+            } => {
+                let short_id = peer_id_to_short_id.get(&peer_id).or_else(|| {
+                    tracing::error!(
+                        incoming_peer_id = %peer_id,
+                        incoming_remote_addr = %endpoint.get_remote_address(),
+                        "Libp2p node closed connection with an unknown peer"
+                    );
+                    None
+                });
+
+                if num_established == 0 {
+                    // No more connections, report disconnect
+                    Metrics::report_peer_disconnected();
+                }
+
+                tracing::info!(
+                    incoming_peer_id = %peer_id,
+                    incoming_short_id = ?short_id,
+                    incoming_remote_addr = %endpoint.get_remote_address(),
+                    remaining_connections = num_established,
+                    ?cause,
+                    "Libp2p node closed connection to peer"
+                );
+            }
+
             SwarmEvent::NewListenAddr { address, .. } => {
                 tracing::info!(
                     "Local node is listening on {}",

dcipher-agents/src/signer/threshold_signer/metrics.rs

@@ -0,0 +1,83 @@
+use prometheus::proto::MetricFamily;
+use prometheus::{IntCounter, IntGauge, Registry};
+use std::sync::LazyLock;
+
+pub struct Metrics {
+    registry: Registry,
+    connected_peers: IntGauge,
+    partials_sent: IntCounter,
+    partials_received: IntCounter,
+    invalid_partials: IntCounter,
+}
+
+static METRICS: LazyLock<Metrics> = LazyLock::new(|| {
+    let registry = Registry::new();
+
+    let connected_peers = IntGauge::new("libp2p_connected_peers", "Number of connected peers")
+        .expect("metrics failed to initialise");
+
+    let partials_sent = IntCounter::new(
+        "partial_signatures_sent",
+        "Number of partial signatures sent",
+    )
+    .expect("metrics failed to initialise");
+
+    let partials_received = IntCounter::new(
+        "partial_signatures_received",
+        "Number of partial signatures received",
+    )
+    .expect("metrics failed to initialise");
+
+    let invalid_partials = IntCounter::new(
+        "invalid_partial_signature_received",
+        "Number of invalid partial signatures received",
+    )
+    .expect("metrics failed to initialise");
+
+    registry
+        .register(Box::new(connected_peers.clone()))
+        .expect("metrics failed to initialise");
+    registry
+        .register(Box::new(partials_sent.clone()))
+        .expect("metrics failed to initialise");
+    registry
+        .register(Box::new(partials_received.clone()))
+        .expect("metrics failed to initialise");
+    registry
+        .register(Box::new(invalid_partials.clone()))
+        .expect("metrics failed to initialise");
+
+    Metrics {
+        registry,
+        connected_peers,
+        partials_received,
+        partials_sent,
+        invalid_partials,
+    }
+});
+
+impl Metrics {
+    pub(super) fn report_peer_connected() {
+        METRICS.connected_peers.inc();
+    }
+
+    pub(super) fn report_peer_disconnected() {
+        METRICS.connected_peers.dec();
+    }
+
+    pub(super) fn report_partials_received(count: u64) {
+        METRICS.partials_received.inc_by(count)
+    }
+
+    pub(super) fn report_partials_sent(count: u64) {
+        METRICS.partials_sent.inc_by(count)
+    }
+
+    pub(super) fn report_invalid_partials(count: u64) {
+        METRICS.invalid_partials.inc_by(count)
+    }
+
+    pub fn gather() -> Vec<MetricFamily> {
+        METRICS.registry.gather()
+    }
+}

randomness-agent/examples/randomness/healthcheck.rs

@@ -1,5 +1,6 @@
 use anyhow::anyhow;
-use dcipher_agents::agents::randomness::metrics::Metrics;
+use dcipher_agents::agents::blocklock::metrics::Metrics as BlocklockMetrics;
+use dcipher_agents::signer::threshold_signer::metrics::Metrics as ThresholdSignerMetrics;
 use prometheus::{Encoder, TextEncoder};
 use std::net::IpAddr;
 use warp::Filter;
@@ -12,7 +13,7 @@ pub async fn start_api(listen_addr: IpAddr, port: u16) -> anyhow::Result<()> {
 
     let metrics = warp::path!("metrics").map(|| {
         let encoder = TextEncoder::new();
-        let metrics = Metrics::gather();
+        let metrics = [BlocklockMetrics::gather(), ThresholdSignerMetrics::gather()].concat();
         let mut buffer = Vec::new();
 
         match encoder.encode(&metrics, &mut buffer) {