feat(metrics): add push gateway support for Prometheus metrics (#19243)

Author: mablr

crates/node/builder/src/launch/common.rs

@@ -610,7 +610,7 @@ where
                         }
                     })
                     .build(),
-            );
+            ).with_push_gateway(self.node_config().metrics.push_gateway_url.clone(), self.node_config().metrics.push_gateway_interval);
 
             MetricServer::new(config).serve().await?;
         }

crates/node/core/src/args/metric.rs

@@ -1,6 +1,6 @@
 use clap::Parser;
-use reth_cli_util::parse_socket_address;
-use std::net::SocketAddr;
+use reth_cli_util::{parse_duration_from_secs, parse_socket_address};
+use std::{net::SocketAddr, time::Duration};
 
 /// Metrics configuration.
 #[derive(Debug, Clone, Default, Parser)]
@@ -10,4 +10,26 @@ pub struct MetricArgs {
     /// The metrics will be served at the given interface and port.
     #[arg(long="metrics", alias = "metrics.prometheus", value_name = "PROMETHEUS", value_parser = parse_socket_address, help_heading = "Metrics")]
     pub prometheus: Option<SocketAddr>,
+
+    /// URL for pushing Prometheus metrics to a push gateway.
+    ///
+    /// If set, the node will periodically push metrics to the specified push gateway URL.
+    #[arg(
+        long = "metrics.prometheus.push.url",
+        value_name = "PUSH_GATEWAY_URL",
+        help_heading = "Metrics"
+    )]
+    pub push_gateway_url: Option<String>,
+
+    /// Interval in seconds for pushing metrics to push gateway.
+    ///
+    /// Default: 5 seconds
+    #[arg(
+        long = "metrics.prometheus.push.interval",
+        default_value = "5",
+        value_parser = parse_duration_from_secs,
+        value_name = "SECONDS",
+        help_heading = "Metrics"
+    )]
+    pub push_gateway_interval: Duration,
 }

crates/node/core/src/node_config.rs

@@ -234,7 +234,7 @@ impl<ChainSpec> NodeConfig<ChainSpec> {
     }
 
     /// Set the metrics address for the node
