apollo_network: gossipsub event metrics

Author: sirandreww-starkware

crates/apollo_consensus_manager/src/consensus_manager.rs

@@ -98,6 +98,7 @@ impl ConsensusManager {
             num_blacklisted_peers: CONSENSUS_NUM_BLACKLISTED_PEERS,
             broadcast_metrics_by_topic: Some(broadcast_metrics_by_topic),
             sqmr_metrics: None,
+            gossipsub_metrics: None,
         });
         let mut network_manager =
             NetworkManager::new(self.config.network_config.clone(), None, network_manager_metrics);

crates/apollo_mempool_p2p/src/lib.rs

@@ -50,6 +50,7 @@ pub fn create_p2p_propagator_and_runner(
         num_blacklisted_peers: MEMPOOL_P2P_NUM_BLACKLISTED_PEERS,
         broadcast_metrics_by_topic: Some(broadcast_metrics_by_topic),
         sqmr_metrics: None,
+        gossipsub_metrics: None,
     });
     let mut network_manager = NetworkManager::new(
         mempool_p2p_config.network_config,

crates/apollo_network/src/gossipsub_impl.rs

@@ -12,8 +12,26 @@ pub type Topic = gossipsub::Sha256Topic;
 
 #[derive(Debug)]
 pub enum ExternalEvent {
-    #[allow(dead_code)]
-    Received { originated_peer_id: PeerId, message: Bytes, topic_hash: TopicHash },
+    Received { 
+        originated_peer_id: PeerId, 
+        message: Bytes, 
+        topic_hash: TopicHash 
+    },
+    Subscribed { 
+        peer_id: PeerId, 
+        topic_hash: TopicHash 
+    },
+    Unsubscribed { 
+        peer_id: PeerId, 
+        topic_hash: TopicHash 
+    },
+    GossipsubNotSupported { 
+        peer_id: PeerId 
+    },
+    SlowPeer { 
+        peer_id: PeerId, 
+        failed_messages: gossipsub::FailedMessages 
+    },
 }
 
 impl From<gossipsub::Event> for mixed_behaviour::Event {
@@ -40,9 +58,37 @@ impl From<gossipsub::Event> for mixed_behaviour::Event {
                     },
                 ))
             }
-            _ => mixed_behaviour::Event::ToOtherBehaviourEvent(
-                mixed_behaviour::ToOtherBehaviourEvent::NoOp,
-            ),
+            gossipsub::Event::Subscribed { peer_id, topic } => {
+                mixed_behaviour::Event::ExternalEvent(mixed_behaviour::ExternalEvent::GossipSub(
+                    ExternalEvent::Subscribed {
+                        peer_id,
+                        topic_hash: topic,
+                    },
+                ))
+            }
+            gossipsub::Event::Unsubscribed { peer_id, topic } => {
+                mixed_behaviour::Event::ExternalEvent(mixed_behaviour::ExternalEvent::GossipSub(
+                    ExternalEvent::Unsubscribed {
+                        peer_id,
+                        topic_hash: topic,
+                    },
+                ))
+            }
+            gossipsub::Event::GossipsubNotSupported { peer_id } => {
+                mixed_behaviour::Event::ExternalEvent(mixed_behaviour::ExternalEvent::GossipSub(
+                    ExternalEvent::GossipsubNotSupported {
+                        peer_id,
+                    },
+                ))
+            }
+            gossipsub::Event::SlowPeer { peer_id, failed_messages } => {
+                mixed_behaviour::Event::ExternalEvent(mixed_behaviour::ExternalEvent::GossipSub(
+                    ExternalEvent::SlowPeer {
+                        peer_id,
+                        failed_messages,
+                    },
+                ))
+            }
         }
     }
 }

crates/apollo_network/src/network_manager/metrics.rs

@@ -29,13 +29,42 @@ impl SqmrNetworkMetrics {
     }
 }
 
