add detailed metrics in slow operations

node/storage/src/log_store/flow_store.rs

@@ -1,13 +1,14 @@
-use super::load_chunk::EntryBatch;
-use super::seal_task_manager::SealTaskManager;
-use super::{MineLoadChunk, SealAnswer, SealTask};
 use crate::config::ShardConfig;
 use crate::error::Error;
+use crate::log_store::load_chunk::EntryBatch;
 use crate::log_store::log_manager::{
     bytes_to_entries, data_to_merkle_leaves, COL_ENTRY_BATCH, COL_ENTRY_BATCH_ROOT,
     COL_FLOW_MPT_NODES, ENTRY_SIZE, PORA_CHUNK_SIZE,
 };
-use crate::log_store::{FlowRead, FlowSeal, FlowWrite};
+use crate::log_store::seal_task_manager::SealTaskManager;
+use crate::log_store::{
+    metrics, FlowRead, FlowSeal, FlowWrite, MineLoadChunk, SealAnswer, SealTask,
+};
 use crate::{try_option, ZgsKeyValueDB};
 use any::Any;
 use anyhow::{anyhow, bail, Result};
@@ -22,6 +23,7 @@ use std::cmp::Ordering;
 use std::collections::BTreeMap;
 use std::fmt::Debug;
 use std::sync::Arc;
+use std::time::Instant;
 use std::{any, cmp, mem};
 use tracing::{debug, error, trace};
 use zgs_spec::{BYTES_PER_SECTOR, SEALS_PER_LOAD, SECTORS_PER_LOAD, SECTORS_PER_SEAL};
@@ -42,7 +44,11 @@ impl FlowStore {
     }
 
     pub fn put_batch_root_list(&self, root_map: BTreeMap<usize, (DataRoot, usize)>) -> Result<()> {
-        self.db.put_batch_root_list(root_map)
+        let start_time = Instant::now();
+        let res = self.db.put_batch_root_list(root_map);
+
+        metrics::PUT_BATCH_ROOT_LIST.update_since(start_time);
+        res
     }
 
     pub fn insert_subtree_list_for_batch(
@@ -50,6 +56,7 @@ impl FlowStore {
         batch_index: usize,
         subtree_list: Vec<(usize, usize, DataRoot)>,
     ) -> Result<()> {
+        let start_time = Instant::now();
         let mut batch = self
             .db
             .get_entry_batch(batch_index as u64)?
@@ -57,6 +64,8 @@ impl FlowStore {
         batch.set_subtree_list(subtree_list);
         self.db.put_entry_raw(vec![(batch_index as u64, batch)])?;
 
+        metrics::INSERT_SUBTREE_LIST.update_since(start_time);
+
         Ok(())
     }
 
@@ -75,7 +84,10 @@ impl FlowStore {
     }
 
     pub fn put_mpt_node_list(&self, node_list: Vec<(usize, usize, DataRoot)>) -> Result<()> {
-        self.db.put_mpt_node_list(node_list)
+        let start_time = Instant::now();
+        let res = self.db.put_mpt_node_list(node_list);
+        metrics::PUT_MPT_NODE.update_since(start_time);
+        res
     }
 
     pub fn delete_batch_list(&self, batch_list: &[u64]) -> Result<()> {

node/storage/src/log_store/log_manager.rs

@@ -934,6 +934,7 @@ impl LogManager {
     #[instrument(skip(self, merkle))]
     fn pad_tx(&self, tx_start_index: u64, merkle: &mut MerkleManager) -> Result<()> {
         // Check if we need to pad the flow.
+        let start_time = Instant::now();
         let mut tx_start_flow_index =
             merkle.last_chunk_start_index() + merkle.last_chunk_merkle.leaves() as u64;
         let pad_size = tx_start_index - tx_start_flow_index;
@@ -1022,6 +1023,8 @@ impl LogManager {
             merkle.pora_chunks_merkle.leaves(),
             merkle.last_chunk_merkle.leaves()
         );
+
+        metrics::PAD_TX.update_since(start_time);
         Ok(())
     }
 

node/storage/src/log_store/metrics.rs

@@ -5,10 +5,21 @@ use metrics::{register_timer, Timer};
 lazy_static::lazy_static! {
     pub static ref TX_STORE_PUT: Arc<dyn Timer> = register_timer("log_store_tx_store_put_tx");
 
+    pub static ref CHECK_TX_COMPLETED: Arc<dyn Timer> =
+        register_timer("log_store_log_manager_check_tx_completed");
 
     pub static ref APPEND_SUBTREE_LIST: Arc<dyn Timer> =
         register_timer("log_store_log_manager_append_subtree_list");
 
     pub static ref COPY_TX_AND_FINALIZE: Arc<dyn Timer> =
         register_timer("log_store_log_manager_copy_tx_and_finalize");
+
+    pub static ref PAD_TX: Arc<dyn Timer> = register_timer("log_store_log_manager_pad_tx");
+
+    pub static ref PUT_BATCH_ROOT_LIST: Arc<dyn Timer> = register_timer("log_store_flow_store_put_batch_root_list");
+
+    pub static ref INSERT_SUBTREE_LIST: Arc<dyn Timer> =
+        register_timer("log_store_log_manager_insert_subtree_list");
+
+    pub static ref PUT_MPT_NODE: Arc<dyn Timer> = register_timer("log_store_log_manager_put_mpt_node");
 }

node/storage/src/log_store/tx_store.rs

@@ -180,8 +180,12 @@ impl TransactionStore {
     }
 
     pub fn check_tx_completed(&self, tx_seq: u64) -> Result<bool> {
-        Ok(self.kvdb.get(COL_TX_COMPLETED, &tx_seq.to_be_bytes())?
-            == Some(vec![TX_STATUS_FINALIZED]))
+        let start_time = Instant::now();
+        let res = self.kvdb.get(COL_TX_COMPLETED, &tx_seq.to_be_bytes())?
+            == Some(vec![TX_STATUS_FINALIZED]);
+
+        metrics::CHECK_TX_COMPLETED.update_since(start_time);
+        Ok(res)
     }
 
     pub fn check_tx_pruned(&self, tx_seq: u64) -> Result<bool> {