feat(instrumentation): otel collector

.github/workflows/nexus-e2e-integrations-test.yml

@@ -51,7 +51,8 @@ jobs:
           echo "${HOME}/.risc0/bin" >> $GITHUB_PATH
           # Need to install for specific version next.
           ${HOME}/.risc0/bin/rzup install
-          ${HOME}/.risc0/bin/rzup install cargo-risczero 1.2.0
+          ${HOME}/.risc0/bin/rzup install cargo-risczero 1.2.5
+          ${HOME}/.risc0/bin/rzup install r0vm 1.2.5
 
       - name: Build binaries
         run : |

.github/workflows/nexus-integration-tests.yml

@@ -26,7 +26,8 @@ jobs:
           echo "${HOME}/.risc0/bin" >> $GITHUB_PATH
           # Need to install for specific version next.
           ${HOME}/.risc0/bin/rzup install
-          ${HOME}/.risc0/bin/rzup install cargo-risczero 1.2.0
+          ${HOME}/.risc0/bin/rzup install cargo-risczero 1.2.5
+          ${HOME}/.risc0/bin/rzup install r0vm 1.2.5
 
       - name: Run integration tests
         run: |

Cargo.toml

@@ -67,4 +67,4 @@ bumpalo = { git = "https://github.com/fitzgen/bumpalo", tag = "3.14.0" }
 ethereum_hashing = { git = "https://github.com/ncitron/ethereum_hashing", rev = "7ee70944ed4fabe301551da8c447e4f4ae5e6c35" }
 # Need to add this here so the precompile is added when building the helios deps.
 # IMP : Comment it out when building with risc0 feature flag (May cause issues otherwise in guest code build)
-bls12_381 = { git = "https://github.com/sp1-patches/bls12_381", tag = "patch-0.8.0-sp1-4.0.0", optional = true }
+# bls12_381 = { git = "https://github.com/sp1-patches/bls12_381", tag = "patch-0.8.0-sp1-4.0.0", optional = true }

core/Cargo.toml

@@ -37,6 +37,10 @@ hex = "0.4.3"
 winnow = "0.6.18"
 tracing = { workspace = true, optional = true }
 utoipa = { version = "5.3", optional = true }
+opentelemetry = { version = "0.24.0", features = [
+    "metrics",
+    "logs",
+], optional = true }
 
 [features]
 # default = ["native-risc0"]
@@ -49,6 +53,7 @@ native = [
     "tokio",
     "dep:tracing",
     "utoipa",
+    "dep:opentelemetry",
 ]
 zkvm = [ "sparse-merkle-tree/arch-32", "sparse-merkle-tree/std" ]
 native-sp1 = [ "sp1-sdk", "sp1-zkvm/verify", "sha2", "native" ]

core/src/lib.rs

@@ -4,6 +4,8 @@ pub mod db;
 mod h256;
 #[cfg(any(feature = "native"))]
 pub mod mempool;
+#[cfg(any(feature = "native"))]
+pub mod metrics;
 pub mod prover;
 pub mod state;
 #[cfg(any(feature = "native"))]

core/src/mempool.rs

