feat(aggregation-mode): add grafana dashboard

Author: maximopalopoli

aggregation_mode/Cargo.lock

@@ -16,6 +16,9 @@ aligned-sdk = { path = "../crates/sdk/" }
 db = { path = "./db" }
 sp1-sdk = "5.0.0"
 risc0-zkvm = { version = "3.0.3" }
+prometheus = { version = "0.13.4", features = ["process"] }
+anyhow = { version = "1.0" }
+warp = "0.3.7"
 
 [profile.release]
 opt-level = 3

aggregation_mode/Cargo.toml

@@ -7,6 +7,9 @@ edition = "2021"
 serde =  { workspace = true }
 serde_json = { workspace = true }
 serde_yaml = { workspace = true }
+prometheus = { workspace = true }
+anyhow = { workspace = true }
+warp = { workspace = true }
 agg_mode_sdk = { path = "../sdk"}
 aligned-sdk = { workspace = true }
 sp1-sdk = { workspace = true }
@@ -15,6 +18,7 @@ tracing-subscriber = { version = "0.3.0", features = ["env-filter"] }
 bincode = "1.3.3"
 actix-web = "4"
 actix-multipart = "0.7.2"
+actix-web-prometheus = "0.1.0-beta.8"
 alloy = { workspace = true }
 tokio = { version = "1", features = ["time"]}
 # TODO: enable tls

aggregation_mode/gateway/Cargo.toml

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

aggregation_mode/gateway/src/config.rs

@@ -1,4 +1,5 @@
 use std::{
+    collections::HashMap,
     str::FromStr,
     time::{SystemTime, UNIX_EPOCH},
 };
@@ -8,6 +9,7 @@ use actix_web::{
     web::{self, Data},
     App, HttpRequest, HttpResponse, HttpServer, Responder,
 };
+use actix_web_prometheus::PrometheusMetricsBuilder;
 use agg_mode_sdk::types::Network;
 use aligned_sdk::aggregation_layer::AggregationModeProvingSystem;
 use alloy::signers::Signature;
@@ -23,6 +25,7 @@ use crate::{
     config::Config,
     db::Db,
     helpers::get_time_left_day_formatted,
+    metrics::GatewayMetrics,
     types::{GetReceiptsResponse, SubmitProofRequestRisc0, SubmitProofRequestSP1},
     verifiers::{verify_sp1_proof, VerificationError},
 };
@@ -32,15 +35,22 @@ pub struct GatewayServer {
     db: Db,
     config: Config,
     network: Network,
+    metrics: GatewayMetrics,
 }
 
 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");
+
         Self {
             db,
             config,
             network,
+            metrics,
         }
     }
 
@@ -49,10 +59,19 @@ impl GatewayServer {
         let port = self.config.port;
         let state = self.clone();
 
+        let mut labels = HashMap::new();
+        labels.insert("label1".to_string(), "value1".to_string());
+        let prometheus = PrometheusMetricsBuilder::new("api")
+            .endpoint("/metrics")
+            .const_labels(labels)
+            .build()
+            .unwrap();
+
         tracing::info!("Starting server at port {}", self.config.port);
         HttpServer::new(move || {
             App::new()
                 .app_data(Data::new(state.clone()))
+                .wrap(prometheus.clone())
                 .route("/nonce/{address}", web::get().to(Self::get_nonce))
                 .route("/receipts", web::get().to(Self::get_receipts))
                 .route("/proof/sp1", web::post().to(Self::post_proof_sp1))

aggregation_mode/gateway/src/http.rs

@@ -2,5 +2,6 @@ pub mod config;
 pub mod db;
 mod helpers;
 pub mod http;
+mod metrics;
 mod types;
 mod verifiers;

aggregation_mode/gateway/src/lib.rs

@@ -0,0 +1,72 @@
+use prometheus::{self, opts, register_int_counter, IntCounter};
+use warp::{reject::Rejection, reply::Reply, Filter};
+
+#[derive(Clone, Debug)]
+pub struct GatewayMetrics {
+    pub success_response: IntCounter,
+    pub server_error_response: IntCounter,
+    pub user_error_response: IntCounter,
+}
+
+impl GatewayMetrics {
+    pub fn start(metrics_port: u16) -> anyhow::Result<Self> {
+        let registry = prometheus::Registry::new();
+
+        let success_response =
+            register_int_counter!(opts!("success_response_count", "Success Responses"))?;
+
+        let server_error_response =
+            register_int_counter!(opts!("server_error_response_count", "Success Responses"))?;
+
+        let user_error_response =
+            register_int_counter!(opts!("user_error_response_count", "Success Responses"))?;
+
+        registry.register(Box::new(success_response.clone()))?;
+        registry.register(Box::new(server_error_response.clone()))?;
+        registry.register(Box::new(user_error_response.clone()))?;
+
+        let metrics_route = warp::path!("metrics")
+            .and(warp::any().map(move || registry.clone()))
+            .and_then(GatewayMetrics::metrics_handler);
+
+        tokio::task::spawn(async move {
+            warp::serve(metrics_route)
+                .run(([0, 0, 0, 0], metrics_port))
+                .await;
+        });
+
+        Ok(Self {
+            success_response,
+            server_error_response,
+            user_error_response,
+        })
+    }
+
+    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 inc_success_response(&self) {
+        self.success_response.inc();
+    }
+
+    pub fn inc_server_error_response(&self) {
+        self.server_error_response.inc();
+    }
+
+    pub fn inc_user_error_response(&self) {
+        self.user_error_response.inc();
+    }
+}

aggregation_mode/gateway/src/metrics.rs

@@ -4,3 +4,7 @@ eth_rpc_url: "http://localhost:8545"
 payment_service_address: "0x922D6956C99E12DFeB3224DEA977D0939758A1Fe"
 network: "devnet"
 max_daily_proofs_per_user: 32
+last_block_fetched_filepath: "config-files/proof-aggregator.last_block_fetched.json"
+
+# Metrics
+metrics_port: 9093

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

@@ -5,3 +5,6 @@ payment_service_address: "0x922D6956C99E12DFeB3224DEA977D0939758A1Fe"
 network: "devnet"
 max_daily_proofs_per_user: 4
 last_block_fetched_filepath: "config-files/proof-aggregator.last_block_fetched.json"
+
+# Metrics
+metrics_port: 9093