apollo_network_benchmark: added receive metrics

Author: sirandreww-starkware

Cargo.lock

@@ -9,6 +9,7 @@ license-file.workspace = true
 testing = []
 
 [dependencies]
+apollo_metrics.workspace = true
 apollo_network.workspace = true
 clap = { workspace = true, features = ["derive", "env"] }
 futures.workspace = true

crates/apollo_network_benchmark/Cargo.toml

@@ -1,11 +1,18 @@
 use std::time::SystemTime;
 
+use apollo_metrics::metrics::LossyIntoF64;
 use libp2p::PeerId;
-use tracing::trace;
 
 use crate::message::StressTestMessage;
+use crate::metrics::{
+    RECEIVE_MESSAGE_BYTES,
+    RECEIVE_MESSAGE_BYTES_SUM,
+    RECEIVE_MESSAGE_COUNT,
+    RECEIVE_MESSAGE_DELAY_SECONDS,
+    RECEIVE_MESSAGE_NEGATIVE_DELAY_SECONDS,
+};
 
-pub fn receive_stress_test_message(received_message: Vec<u8>, sender_peer_id: Option<PeerId>) {
+pub fn receive_stress_test_message(received_message: Vec<u8>, _sender_peer_id: Option<PeerId>) {
     let end_time = SystemTime::now();
 
     let received_message: StressTestMessage = received_message.into();
@@ -18,6 +25,17 @@ pub fn receive_stress_test_message(received_message: Vec<u8>, sender_peer_id: Op
         }
     };
 
-    // TODO(AndrewL): Replace this with metric updates
-    trace!("Received stress test message from {sender_peer_id:?} in {delay_seconds} seconds");
+    // Use apollo_metrics for all metrics including labeled ones
+    RECEIVE_MESSAGE_BYTES.set(received_message.len().into_f64());
+    RECEIVE_MESSAGE_COUNT.increment(1);
+    RECEIVE_MESSAGE_BYTES_SUM.increment(
+        u64::try_from(received_message.len()).expect("Message length too large for u64"),
+    );
+
+    // Use apollo_metrics histograms for latency measurements
+    if delay_seconds.is_sign_positive() {
+        RECEIVE_MESSAGE_DELAY_SECONDS.record(delay_seconds);
+    } else {
+        RECEIVE_MESSAGE_NEGATIVE_DELAY_SECONDS.record(-delay_seconds);
+    }
 }

crates/apollo_network_benchmark/src/bin/broadcast_network_stress_test_node/handlers.rs

@@ -12,6 +12,7 @@ mod message_test;
 
 mod handlers;
 mod message;
+pub mod metrics;
 mod protocol;
 mod stress_test_node;
 

crates/apollo_network_benchmark/src/bin/broadcast_network_stress_test_node/main.rs

@@ -0,0 +1,11 @@
+use apollo_metrics::define_metrics;
+
+define_metrics!(
+    Infra => {
+        MetricGauge { RECEIVE_MESSAGE_BYTES, "receive_message_bytes", "Size of the stress test received message in bytes" },
+        MetricCounter { RECEIVE_MESSAGE_COUNT, "receive_message_count", "Number of stress test messages received via broadcast", init = 0 },
+        MetricCounter { RECEIVE_MESSAGE_BYTES_SUM, "receive_message_bytes_sum", "Sum of the stress test messages received via broadcast", init = 0 },
+        MetricHistogram { RECEIVE_MESSAGE_DELAY_SECONDS, "receive_message_delay_seconds", "Message delay in seconds" },
+        MetricHistogram { RECEIVE_MESSAGE_NEGATIVE_DELAY_SECONDS, "receive_message_negative_delay_seconds", "Negative message delay in seconds" },
+    },
+);