From ca00548795ffd4ec71af2152dfea346e85c906d9 Mon Sep 17 00:00:00 2001
From: Tumas <tumas@grandine.io>
Date: Mon, 9 Dec 2024 17:29:38 +0200
Subject: [PATCH] Add extra gossipsub metrics

- update `eth2_libp2p` metrics from within internal ticker
---
 Cargo.lock             |  1 -
 eth2_libp2p            |  2 +-
 metrics/Cargo.toml     |  1 -
 metrics/src/server.rs  | 14 --------------
 p2p/src/network.rs     | 17 ++++++++++++++++-
 runtime/src/runtime.rs |  1 -
 6 files changed, 17 insertions(+), 19 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index d805f11d..718045b2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5000,7 +5000,6 @@ dependencies = [
  "derive_more 1.0.0",
  "directories",
  "eth1_api",
- "eth2_libp2p",
  "fork_choice_control",
  "futures",
  "grandine_version",
diff --git a/eth2_libp2p b/eth2_libp2p
index 6985e3bb..6b9e80ea 160000
--- a/eth2_libp2p
+++ b/eth2_libp2p
@@ -1 +1 @@
-Subproject commit 6985e3bb0ff2fe260e00853f9443ce1bb535be9d
+Subproject commit 6b9e80ea7ccb4dcbae81f857d96e4e5126779a4e
diff --git a/metrics/Cargo.toml b/metrics/Cargo.toml
index 406211ac..fe79c225 100644
--- a/metrics/Cargo.toml
+++ b/metrics/Cargo.toml
@@ -15,7 +15,6 @@ chrono = { workspace = true }
 derive_more = { workspace = true }
 directories = { workspace = true }
 eth1_api = { workspace = true }
-eth2_libp2p = { workspace = true }
 fork_choice_control = { workspace = true }
 futures = { workspace = true }
 grandine_version = { workspace = true }
diff --git a/metrics/src/server.rs b/metrics/src/server.rs
index 3d9263f4..fde38d61 100644
--- a/metrics/src/server.rs
+++ b/metrics/src/server.rs
@@ -15,7 +15,6 @@ use axum::{
 };
 use directories::Directories;
 use eth1_api::ApiController;
-use eth2_libp2p::NetworkGlobals;
 use fork_choice_control::Wait;
 use futures::channel::mpsc::UnboundedSender;
 use helper_functions::misc;
@@ -58,7 +57,6 @@ pub struct MetricsState<P: Preset, W: Wait> {
     pub libp2p_registry: Option<Arc<Registry>>,
     pub metrics: Arc<Metrics>,
     pub metrics_to_metrics_tx: Option<UnboundedSender<MetricsToMetrics>>, // TODO: is still relevant, update naming if so
-    pub network_globals: Arc<NetworkGlobals>,
 }
 
 impl<P: Preset, W: Wait> FromRef<MetricsState<P, W>> for ApiController<P, W> {
@@ -91,12 +89,6 @@ impl<P: Preset, W: Wait> FromRef<MetricsState<P, W>> for Option<UnboundedSender<
     }
 }
 
-impl<P: Preset, W: Wait> FromRef<MetricsState<P, W>> for Arc<NetworkGlobals> {
-    fn from_ref(state: &MetricsState<P, W>) -> Self {
-        state.network_globals.clone_arc()
-    }
-}
-
 #[derive(Debug, Error)]
 pub enum Error {
     #[error("internal error")]
@@ -127,7 +119,6 @@ pub async fn run_metrics_server<P: Preset, W: Wait>(
     libp2p_registry: Option<Registry>,
     metrics: Arc<Metrics>,
     metrics_to_metrics_tx: Option<UnboundedSender<MetricsToMetrics>>,
-    network_globals: Arc<NetworkGlobals>,
 ) -> Result<()> {
     let addr = SocketAddr::from(&config);
 
@@ -140,7 +131,6 @@ pub async fn run_metrics_server<P: Preset, W: Wait>(
         libp2p_registry: libp2p_registry.map(Arc::new),
         metrics,
         metrics_to_metrics_tx,
-        network_globals,
     };
 
     let router = Router::new()
@@ -171,15 +161,11 @@ pub async fn prometheus_metrics<P: Preset, W: Wait>(
     State(libp2p_registry): State<Option<Arc<Registry>>>,
     State(metrics): State<Arc<Metrics>>,
     State(metrics_to_metrics_tx): State<Option<UnboundedSender<MetricsToMetrics>>>,
-    State(network_globals): State<Arc<NetworkGlobals>>,
 ) -> Result<String, Error> {
     let mut buffer = String::new();
 
     metrics.set_live();
 
-    eth2_libp2p::metrics::scrape_discovery_metrics();
-    eth2_libp2p::metrics::scrape_sync_metrics(&network_globals);
-
     // gossipsub metrics
     if let Some(registry) = libp2p_registry {
         prometheus_client::encoding::text::encode(&mut buffer, &registry)
diff --git a/p2p/src/network.rs b/p2p/src/network.rs
index 8198a310..bfe5840f 100644
--- a/p2p/src/network.rs
+++ b/p2p/src/network.rs
@@ -65,6 +65,7 @@ use crate::{
 };
 
 const GOSSIPSUB_PARAMETER_UPDATE_INTERVAL: Duration = Duration::from_secs(60);
+const NETWORK_METRICS_UPDATE_INTERVAL: Duration = Duration::from_secs(1);
 
 // > Clients MAY limit the number of blocks and sidecars in the response.
 const MAX_FOR_DOS_PREVENTION: u64 = 64;
@@ -1897,8 +1898,22 @@ fn run_network_service<P: Preset>(
     service_to_network_tx: UnboundedSender<ServiceOutboundMessage<P>>,
 ) {
     tokio::spawn(async move {
+        let mut network_metrics_update_interval =
+            IntervalStream::new(tokio::time::interval(NETWORK_METRICS_UPDATE_INTERVAL)).fuse();
+
+        let metrics_enabled = service.network_globals().network_config.metrics_enabled;
+
         loop {
-            select! {
+            tokio::select! {
+                _ = network_metrics_update_interval.select_next_some(), if metrics_enabled => {
+                    eth2_libp2p::metrics::update_discovery_metrics();
+                    eth2_libp2p::metrics::update_sync_metrics(service.network_globals());
+                    eth2_libp2p::metrics::update_gossipsub_extended_metrics(
+                        service.gossipsub(),
+                        service.network_globals(),
+                    );
+                },
+
                 network_event = service.next_event().fuse() => {
                     ServiceOutboundMessage::NetworkEvent(network_event).send(&service_to_network_tx);
                 }
diff --git a/runtime/src/runtime.rs b/runtime/src/runtime.rs
index 56a1cffa..4f9ba0f0 100644
--- a/runtime/src/runtime.rs
+++ b/runtime/src/runtime.rs
@@ -616,7 +616,6 @@ pub async fn run_after_genesis<P: Preset>(
                 .clone()
                 .expect("Metrics registry must be present for metrics server"),
             metrics_to_metrics_tx,
-            network.network_globals().clone_arc(),
         )),
         None => Either::Right(core::future::pending()),
     };