Improved metrics for the database (#2007)

- Added database metrics per column. - Added statistic about commit time of each database. - Refactored how metrics are registered: Now, we use only one register shared between all metrics. This global register is used to encode all metrics. ### Before requesting review - [x] I have reviewed the code myself --------- Co-authored-by: Hannes Karppila <2204863+Dentosal@users.noreply.github.com>
FuelLabs · Jul 5, 2024 · 6979ce6 · 6979ce6
1 parent 916b9cb
commit 6979ce6
Show file tree

Hide file tree

Showing 20 changed files with 247 additions and 247 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ## [Unreleased]
 
 ### Added
+- [#2007](https://github.com/FuelLabs/fuel-core/pull/2007): Improved metrics:
+  - Added database metrics per column.
+  - Added statistic about commit time of each database.
+  - Refactored how metrics are registered: Now, we use only one register shared between all metrics. This global register is used to encode all metrics.
 - [#1996](https://github.com/FuelLabs/fuel-core/pull/1996): Added support for rollback command when state rewind feature is enabled. The command allows the rollback of the state of the blockchain several blocks behind until the end of the historical window. The default historical window it 7 days.
 - [#1996](https://github.com/FuelLabs/fuel-core/pull/1996): Added support for the state rewind feature. The feature allows the execution of the blocks in the past and the same execution results to be received. Together with forkless upgrades, execution of any block from the past is possible if historical data exist for the target block height.
 - [#1994](https://github.com/FuelLabs/fuel-core/pull/1994): Added the actual implementation for the `AtomicView::latest_view`.

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/fuel-core/src/database/database_description.rs b/crates/fuel-core/src/database/database_description.rs
@@ -57,7 +57,7 @@ pub trait DatabaseDescription: 'static + Copy + Debug + Send + Sync {
     fn version() -> u32;
 
     /// Returns the name of the database.
-    fn name() -> &'static str;
+    fn name() -> String;
 
     /// Returns the column used to store the metadata.
     fn metadata_column() -> Self::Column;

diff --git a/crates/fuel-core/src/database/database_description/off_chain.rs b/crates/fuel-core/src/database/database_description/off_chain.rs
@@ -15,8 +15,8 @@ impl DatabaseDescription for OffChain {
         0
     }
 
-    fn name() -> &'static str {
-        "off_chain"
+    fn name() -> String {
+        "off_chain".to_string()
     }
 
     fn metadata_column() -> Self::Column {

diff --git a/crates/fuel-core/src/database/database_description/on_chain.rs b/crates/fuel-core/src/database/database_description/on_chain.rs
@@ -12,8 +12,8 @@ impl DatabaseDescription for OnChain {
         0
     }
 
-    fn name() -> &'static str {
-        "on_chain"
+    fn name() -> String {
+        "on_chain".to_string()
     }
 
     fn metadata_column() -> Self::Column {

diff --git a/crates/fuel-core/src/database/database_description/relayer.rs b/crates/fuel-core/src/database/database_description/relayer.rs
@@ -45,8 +45,8 @@ impl DatabaseDescription for Relayer {
         0
     }
 
-    fn name() -> &'static str {
-        "relayer"
+    fn name() -> String {
+        "relayer".to_string()
     }
 
     fn metadata_column() -> Self::Column {

diff --git a/crates/fuel-core/src/service/metrics.rs b/crates/fuel-core/src/service/metrics.rs
@@ -1,9 +1,31 @@
 use axum::{
     body::Body,
     http::Request,
-    response::IntoResponse,
+    response::{
+        IntoResponse,
+        Response,
+    },
 };
-use fuel_core_metrics::response::encode_metrics_response;
+use fuel_core_metrics::encode_metrics;
+
+pub fn encode_metrics_response() -> impl IntoResponse {
+    let encoded = encode_metrics();
+    if let Ok(body) = encoded {
+        Response::builder()
+            .status(200)
+            .body(Body::from(body))
+            .unwrap()
+    } else {
+        error_body()
+    }
+}
+
+fn error_body() -> Response<Body> {
+    Response::builder()
+        .status(503)
+        .body(Body::from(""))
+        .unwrap()
+}
 
 pub async fn metrics(_req: Request<Body>) -> impl IntoResponse {
     encode_metrics_response()

diff --git a/crates/fuel-core/src/state/historical_rocksdb/description.rs b/crates/fuel-core/src/state/historical_rocksdb/description.rs
@@ -67,8 +67,8 @@ where
         Description::version()
     }
 
-    fn name() -> &'static str {
-        Description::name()
+    fn name() -> String {
+        format!("{}_history", Description::name())
     }
 
     fn metadata_column() -> Self::Column {

diff --git a/crates/fuel-core/src/state/rocks_db.rs b/crates/fuel-core/src/state/rocks_db.rs
@@ -7,7 +7,7 @@ use crate::{
     },
     state::IterDirection,
 };
-use fuel_core_metrics::core_metrics::database_metrics;
+use fuel_core_metrics::core_metrics::DatabaseMetrics;
 use fuel_core_storage::{
     iter::{
         BoxedIter,
@@ -124,6 +124,7 @@ pub struct RocksDb<Description> {
     read_options: ReadOptions,
     db: Arc<DB>,
     snapshot: Option<rocksdb::SnapshotWithThreadMode<'static, DB>>,
+    metrics: Arc<DatabaseMetrics>,
     // used for RAII
     _drop: Arc<DropResources>,
     _marker: core::marker::PhantomData<Description>,
@@ -258,6 +259,14 @@ where
     {
         let original_path = path.as_ref().to_path_buf();
         let path = original_path.join(Description::name());
+        let metric_columns = columns
+            .iter()
+            .map(|column| (column.id(), column.name()))
+            .collect::<Vec<_>>();
+        let metrics = Arc::new(DatabaseMetrics::new(
+            Description::name().as_str(),
+            &metric_columns,
+        ));
         let mut block_opts = BlockBasedOptions::default();
         // See https://github.com/facebook/rocksdb/blob/a1523efcdf2f0e8133b9a9f6e170a0dad49f928f/include/rocksdb/table.h#L246-L271 for details on what the format versions are/do.
         block_opts.set_format_version(5);
@@ -348,6 +357,7 @@ where
             read_options: Self::generate_read_options(&None),
             snapshot: None,
             db,
+            metrics,
             _drop: Default::default(),
             _marker: Default::default(),
         };
@@ -377,6 +387,7 @@ where
         &self,
     ) -> RocksDb<TargetDescription> {
         let db = self.db.clone();
+        let metrics = self.metrics.clone();
         let _drop = self._drop.clone();
 
         // Safety: We are transmuting the snapshot to 'static lifetime, but it's safe
@@ -393,6 +404,7 @@ where
             read_options: Self::generate_read_options(&snapshot),
             snapshot,
             db,
+            metrics,
             _drop,
             _marker: Default::default(),
         }
@@ -489,16 +501,18 @@ where
         opts: ReadOptions,
         iter_mode: IteratorMode,
     ) -> impl Iterator<Item = KVItem> + '_ {
+        let column_metrics = self.metrics.columns_read_statistic.get(&column.id());
         self.db
             .iterator_cf_opt(&self.cf(column), opts, iter_mode)
-            .map(|item| {
+            .map(move |item| {
                 item.map(|(key, value)| {
                     let value_as_vec = Vec::from(value);
                     let key_as_vec = Vec::from(key);
 
-                    database_metrics().read_meter.inc();
-                    database_metrics().bytes_read.observe(
-                        (key_as_vec.len().saturating_add(value_as_vec.len())) as f64,
+                    self.metrics.read_meter.inc();
+                    column_metrics.map(|metric| metric.inc());
+                    self.metrics.bytes_read.inc_by(
+                        key_as_vec.len().saturating_add(value_as_vec.len()) as u64,
                     );
 
                     (key_as_vec, Arc::new(value_as_vec))
@@ -516,12 +530,23 @@ where
         I: Iterator<Item = K>,
         K: AsRef<[u8]>,
     {
+        let column_metrics = self.metrics.columns_read_statistic.get(&column);
         let cl = self.cf_u32(column);
         let results = self
             .db
             .multi_get_cf_opt(iterator.map(|k| (&cl, k)), &self.read_options)
             .into_iter()
-            .map(|el| el.map_err(|err| DatabaseError::Other(err.into())))
+            .map(|el| {
+                self.metrics.read_meter.inc();
+                column_metrics.map(|metric| metric.inc());
+                el.map(|value| {
+                    value.map(|vec| {
+                        self.metrics.bytes_read.inc_by(vec.len() as u64);
+                        vec
+                    })
+                })
+                .map_err(|err| DatabaseError::Other(err.into()))
+            })
             .try_collect()?;
         Ok(results)
     }
@@ -538,7 +563,9 @@ where
         key: &[u8],
         column: Self::Column,
     ) -> StorageResult<Option<usize>> {
-        database_metrics().read_meter.inc();
+        self.metrics.read_meter.inc();
+        let column_metrics = self.metrics.columns_read_statistic.get(&column.id());
+        column_metrics.map(|metric| metric.inc());
 
         Ok(self
             .db
@@ -548,15 +575,17 @@ where
     }
 
     fn get(&self, key: &[u8], column: Self::Column) -> StorageResult<Option<Value>> {
-        database_metrics().read_meter.inc();
+        self.metrics.read_meter.inc();
+        let column_metrics = self.metrics.columns_read_statistic.get(&column.id());
+        column_metrics.map(|metric| metric.inc());
 
         let value = self
             .db
             .get_cf_opt(&self.cf(column), key, &self.read_options)
             .map_err(|e| DatabaseError::Other(e.into()))?;
 
         if let Some(value) = &value {
-            database_metrics().bytes_read.observe(value.len() as f64);
+            self.metrics.bytes_read.inc_by(value.len() as u64);
         }
 
         Ok(value.map(Arc::new))
@@ -568,7 +597,9 @@ where
         column: Self::Column,
         mut buf: &mut [u8],
     ) -> StorageResult<Option<usize>> {
-        database_metrics().read_meter.inc();
+        self.metrics.read_meter.inc();
+        let column_metrics = self.metrics.columns_read_statistic.get(&column.id());
+        column_metrics.map(|metric| metric.inc());
 
         let r = self
             .db
@@ -583,7 +614,7 @@ where
             .transpose()?;
 
         if let Some(r) = &r {
-            database_metrics().bytes_read.observe(*r as f64);
+            self.metrics.bytes_read.inc_by(*r as u64);
         }
 
         Ok(r)
@@ -678,13 +709,18 @@ where
     Description: DatabaseDescription,
 {
     pub fn commit_changes(&self, changes: &Changes) -> StorageResult<()> {
+        let instant = std::time::Instant::now();
         let mut batch = WriteBatch::default();
 
         for (column, ops) in changes {
             let cf = self.cf_u32(*column);
+            let column_metrics = self.metrics.columns_write_statistic.get(column);
             for (key, op) in ops {
+                self.metrics.write_meter.inc();
+                column_metrics.map(|metric| metric.inc());
                 match op {
                     WriteOperation::Insert(value) => {
+                        self.metrics.bytes_written.inc_by(value.len() as u64);
                         batch.put_cf(&cf, key, value.as_ref());
                     }
                     WriteOperation::Remove => {
@@ -694,14 +730,16 @@ where
             }
         }
 
-        database_metrics().write_meter.inc();
-        database_metrics()
-            .bytes_written
-            .observe(batch.size_in_bytes() as f64);
-
         self.db
             .write(batch)
-            .map_err(|e| DatabaseError::Other(e.into()).into())
+            .map_err(|e| DatabaseError::Other(e.into()))?;
+        // TODO: Use `u128` when `AtomicU128` is stable.
+        self.metrics.database_commit_time.inc_by(
+            u64::try_from(instant.elapsed().as_nanos())
+                .expect("The commit shouldn't take longer than `u64`"),
+        );
+
+        Ok(())
     }
 }
 

diff --git a/crates/metrics/Cargo.toml b/crates/metrics/Cargo.toml
@@ -11,8 +11,7 @@ repository = { workspace = true }
 description = "Fuel metrics"
 
 [dependencies]
-axum = { workspace = true }
-once_cell = { workspace = true }
+parking_lot = { workspace = true }
 pin-project-lite = { workspace = true }
 prometheus-client = { workspace = true }
 regex = "1"