Merge pull request #41 from KasarLabs/feat/metrics

feat(metrics): added mapping sync block metrics for prometheus endpoint

CHANGELOG.md

@@ -2,6 +2,7 @@ git # Deoxys Changelog
 
 ## Next release
 
+- feat(metrics): Add prometheus metrics for mapping worker
 - feat(storage): finished migrating contract storage to our backend bonsai trie dbs
 - feat(storage): set up type-safe bonsai storage abstractions for usage in RPC
 - fix(root): fix state root computation

Cargo.toml

@@ -324,6 +324,7 @@ itertools = "0.12.1"
 jsonrpsee = { version = "0.16.3", default-features = false }
 lazy_static = { version = "1.4.0", default-features = false }
 log = { version = "0.4.20", default-features = false, features = ["std"] }
+num-traits = "0.2.17"
 num-bigint = "0.4.4"
 phf = { version = "0.11", default-features = false, features = ["std"] }
 pretty_assertions = "1.4.0"

crates/client/mapping-sync/Cargo.toml

@@ -26,7 +26,9 @@ mp-digest-log = { workspace = true }
 mp-hashers = { workspace = true }
 mp-transactions = { workspace = true }
 mp-types = { workspace = true }
+num-traits = { workspace = true }
 pallet-starknet-runtime-api = { workspace = true }
+prometheus-endpoint = { workspace = true }
 sc-client-api = { workspace = true }
 sp-api = { workspace = true }
 sp-blockchain = { workspace = true }

crates/client/mapping-sync/src/block_metrics.rs

@@ -0,0 +1,29 @@
+use prometheus_endpoint::prometheus::{Counter, Gauge};
+use prometheus_endpoint::{register, PrometheusError, Registry};
+
+#[derive(Clone, Debug)]
+pub struct BlockMetrics {
+    pub block_height: Gauge,
+    pub transaction_count: Counter,
+    pub event_count: Counter,
+    pub l1_gas_price_wei: Gauge,
+    pub l1_gas_price_strk: Gauge,
+}
+
+impl BlockMetrics {
+    pub fn register(registry: &Registry) -> Result<Self, PrometheusError> {
+        Ok(Self {
+            block_height: register(Gauge::new("deoxys_block_height", "Gauge for deoxys block height")?, registry)?,
+            transaction_count: register(
+                Counter::new("deoxys_transaction_count", "Counter for deoxys transaction count")?,
+                registry,
+            )?,
+            event_count: register(Counter::new("deoxys_event_count", "Counter for deoxys event count")?, registry)?,
+            l1_gas_price_wei: register(Gauge::new("deoxys_l1_gas_price", "Gauge for deoxys l1 gas price")?, registry)?,
+            l1_gas_price_strk: register(
+                Gauge::new("deoxys_l1_gas_price_strk", "Gauge for deoxys l1 gas price in strk")?,
+                registry,
+            )?,
+        })
+    }
+}

crates/client/mapping-sync/src/lib.rs

@@ -8,6 +8,7 @@
 //! # Usage
 //! The madara node should spawn a `MappingSyncWorker` among it's services.
 
+mod block_metrics;
 mod sync_blocks;
 
 use std::marker::PhantomData;
@@ -22,12 +23,15 @@ use log::debug;
 use mp_hashers::HasherT;
 use mp_types::block::{DBlockT, DHeaderT};
 use pallet_starknet_runtime_api::StarknetRuntimeApi;
+use prometheus_endpoint::prometheus;
 use sc_client_api::backend::{Backend, StorageProvider};
 use sc_client_api::client::ImportNotifications;
 use sp_api::ProvideRuntimeApi;
 use sp_blockchain::HeaderBackend;
 use sp_runtime::traits::Header as HeaderT;
 