-    pub const fn with_metrics(mut self, metrics: MetricArgs) -> Self {
+    pub fn with_metrics(mut self, metrics: MetricArgs) -> Self {
         self.metrics = metrics;
         self
     }

crates/node/metrics/Cargo.toml

@@ -21,6 +21,7 @@ tokio.workspace = true
 jsonrpsee-server.workspace = true
 http.workspace = true
 tower.workspace = true
+reqwest.workspace = true
 
 tracing.workspace = true
 eyre.workspace = true

crates/node/metrics/src/server.rs

@@ -8,9 +8,10 @@ use eyre::WrapErr;
 use http::{header::CONTENT_TYPE, HeaderValue, Response};
 use metrics::describe_gauge;
 use metrics_process::Collector;
+use reqwest::Client;
 use reth_metrics::metrics::Unit;
 use reth_tasks::TaskExecutor;
-use std::{convert::Infallible, net::SocketAddr, sync::Arc};
+use std::{convert::Infallible, net::SocketAddr, sync::Arc, time::Duration};
 
 /// Configuration for the [`MetricServer`]
 #[derive(Debug)]
@@ -20,6 +21,8 @@ pub struct MetricServerConfig {
     chain_spec_info: ChainSpecInfo,
     task_executor: TaskExecutor,
     hooks: Hooks,
+    push_gateway_url: Option<String>,
+    push_gateway_interval: Duration,
 }
 
 impl MetricServerConfig {
@@ -31,7 +34,22 @@ impl MetricServerConfig {
         task_executor: TaskExecutor,
         hooks: Hooks,
     ) -> Self {
-        Self { listen_addr, hooks, task_executor, version_info, chain_spec_info }
+        Self {
+            listen_addr,
+            hooks,
+            task_executor,
+            version_info,
+            chain_spec_info,
+            push_gateway_url: None,
+            push_gateway_interval: Duration::from_secs(5),
+        }
+    }
+
+    /// Set the gateway URL and interval for pushing metrics
+    pub fn with_push_gateway(mut self, url: Option<String>, interval: Duration) -> Self {
+        self.push_gateway_url = url;
+        self.push_gateway_interval = interval;
+        self
     }
 }
 
@@ -49,18 +67,35 @@ impl MetricServer {
 
     /// Spawns the metrics server
     pub async fn serve(&self) -> eyre::Result<()> {
-        let MetricServerConfig { listen_addr, hooks, task_executor, version_info, chain_spec_info } =
-            &self.config;
+        let MetricServerConfig {
+            listen_addr,
+            hooks,
+            task_executor,
+            version_info,
+            chain_spec_info,
+            push_gateway_url,
+            push_gateway_interval,
+        } = &self.config;
 
-        let hooks = hooks.clone();
+        let hooks_for_endpoint = hooks.clone();
         self.start_endpoint(
             *listen_addr,
-            Arc::new(move || hooks.iter().for_each(|hook| hook())),
+            Arc::new(move || hooks_for_endpoint.iter().for_each(|hook| hook())),
             task_executor.clone(),
         )
         .await
         .wrap_err_with(|| format!("Could not start Prometheus endpoint at {listen_addr}"))?;
 
+        // Start push-gateway task if configured
+        if let Some(url) = push_gateway_url {
+            self.start_push_gateway_task(
+                url.clone(),
+                *push_gateway_interval,
+                hooks.clone(),
+                task_executor.clone(),
+            )?;
+        }
+
         // Describe metrics after recorder installation
         describe_db_metrics();
         describe_static_file_metrics();
@@ -128,6 +163,51 @@ impl MetricServer {
 
         Ok(())
     }
+
+    /// Starts a background task to push metrics to a metrics gateway
+    fn start_push_gateway_task(
+        &self,
+        url: String,
+        interval: Duration,
+        hooks: Hooks,
+        task_executor: TaskExecutor,
+    ) -> eyre::Result<()> {
+        let client = Client::builder()
+            .build()
+            .wrap_err("Could not create HTTP client to push metrics to gateway")?;
+        task_executor.spawn_with_graceful_shutdown_signal(move |mut signal| {
+            Box::pin(async move {
+                tracing::info!(url = %url, interval = ?interval, "Starting task to push metrics to gateway");
+                let handle = install_prometheus_recorder();
+                loop {
+                    tokio::select! {
+                        _ = &mut signal => {
+                            tracing::info!("Shutting down task to push metrics to gateway");
+                            break;
+                        }
+                        _ = tokio::time::sleep(interval) => {
+                            hooks.iter().for_each(|hook| hook());
+                            let metrics = handle.handle().render();
+                            match client.put(&url).header("Content-Type", "text/plain").body(metrics).send().await {
+                                Ok(response) => {
+                                    if !response.status().is_success() {
+                                        tracing::warn!(
+                                            status = %response.status(),
+                                            "Failed to push metrics to gateway"
+                                        );
+                                    }
+                                }
+                                Err(err) => {
+                                    tracing::warn!(%err, "Failed to push metrics to gateway");
+                                }
+                            }
+                        }
+                    }
+                }
+            })
+        });
+        Ok(())
+    }
 }
 
 fn describe_db_metrics() {

docs/vocs/docs/pages/cli/reth/node.mdx

@@ -44,6 +44,18 @@ Metrics:
 
           The metrics will be served at the given interface and port.
 
+      --metrics.prometheus.push.url <PUSH_GATEWAY_URL>
+          URL for pushing Prometheus metrics to a push gateway.
+
+          If set, the node will periodically push metrics to the specified push gateway URL.
+
+      --metrics.prometheus.push.interval <SECONDS>
+          Interval in seconds for pushing metrics to push gateway.
+
+          Default: 5 seconds
+
+          [default: 5]
+
 Datadir:
       --datadir <DATA_DIR>
           The path to the data dir for all reth files and subdirectories.