Add observability batcher (#1226)

Co-authored-by: IAvecilla <ignacio.avecilla@lambdaclass.com>
Co-authored-by: MauroFab <maurotoscano2@gmail.com>

Author: 3 people

batcher/Cargo.lock

@@ -31,3 +31,7 @@ bincode = "1.3.3"
 aligned-sdk = { path = "../aligned-sdk" }
 ciborium = "=0.2.2"
 priority-queue = "2.1.0"
+
+once_cell = "1.20.2"
+warp = "0.3.7"
+prometheus = { version = "0.13.4", features = ["process"] }

batcher/aligned-batcher/Cargo.toml

@@ -37,6 +37,7 @@ mod config;
 mod connection;
 mod eth;
 pub mod gnark;
+pub mod metrics;
 pub mod risc_zero;
 pub mod s3;
 pub mod sp1;
@@ -212,6 +213,7 @@ impl Batcher {
 
         // Let's spawn the handling of each connection in a separate task.
         while let Ok((stream, addr)) = listener.accept().await {
+            metrics::OPEN_CONNECTIONS.inc();
             let batcher = self.clone();
             tokio::spawn(batcher.handle_connection(stream, addr));
         }
@@ -294,6 +296,7 @@ impl Batcher {
             Ok(_) => info!("{} disconnected", &addr),
         }
 
+        metrics::OPEN_CONNECTIONS.dec();
         Ok(())
     }
 
@@ -313,6 +316,7 @@ impl Batcher {
         };
         let msg_nonce = client_msg.verification_data.nonce;
         debug!("Received message with nonce: {msg_nonce:?}",);
+        metrics::RECEIVED_PROOFS.inc();
 
         // * ---------------------------------------------------*
         // *        Perform validations over the message        *
@@ -1012,6 +1016,8 @@ impl Batcher {
 
         let proof_submitters = finalized_batch.iter().map(|entry| entry.sender).collect();
 
+        metrics::GAS_PRICE_USED_ON_LATEST_BATCH.set(gas_price.as_u64() as i64);
+
         match self
             .create_new_task(
                 *batch_merkle_root,
@@ -1023,6 +1029,7 @@ impl Batcher {
         {
             Ok(_) => {
                 info!("Batch verification task created on Aligned contract");
+                metrics::SENT_BATCHES.inc();
                 Ok(())
             }
             Err(e) => {
@@ -1031,6 +1038,7 @@ impl Batcher {
                     e
                 );
 
+                metrics::REVERTED_BATCHES.inc();
                 Err(e)
             }
         }

batcher/aligned-batcher/src/lib.rs

@@ -5,7 +5,9 @@ use std::sync::Arc;
 use clap::Parser;
 use env_logger::Env;
 
+use aligned_batcher::metrics;
 use aligned_batcher::{types::errors::BatcherError, Batcher};
+use warp::Filter;
 
 /// Batcher main flow:
 /// There are two main tasks spawned: `listen_connections` and `listen_new_blocks`
@@ -38,6 +40,14 @@ async fn main() -> Result<(), BatcherError> {
 
     env_logger::Builder::from_env(Env::default().default_filter_or("info")).init();
 
+    // Endpoint for Prometheus
+    metrics::init_variables();
+    let metrics_route = warp::path!("metrics").and_then(metrics::metrics_handler);
+    println!("Starting Batcher metrics on port 9093");
+    tokio::task::spawn(async move {
+        warp::serve(metrics_route).run(([0, 0, 0, 0], 9093)).await;
+    }); //TODO read from config
+
     let batcher = Batcher::new(cli.config).await;
     let batcher = Arc::new(batcher);
 
@@ -53,6 +63,8 @@ async fn main() -> Result<(), BatcherError> {
         }
     });
 
+    metrics::batcher_started();
+
     batcher.listen_connections(&addr).await?;
 
     Ok(())

batcher/aligned-batcher/src/main.rs

@@ -0,0 +1,67 @@
+// Prometheus
+use prometheus::{opts, register_int_counter, register_int_gauge, IntCounter, IntGauge};
+
+use warp::{Rejection, Reply};
+
+use once_cell::sync::Lazy;
+use std::{thread, time};
+
+// Prometheus setup
+pub static BATCHER_STARTED: Lazy<IntCounter> =
+    Lazy::new(|| register_int_counter!(opts!("batcher_started", "Batcher Started")).unwrap());
+
+pub static OPEN_CONNECTIONS: Lazy<IntGauge> =
+    Lazy::new(|| register_int_gauge!(opts!("open_connections", "Open Connections")).unwrap());
+
+pub static RECEIVED_PROOFS: Lazy<IntCounter> =
+    Lazy::new(|| register_int_counter!(opts!("received_proofs", "Received Proofs")).unwrap());
+
+pub static SENT_BATCHES: Lazy<IntCounter> =
+    Lazy::new(|| register_int_counter!(opts!("sent_batches", "Sent Batches")).unwrap());
+
+pub static REVERTED_BATCHES: Lazy<IntCounter> =
+    Lazy::new(|| register_int_counter!(opts!("reverted_batches", "Reverted Batches")).unwrap());
+
+pub static GAS_PRICE_USED_ON_LATEST_BATCH: Lazy<IntGauge> = Lazy::new(|| {
+    register_int_gauge!(opts!("gas_price_used_on_latest_batch", "Gas Price")).unwrap()
+});
+
+// so Prometheus can collect our metrics.
+pub async fn metrics_handler() -> Result<impl Reply, Rejection> {
+    use prometheus::Encoder;
+    let encoder = prometheus::TextEncoder::new();
+
+    let mut buffer = Vec::new();
+    if let Err(e) = encoder.encode(&prometheus::gather(), &mut buffer) {
+        eprintln!("could not encode prometheus metrics: {}", e);
+    };
+    let res = match String::from_utf8(buffer.clone()) {
+        Ok(v) => v,
+        Err(e) => {
+            eprintln!("prometheus metrics could not be from_utf8'd: {}", e);
+            String::default()
+        }
+    };
+    buffer.clear();
+
+    Ok(res)
+}
+
+pub fn init_variables() {
+    BATCHER_STARTED.reset();
+
+    OPEN_CONNECTIONS.set(0);
+
+    RECEIVED_PROOFS.reset();
+
+    SENT_BATCHES.reset();
+
+    REVERTED_BATCHES.reset();
+
+    GAS_PRICE_USED_ON_LATEST_BATCH.set(0);
+}
+
+pub fn batcher_started() {
+    thread::sleep(time::Duration::from_secs(10));
+    BATCHER_STARTED.inc();
+}