+use crate::block_metrics::BlockMetrics;
+
 /// The worker in charge of syncing the Madara db when it receive a new Substrate block
 pub struct MappingSyncWorker<C, BE, H> {
     import_notifications: ImportNotifications<DBlockT>,
@@ -41,6 +45,7 @@ pub struct MappingSyncWorker<C, BE, H> {
     have_next: bool,
     retry_times: usize,
     sync_from: <DHeaderT as HeaderT>::Number,
+    block_metrics: Option<BlockMetrics>,
 }
 
 impl<C, BE, H> Unpin for MappingSyncWorker<C, BE, H> {}
@@ -54,7 +59,11 @@ impl<C, BE, H> MappingSyncWorker<C, BE, H> {
         substrate_backend: Arc<BE>,
         retry_times: usize,
         sync_from: <DHeaderT as HeaderT>::Number,
+        prometheus_registry: Option<prometheus::Registry>,
     ) -> Self {
+        let block_metrics =
+            prometheus_registry.and_then(|registry| block_metrics::BlockMetrics::register(&registry).ok());
+
         Self {
             import_notifications,
             timeout,
@@ -67,6 +76,7 @@ impl<C, BE, H> MappingSyncWorker<C, BE, H> {
             have_next: true,
             retry_times,
             sync_from,
+            block_metrics,
         }
     }
 }
@@ -116,6 +126,7 @@ where
                 self.substrate_backend.as_ref(),
                 self.retry_times,
                 self.sync_from,
+                self.block_metrics.as_ref(),
             ) {
                 Ok(have_next) => {
                     self.have_next = have_next;

crates/client/mapping-sync/src/sync_blocks.rs

@@ -4,13 +4,17 @@ use mp_digest_log::{find_starknet_block, FindLogError};
 use mp_hashers::HasherT;
 use mp_transactions::compute_hash::ComputeTransactionHash;
 use mp_types::block::{DBlockT, DHashT, DHeaderT};
+use num_traits::FromPrimitive;
 use pallet_starknet_runtime_api::StarknetRuntimeApi;
+use prometheus_endpoint::prometheus::core::Number;
 use sc_client_api::backend::{Backend, StorageProvider};
 use sp_api::ProvideRuntimeApi;
 use sp_blockchain::{Backend as _, HeaderBackend};
 use sp_runtime::traits::Header as HeaderT;
 
-fn sync_block<C, BE, H>(client: &C, header: &DHeaderT) -> anyhow::Result<()>
+use crate::block_metrics::BlockMetrics;
+
+fn sync_block<C, BE, H>(client: &C, header: &DHeaderT, block_metrics: Option<&BlockMetrics>) -> anyhow::Result<()>
 where
     // TODO: refactor this!
     C: HeaderBackend<DBlockT> + StorageProvider<DBlockT, BE>,
@@ -60,6 +64,27 @@ where
                                 .collect(),
                         };
 
+                        if let Some(block_metrics) = block_metrics {
+                            let starknet_block = &digest_starknet_block.clone();
+                            block_metrics.block_height.set(starknet_block.header().block_number.into_f64());
+
+                            // sending f64::MIN in case we exceed f64 (highly unlikely). The min numbers will
+                            // allow dashboards to catch anomalies so that it can be investigated.
+                            block_metrics
+                                .transaction_count
+                                .inc_by(f64::from_u128(starknet_block.header().transaction_count).unwrap_or(f64::MIN));
+                            block_metrics
+                                .event_count
+                                .inc_by(f64::from_u128(starknet_block.header().event_count).unwrap_or(f64::MIN));
+                            block_metrics.l1_gas_price_wei.set(
+                                f64::from_u128(starknet_block.header().l1_gas_price.price_in_wei).unwrap_or(f64::MIN),
+                            );
+
+                            block_metrics
+                                .l1_gas_price_strk
+                                .set(starknet_block.header().l1_gas_price.price_in_strk.unwrap_or(0).into_f64());
+                        }
+
                         DeoxysBackend::mapping().write_hashes(mapping_commitment).map_err(|e| anyhow::anyhow!(e))
                     }
                 }
@@ -106,6 +131,7 @@ fn sync_one_block<C, BE, H>(
     client: &C,
     substrate_backend: &BE,
     sync_from: <DHeaderT as HeaderT>::Number,
+    block_metrics: Option<&BlockMetrics>,
 ) -> anyhow::Result<bool>
 where
     C: ProvideRuntimeApi<DBlockT>,
@@ -145,7 +171,7 @@ where
         DeoxysBackend::meta().write_current_syncing_tips(current_syncing_tips)?;
         Ok(true)
     } else {
-        sync_block::<_, _, H>(client, &operating_header)?;
+        sync_block::<_, _, H>(client, &operating_header, block_metrics)?;
 
         current_syncing_tips.push(*operating_header.parent_hash());
         DeoxysBackend::meta().write_current_syncing_tips(current_syncing_tips)?;
@@ -158,6 +184,7 @@ pub fn sync_blocks<C, BE, H>(
     substrate_backend: &BE,
     limit: usize,
     sync_from: <DHeaderT as HeaderT>::Number,
+    block_metrics: Option<&BlockMetrics>,
 ) -> anyhow::Result<bool>
 where
     C: ProvideRuntimeApi<DBlockT>,
@@ -169,7 +196,7 @@ where
     let mut synced_any = false;
 
     for _ in 0..limit {
-        synced_any = synced_any || sync_one_block::<_, _, H>(client, substrate_backend, sync_from)?;
+        synced_any = synced_any || sync_one_block::<_, _, H>(client, substrate_backend, sync_from, block_metrics)?;
     }
 
     Ok(synced_any)

crates/node/src/service.rs

@@ -419,6 +419,7 @@ pub fn new_full(
             backend.clone(),
             3,
             0,
+            prometheus_registry.clone(),
         )
         .for_each(|()| future::ready(())),
     );