@@ -1,3 +1,4 @@
+use crate::metrics::MempoolMetrics;
 use crate::{
     db::NodeDB,
     traits::NexusTransaction,
@@ -12,6 +13,7 @@ use tracing::{debug, error, event, info, instrument, span, warn, Level};
 pub struct Mempool {
     tx_list: Arc<Mutex<Vec<Transaction>>>,
     node_db: Arc<Mutex<NodeDB>>,
+    mempool_metrics: MempoolMetrics,
 }
 
 impl Mempool {
@@ -21,6 +23,7 @@ impl Mempool {
         Self {
             tx_list: Arc::new(Mutex::new(vec![])),
             node_db,
+            mempool_metrics: MempoolMetrics::init(),
         }
     }
 
@@ -41,6 +44,7 @@ impl Mempool {
     #[instrument(level = "debug", skip(self))]
     pub async fn clear_upto_tx(&self, index: usize) -> () {
         debug!("Clearing transactions up to index {} from mempool", index);
+        let (_, len) = self.get_current_txs().await;
         let mut tx_list = self.tx_list.lock().await;
         // Clear transactions up to the specified index
         if index < tx_list.len() {
@@ -49,6 +53,9 @@ impl Mempool {
             // Handle case where index exceeds the length of tx_list
             tx_list.clear();
         }
+        if let Some(len) = len {
+            self.mempool_metrics.mempool_txn_count_gauge.record((len - index) as u64, &[]);
+        }
     }
 
     #[instrument(level = "debug", skip(self))]
@@ -74,6 +81,8 @@ impl Mempool {
 
                 tx_list.push(tx);
 
+                self.mempool_metrics.mempool_txn_count_gauge.record(tx_list.len() as u64, &[]);
+
                 info!("Transaction successfully added to mempool");
                 Ok(())
             }

core/src/metrics.rs

@@ -0,0 +1,222 @@
+use opentelemetry::global;
+use opentelemetry::metrics::{Counter, Gauge, Histogram};
+use std::time::Instant;
+
+#[derive(Clone)]
+pub struct BatchMetrics {
+    pub batch_number_counter: Counter<u64>,
+    pub number_of_transactions_batch: Histogram<u64>,
+}
+impl BatchMetrics {
+    pub fn init() -> Self {
+        let metrics_meter = global::meter("nexus-metrics");
+        let batch_number_counter = metrics_meter.u64_counter("block_number").with_unit("Blocks").init();
+        let number_of_transactions_batch = metrics_meter.u64_histogram("number_of_transactions").init();
+        Self {
+            batch_number_counter,
+            number_of_transactions_batch,
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct MempoolMetrics {
+    pub mempool_txn_count_gauge: Gauge<u64>,
+}
+impl MempoolMetrics {
+    pub fn init() -> Self {
+        let metrics_meter = global::meter("nexus-metrics");
+        let mempool_txn_count_gauge = metrics_meter.u64_gauge("mempool_txn_count").with_unit("Transactions").init();
+        Self { mempool_txn_count_gauge }
+    }
+}
+
+#[derive(Clone)]
+pub struct ExecutionMetrics {
+    pub batch_execution_time: Histogram<u64>,
+    pub batch_proving_time: Histogram<u64>,
+    pub total_batch_execution_time: Histogram<u64>,
+}
+impl ExecutionMetrics {
+    pub fn init() -> Self {
+        let metrics_meter = global::meter("nexus-metrics");
+        let batch_execution_time = metrics_meter.u64_histogram("batch_execution_time").init();
+        let batch_proving_time = metrics_meter.u64_histogram("batch_proving_time").init();
+        let total_batch_execution_time = metrics_meter.u64_histogram("total_batch_execution_time").init();
+        Self {
+            batch_execution_time,
+            batch_proving_time,
+            total_batch_execution_time,
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct ApiMetrics {
+    submit_tx_counter: Counter<u64>,
+    tx_status_counter: Counter<u64>,
+    get_block_counter: Counter<u64>,
+    get_state_counter: Counter<u64>,
+    get_state_hex_counter: Counter<u64>,
+    get_header_counter: Counter<u64>,
+    range_counter: Counter<u64>,
+
+    submit_tx_response_time: Histogram<f64>,
+    tx_status_response_time: Histogram<f64>,
+    get_block_response_time: Histogram<f64>,
+    get_state_response_time: Histogram<f64>,
+    get_state_hex_response_time: Histogram<f64>,
+    get_header_response_time: Histogram<f64>,
+    range_response_time: Histogram<f64>,
+}
+
+impl ApiMetrics {
+    pub fn new() -> Self {
+        let meter = global::meter("nexus_api");
+
+        let submit_tx_counter = meter
+            .u64_counter("nexus_api_submit_tx_requests")
+            .with_description("Number of transaction submission requests")
+            .init();
+
+        let tx_status_counter = meter
+            .u64_counter("nexus_api_tx_status_requests")
+            .with_description("Number of transaction status requests")
+            .init();
+
+        let get_block_counter = meter
+            .u64_counter("nexus_api_get_block_requests")
+            .with_description("Number of block retrieval requests")
+            .init();
+
+        let get_state_counter = meter
+            .u64_counter("nexus_api_get_state_requests")
+            .with_description("Number of account state requests")
+            .init();
+
+        let get_state_hex_counter = meter
+            .u64_counter("nexus_api_get_state_hex_requests")
+            .with_description("Number of hex account state requests")
+            .init();
+
+        let get_header_counter = meter
+            .u64_counter("nexus_api_get_header_requests")
+            .with_description("Number of header requests")
+            .init();
+
+        let range_counter = meter
+            .u64_counter("nexus_api_range_requests")
+            .with_description("Number of block range requests")
+            .init();
+
+        // Create histograms for response times (in milliseconds)
+        let submit_tx_response_time = meter
+            .f64_histogram("nexus_api_submit_tx_response_time")
+            .with_description("Response time for transaction submission")
+            .with_unit("ms")
+            .init();
+
+        let tx_status_response_time = meter
+            .f64_histogram("nexus_api_tx_status_response_time")
+            .with_description("Response time for transaction status requests")
+            .with_unit("ms")
+            .init();
+
+        let get_block_response_time = meter
+            .f64_histogram("nexus_api_get_block_response_time")
+            .with_description("Response time for block retrieval")
+            .with_unit("ms")
+            .init();
+
+        let get_state_response_time = meter
+            .f64_histogram("nexus_api_get_state_response_time")
+            .with_description("Response time for account state requests")
+            .with_unit("ms")
+            .init();
+
+        let get_state_hex_response_time = meter
+            .f64_histogram("nexus_api_get_state_hex_response_time")
+            .with_description("Response time for hex account state requests")
+            .with_unit("ms")
+            .init();
+
+        let get_header_response_time = meter
+            .f64_histogram("nexus_api_get_header_response_time")
+            .with_description("Response time for header requests")
+            .with_unit("ms")
+            .init();
+
+        let range_response_time = meter
+            .f64_histogram("nexus_api_range_response_time")
+            .with_description("Response time for block range requests")
+            .with_unit("ms")
+            .init();
+
+        Self {
+            submit_tx_counter,
+            tx_status_counter,
+            get_block_counter,
+            get_state_counter,
+            get_state_hex_counter,
+            get_header_counter,
+            range_counter,
+
+            submit_tx_response_time,
+            tx_status_response_time,
+            get_block_response_time,
+            get_state_response_time,
+            get_state_hex_response_time,
+            get_header_response_time,
+            range_response_time,
+        }
+    }
+
+    // Measure response time and increment counter for submit_tx endpoint
+    pub fn record_submit_tx_request(&self, start_time: Instant) {
+        self.submit_tx_counter.add(1, &[]);
+        let duration = start_time.elapsed().as_millis() as f64;
+        self.submit_tx_response_time.record(duration, &[]);
+    }
+
+    // Measure response time and increment counter for tx_status endpoint
+    pub fn record_tx_status_request(&self, start_time: Instant) {
+        self.tx_status_counter.add(1, &[]);
+        let duration = start_time.elapsed().as_millis() as f64;
+        self.tx_status_response_time.record(duration, &[]);
+    }
+
+    // Measure response time and increment counter for get_block endpoint
+    pub fn record_get_block_request(&self, start_time: Instant) {
+        self.get_block_counter.add(1, &[]);
+        let duration = start_time.elapsed().as_millis() as f64;
+        self.get_block_response_time.record(duration, &[]);
+    }
+
+    // Measure response time and increment counter for get_state endpoint
+    pub fn record_get_state_request(&self, start_time: Instant) {
+        self.get_state_counter.add(1, &[]);
+        let duration = start_time.elapsed().as_millis() as f64;
+        self.get_state_response_time.record(duration, &[]);
+    }
+
+    // Measure response time and increment counter for get_state_hex endpoint
+    pub fn record_get_state_hex_request(&self, start_time: Instant) {
+        self.get_state_hex_counter.add(1, &[]);
+        let duration = start_time.elapsed().as_millis() as f64;
+        self.get_state_hex_response_time.record(duration, &[]);
+    }
+
+    // Measure response time and increment counter for get_header endpoint
+    pub fn record_get_header_request(&self, start_time: Instant) {
+        self.get_header_counter.add(1, &[]);
+        let duration = start_time.elapsed().as_millis() as f64;
+        self.get_header_response_time.record(duration, &[]);
+    }
+
+    // Measure response time and increment counter for range endpoint
+    pub fn record_range_request(&self, start_time: Instant) {
+        self.range_counter.add(1, &[]);
+        let duration = start_time.elapsed().as_millis() as f64;
+        self.range_response_time.record(duration, &[]);
+    }
+}