+pub struct GossipsubMetrics {
+    pub num_messages_received: MetricCounter,
+    pub num_messages_published: MetricCounter,
+    pub num_subscriptions: MetricCounter,
+    pub num_unsubscriptions: MetricCounter,
+    pub num_peer_subscribed: MetricCounter,
+    pub num_peer_unsubscribed: MetricCounter,
+    pub num_gossipsub_not_supported: MetricCounter,
+    pub num_slow_peers: MetricCounter,
+    pub num_peer_added: MetricCounter,
+    pub num_peer_removed: MetricCounter,
+}
+
+impl GossipsubMetrics {
+    pub fn register(&self) {
+        self.num_messages_received.register();
+        self.num_messages_published.register();
+        self.num_subscriptions.register();
+        self.num_unsubscriptions.register();
+        self.num_peer_subscribed.register();
+        self.num_peer_unsubscribed.register();
+        self.num_gossipsub_not_supported.register();
+        self.num_slow_peers.register();
+        self.num_peer_added.register();
+        self.num_peer_removed.register();
+    }
+}
+
 // TODO(alonl, shahak): Consider making these fields private and receive Topics instead of
 // TopicHashes in the constructor
 pub struct NetworkMetrics {
     pub num_connected_peers: MetricGauge,
     pub num_blacklisted_peers: MetricGauge,
     pub broadcast_metrics_by_topic: Option<HashMap<TopicHash, BroadcastNetworkMetrics>>,
     pub sqmr_metrics: Option<SqmrNetworkMetrics>,
+    pub gossipsub_metrics: Option<GossipsubMetrics>,
 }
 
 impl NetworkMetrics {
@@ -52,5 +81,8 @@ impl NetworkMetrics {
         if let Some(sqmr_metrics) = self.sqmr_metrics.as_ref() {
             sqmr_metrics.register();
         }
+        if let Some(gossipsub_metrics) = self.gossipsub_metrics.as_ref() {
+            gossipsub_metrics.register();
+        }
     }
 }

crates/apollo_network/src/network_manager/mod.rs

@@ -526,23 +526,58 @@ impl<SwarmT: SwarmTrait> GenericNetworkManager<SwarmT> {
         }
     }
 
+    /// Update the metrics for gossipsub events if these metrics were requested.
+    fn update_gossipsub_metrics_on_event(&self, event: &gossipsub_impl::ExternalEvent) {
+        if let Some(gossipsub_metrics) =
+            self.metrics.as_ref().and_then(|metrics| metrics.gossipsub_metrics.as_ref())
+        {
+            match &event {
+                gossipsub_impl::ExternalEvent::Received { .. } => {
+                    gossipsub_metrics.num_messages_received.increment(1);
+                }
+                gossipsub_impl::ExternalEvent::Subscribed { .. } => {
+                    gossipsub_metrics.num_peer_subscribed.increment(1);
+                }
+                gossipsub_impl::ExternalEvent::Unsubscribed { .. } => {
+                    gossipsub_metrics.num_peer_unsubscribed.increment(1);
+                }
+                gossipsub_impl::ExternalEvent::GossipsubNotSupported { .. } => {
+                    gossipsub_metrics.num_gossipsub_not_supported.increment(1);
+                }
+                gossipsub_impl::ExternalEvent::SlowPeer { .. } => {
+                    gossipsub_metrics.num_slow_peers.increment(1);
+                }
+            }
+        }
+    }
+
     fn handle_gossipsub_behaviour_event(
         &mut self,
         event: gossipsub_impl::ExternalEvent,
     ) -> Result<(), NetworkError> {
+        // Record gossipsub metrics if available
+        self.update_gossipsub_metrics_on_event(&event);
+
+        let gossipsub_impl::ExternalEvent::Received { originated_peer_id, message, topic_hash } =
+            event
+        else {
+            return Ok(());
+        };
+
+        // Record broadcast metrics for legacy compatibility
         if let Some(broadcast_metrics_by_topic) =
             self.metrics.as_ref().and_then(|metrics| metrics.broadcast_metrics_by_topic.as_ref())
         {
-            let gossipsub_impl::ExternalEvent::Received { ref topic_hash, .. } = event;
-            match broadcast_metrics_by_topic.get(topic_hash) {
+            match broadcast_metrics_by_topic.get(&topic_hash) {
                 Some(broadcast_metrics) => {
                     broadcast_metrics.num_received_broadcast_messages.increment(1)
                 }
-                None => error!("Attempted to update topic metric with unregistered topic_hash"),
+                None => {
+                    error!("Attempted to update topic metric with unregistered topic_hash")
+                }
             }
         }
-        let gossipsub_impl::ExternalEvent::Received { originated_peer_id, message, topic_hash } =
-            event;
+
         trace!("Received broadcast message with topic hash: {topic_hash:?}");
         let broadcasted_message_metadata = BroadcastedMessageMetadata {
             originator_id: OpaquePeerId::private_new(originated_peer_id),

crates/apollo_state_sync/src/runner/mod.rs

@@ -201,6 +201,7 @@ impl StateSyncRunner {
                     num_active_inbound_sessions: P2P_SYNC_NUM_ACTIVE_INBOUND_SESSIONS,
                     num_active_outbound_sessions: P2P_SYNC_NUM_ACTIVE_OUTBOUND_SESSIONS,
                 }),
+                gossipsub_metrics: None,
             });
             NetworkManager::new(
                 network_config.clone(),