WIP: Add TCP traffic metrics

node/bft/events/Cargo.toml

@@ -18,7 +18,7 @@ edition = "2024"
 
 [features]
 default = [ ]
-metrics = [ "snarkvm/metrics" ]
+metrics = [ "dep:snarkos-node-metrics", "snarkvm/metrics" ]
 
 [dependencies.anyhow]
 workspace = true
@@ -33,6 +33,10 @@ features = [ "serde", "rayon" ]
 [dependencies.serde]
 workspace = true
 
+[dependencies.snarkos-node-metrics]
+workspace = true
+optional = true
+
 [dependencies.snarkos-node-sync-locators]
 workspace = true
 

node/bft/events/src/helpers/codec.rs

@@ -60,8 +60,17 @@ impl<N: Network> Encoder<Event<N>> for EventCodec<N> {
             .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidData, "serialization error"))?;
 
         let serialized_event = dst.split_to(dst.len()).freeze();
-
-        self.codec.encode(serialized_event, dst)
+        #[cfg(feature = "metrics")]
+        let num_bytes = serialized_event.len() as f64;
+        self.codec.encode(serialized_event, dst)?;
+        #[cfg(feature = "metrics")]
+        snarkvm::metrics::histogram_label(
+            snarkos_node_metrics::tcp::TCP_GATEWAY_EVENTS_OUTBOUND,
+            "event",
+            String::from(event.name().clone()),
+            num_bytes,
+        );
+        Ok(())
     }
 }
 
@@ -75,11 +84,21 @@ impl<N: Network> Decoder for EventCodec<N> {
             Some(bytes) => bytes,
             None => return Ok(None),
         };
-
+        #[cfg(feature = "metrics")]
+        let num_bytes = bytes.len() as f64;
         // Convert the bytes to an event, or fail if it is not valid.
         let reader = bytes.reader();
         match Event::read_le(reader) {
-            Ok(event) => Ok(Some(event)),
+            Ok(event) => {
+                #[cfg(feature = "metrics")]
+                snarkvm::metrics::histogram_label(
+                    snarkos_node_metrics::tcp::TCP_GATEWAY_EVENTS_INBOUND,
+                    "event",
+                    String::from(event.name().clone()),
+                    num_bytes,
+                );
+                Ok(Some(event))
+            }
             Err(error) => {
                 error!("Failed to deserialize an event: {}", error);
                 Err(std::io::ErrorKind::InvalidData.into())

node/metrics/src/names.rs

@@ -44,8 +44,15 @@ pub(super) const GAUGE_NAMES: [&str; 26] = [
     tcp::TCP_TASKS,
 ];
 
-pub(super) const HISTOGRAM_NAMES: [&str; 3] =
-    [bft::COMMIT_ROUNDS_LATENCY, consensus::CERTIFICATE_COMMIT_LATENCY, consensus::BLOCK_LATENCY];
+pub(super) const HISTOGRAM_NAMES: [&str; 7] = [
+    bft::COMMIT_ROUNDS_LATENCY,
+    consensus::CERTIFICATE_COMMIT_LATENCY,
+    consensus::BLOCK_LATENCY,
+    tcp::TCP_GATEWAY_EVENTS_OUTBOUND,
+    tcp::TCP_GATEWAY_EVENTS_INBOUND,
+    tcp::TCP_ROUTER_MESSAGES_OUTBOUND,
+    tcp::TCP_ROUTER_MESSAGES_INBOUND,
+];
 
 pub mod bft {
     pub const COMMIT_ROUNDS_LATENCY: &str = "snarkos_bft_commit_rounds_latency_secs"; // <-- This one doesn't even make sense.
@@ -93,4 +100,8 @@ pub mod router {
 
 pub mod tcp {
     pub const TCP_TASKS: &str = "snarkos_tcp_tasks_total";
+    pub const TCP_GATEWAY_EVENTS_OUTBOUND: &str = "snarkos_tcp_gateway_events_outbound";
+    pub const TCP_GATEWAY_EVENTS_INBOUND: &str = "snarkos_tcp_gateway_events_inbound";
+    pub const TCP_ROUTER_MESSAGES_OUTBOUND: &str = "snarkos_tcp_router_messages_outbound";
+    pub const TCP_ROUTER_MESSAGES_INBOUND: &str = "snarkos_tcp_router_messages_inbound";
 }

node/router/messages/Cargo.toml

@@ -19,6 +19,7 @@ edition = "2024"
 [features]
 default = [ ]
 test = [ ]
+metrics = [ "dep:snarkos-node-metrics", "snarkvm/metrics" ]
 
 [dependencies.bytes]
 workspace = true
@@ -29,6 +30,10 @@ workspace = true
 [dependencies.snarkos-node-bft-events]
 workspace = true
 
+[dependencies.snarkos-node-metrics]
+workspace = true
+optional = true
+
 [dependencies.snarkos-node-sync-locators]
 workspace = true
 

node/router/messages/src/helpers/codec.rs

@@ -60,8 +60,17 @@ impl<N: Network> Encoder<Message<N>> for MessageCodec<N> {
             .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidData, "serialization error"))?;
 
         let serialized_message = dst.split_to(dst.len()).freeze();
-
-        self.codec.encode(serialized_message, dst)
+        #[cfg(feature = "metrics")]
+        let num_bytes = serialized_message.len() as f64;
+        self.codec.encode(serialized_message, dst)?;
+        #[cfg(feature = "metrics")]
+        snarkvm::metrics::histogram_label(
+            snarkos_node_metrics::tcp::TCP_ROUTER_MESSAGES_OUTBOUND,
+            "message",
+            String::from(message.name().clone()),
+            num_bytes,
+        );
+        Ok(())
     }
 }
 
@@ -78,10 +87,22 @@ impl<N: Network> Decoder for MessageCodec<N> {
 
         Self::Item::check_size(&bytes)?;
 
+        #[cfg(feature = "metrics")]
+        let num_bytes = bytes.len() as f64;
+
         // Convert the bytes to a message, or fail if it is not valid.
         let reader = bytes.reader();
         match Message::read_le(reader) {
-            Ok(message) => Ok(Some(message)),
+            Ok(message) => {
+                #[cfg(feature = "metrics")]
+                snarkvm::metrics::histogram_label(
+                    snarkos_node_metrics::tcp::TCP_ROUTER_MESSAGES_INBOUND,
+                    "message",
+                    String::from(message.name().clone()),
+                    num_bytes,
+                );
+                Ok(Some(message))
+            }
             Err(error) => {
                 warn!("Failed to deserialize a message - {}", error);
                 Err(std::io::ErrorKind::InvalidData.into())