Add the payments poller metrics integration

Author: maximopalopoli

aggregation_mode/Cargo.lock

@@ -8,7 +8,7 @@ pub struct Config {
     pub db_connection_url: String,
     pub network: String,
     pub max_daily_proofs_per_user: i64,
-    pub metrics_port: u16,
+    pub gateway_metrics_port: u16,
 }
 
 impl Config {

aggregation_mode/gateway/src/config.rs

@@ -42,9 +42,12 @@ impl GatewayServer {
     pub fn new(db: Db, config: Config) -> Self {
         let network = Network::from_str(&config.network).expect("A valid network in config file");
 
-        tracing::info!("Starting metrics server on port {}", config.metrics_port);
-        let metrics =
-            GatewayMetrics::start(config.metrics_port).expect("Failed to start metrics server");
+        tracing::info!(
+            "Starting metrics server on port {}",
+            config.gateway_metrics_port
+        );
+        let metrics = GatewayMetrics::start(config.gateway_metrics_port)
+            .expect("Failed to start metrics server");
 
         Self {
             db,

aggregation_mode/gateway/src/http.rs

@@ -8,6 +8,9 @@ serde =  { workspace = true }
 serde_json = { workspace = true }
 serde_yaml = { workspace = true }
 aligned-sdk = { workspace = true }
+prometheus = { workspace = true }
+anyhow = { workspace = true }
+warp = { workspace = true }
 tracing = { version = "0.1", features = ["log"] }
 tracing-subscriber = { version = "0.3.0", features = ["env-filter"] }
 actix-web = "4"

aggregation_mode/payments_poller/Cargo.toml

@@ -8,6 +8,7 @@ pub struct Config {
     pub eth_rpc_url: String,
     pub payment_service_address: String,
     pub last_block_fetched_filepath: String,
+    pub poller_metrics_port: u16,
 }
 
 #[derive(Debug, Deserialize, Serialize)]

aggregation_mode/payments_poller/src/config.rs

@@ -1,4 +1,5 @@
 pub mod config;
 pub mod db;
+pub mod metrics;
 pub mod payments;
 pub mod types;

aggregation_mode/payments_poller/src/lib.rs

@@ -0,0 +1,54 @@
+use prometheus::{self, opts, register_gauge};
+use warp::{reject::Rejection, reply::Reply, Filter};
+
+#[derive(Clone, Debug)]
+pub struct PaymentsPollerMetrics {
+    pub last_processed_block: prometheus::Gauge,
+}
+
+impl PaymentsPollerMetrics {
+    pub fn start(metrics_port: u16) -> anyhow::Result<Self> {
+        let registry = prometheus::Registry::new();
+
+        let last_processed_block = register_gauge!(opts!(
+            "last_processed_block",
+            "Last processed block by poller"
+        ))?;
+
+        registry.register(Box::new(last_processed_block.clone()))?;
+
+        let metrics_route = warp::path!("metrics")
+            .and(warp::any().map(move || registry.clone()))
+            .and_then(PaymentsPollerMetrics::metrics_handler);
+
+        tokio::task::spawn(async move {
+            warp::serve(metrics_route)
+                .run(([0, 0, 0, 0], metrics_port))
+                .await;
+        });
+
+        Ok(Self {
+            last_processed_block,
+        })
+    }
+
+    pub async fn metrics_handler(registry: prometheus::Registry) -> Result<impl Reply, Rejection> {
+        use prometheus::Encoder;
+        let encoder = prometheus::TextEncoder::new();
+
+        let mut buffer = Vec::new();
+        if let Err(e) = encoder.encode(&registry.gather(), &mut buffer) {
+            eprintln!("could not encode prometheus metrics: {}", e);
+        };
+        let res = String::from_utf8(buffer.clone())
+            .inspect_err(|e| eprintln!("prometheus metrics could not be parsed correctly: {e}"))
+            .unwrap_or_default();
+        buffer.clear();
+
+        Ok(res)
+    }
+
+    pub fn register_last_processed_block(&self, value: u64) {
+        self.last_processed_block.set(value as f64);
+    }
+}

aggregation_mode/payments_poller/src/metrics.rs

@@ -3,6 +3,7 @@ use std::str::FromStr;
 use crate::{
     config::Config,
     db::Db,
+    metrics::PaymentsPollerMetrics,
     types::{AggregationModePaymentService, AggregationModePaymentServiceContract, RpcProvider},
 };
 use alloy::{
@@ -21,6 +22,7 @@ pub struct PaymentsPoller {
     proof_aggregation_service: AggregationModePaymentServiceContract,
     rpc_provider: RpcProvider,
     config: Config,
+    metrics: PaymentsPollerMetrics,
 }
 
 impl PaymentsPoller {
@@ -38,11 +40,19 @@ impl PaymentsPoller {
             .get_last_block_fetched()
             .map_err(|err| PaymentsPollerError::ReadLastBlockError(err.to_string()));
 
+        tracing::info!(
+            "Starting metrics server on port {}",
+            config.poller_metrics_port
+        );
+        let metrics = PaymentsPollerMetrics::start(config.poller_metrics_port)
+            .expect("Failed to start metrics server");
+
         Ok(Self {
             db,
             proof_aggregation_service,
             rpc_provider,
             config,
+            metrics,
         })
     }
 
@@ -121,6 +131,8 @@ impl PaymentsPoller {
                 continue;
             };
 
+            self.metrics.register_last_processed_block(current_block);
+
             tokio::time::sleep(std::time::Duration::from_secs(
                 seconds_to_wait_between_polls,
             ))

aggregation_mode/payments_poller/src/payments.rs

@@ -3,8 +3,9 @@ db_connection_url: "postgres://postgres:postgres@localhost:5435/"
 eth_rpc_url: "http://localhost:8545"
 payment_service_address: "0x922D6956C99E12DFeB3224DEA977D0939758A1Fe"
 network: "devnet"
-max_daily_proofs_per_user: 32
+max_daily_proofs_per_user: 100
 last_block_fetched_filepath: "config-files/proof-aggregator.last_block_fetched.json"
 
 # Metrics
-metrics_port: 9093
+gateway_metrics_port: 9094
+poller_metrics_port: 9095

config-files/config-agg-mode-gateway-ethereum-package.yaml

@@ -3,8 +3,9 @@ db_connection_url: "postgres://postgres:postgres@localhost:5435/"
 eth_rpc_url: "http://localhost:8545"
 payment_service_address: "0x922D6956C99E12DFeB3224DEA977D0939758A1Fe"
 network: "devnet"
-max_daily_proofs_per_user: 4
+max_daily_proofs_per_user: 100
 last_block_fetched_filepath: "config-files/proof-aggregator.last_block_fetched.json"
 
 # Metrics
-metrics_port: 9093
+gateway_metrics_port: 9094
+poller_metrics_port: 9095