add metric for event cache and other missing metrics (#4863)

* add metric for event cache

* Fix sub metrics

* Add metrics for massa-db-worker change history sizes

* Add lru_cache_memory_usage metric

* Disable db metrics for tests

* Refactor db metrics & lock handling

* Review comment

* Add active history events len

---------

Co-authored-by: Leo-Besancon <lb@massa.net>

Author: modship

Cargo.lock

@@ -1099,6 +1099,7 @@ mod tests {
             max_versioning_elements_size: 100,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 100,
+            enable_metrics: false,
         };
         let db: ShareableMassaDBController = Arc::new(RwLock::new(
             Box::new(MassaDB::new(db_config)) as Box<(dyn MassaDBController + 'static)>,
@@ -1141,6 +1142,7 @@ mod tests {
             max_versioning_elements_size: 100,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 100,
+            enable_metrics: false,
         };
         let db: ShareableMassaDBController = Arc::new(RwLock::new(
             Box::new(MassaDB::new(db_config)) as Box<(dyn MassaDBController + 'static)>,
@@ -1203,6 +1205,7 @@ mod tests {
             max_versioning_elements_size: 100,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 100,
+            enable_metrics: false,
         };
         let db: ShareableMassaDBController = Arc::new(RwLock::new(
             Box::new(MassaDB::new(db_config)) as Box<(dyn MassaDBController + 'static)>,
@@ -1260,6 +1263,7 @@ mod tests {
             max_versioning_elements_size: 100,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 100,
+            enable_metrics: false,
         };
         let db: ShareableMassaDBController = Arc::new(RwLock::new(
             Box::new(MassaDB::new(db_config)) as Box<(dyn MassaDBController + 'static)>,
@@ -1317,6 +1321,7 @@ mod tests {
             max_versioning_elements_size: 100,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 100,
+            enable_metrics: false,
         };
         let db: ShareableMassaDBController = Arc::new(RwLock::new(
             Box::new(MassaDB::new(db_config)) as Box<(dyn MassaDBController + 'static)>,
@@ -1367,6 +1372,7 @@ mod tests {
             max_versioning_elements_size: 100,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 100,
+            enable_metrics: false,
         };
         let db: ShareableMassaDBController = Arc::new(RwLock::new(Box::new(MassaDB::new(
             db_config.clone(),

massa-async-pool/src/pool.rs

@@ -451,6 +451,7 @@ fn test_staying_connected_without_message_trigger_read_timeout() {
         max_versioning_elements_size: 100_000_000,
         thread_count: THREAD_COUNT,
         max_ledger_backups: 10,
+        enable_metrics: false,
     }))
         as Box<(dyn MassaDBController + 'static)>));
     let rolls_path = PathBuf::from_str("../massa-node/base_config/initial_rolls.json").unwrap();
@@ -549,6 +550,7 @@ fn test_staying_connected_pass_handshake_but_deadline_after() {
         max_versioning_elements_size: 100_000_000,
         thread_count: THREAD_COUNT,
         max_ledger_backups: 10,
+        enable_metrics: false,
     }))
         as Box<(dyn MassaDBController + 'static)>));
     let rolls_path = PathBuf::from_str("../massa-node/base_config/initial_rolls.json").unwrap();
@@ -647,6 +649,7 @@ fn test_staying_connected_pass_handshake_but_deadline_during_data_exchange() {
         max_versioning_elements_size: 100_000_000,
         thread_count: THREAD_COUNT,
         max_ledger_backups: 10,
+        enable_metrics: false,
     }))
         as Box<(dyn MassaDBController + 'static)>));
     let rolls_path = PathBuf::from_str("../massa-node/base_config/initial_rolls.json").unwrap();

massa-bootstrap/src/tests/binders.rs

@@ -60,6 +60,7 @@ impl TestUniverse for BootstrapClientTestUniverse {
             max_final_state_elements_size: MAX_BOOTSTRAP_FINAL_STATE_PARTS_SIZE as usize,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 10,
+            enable_metrics: false,
         }))
             as Box<(dyn MassaDBController + 'static)>));
         controllers

massa-bootstrap/src/tests/universe_client.rs

@@ -57,6 +57,7 @@ impl BootstrapServerForeignControllers {
             max_final_state_elements_size: MAX_BOOTSTRAP_FINAL_STATE_PARTS_SIZE as usize,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 10,
+            enable_metrics: false,
         }))
             as Box<(dyn MassaDBController + 'static)>));
         Self {

massa-bootstrap/src/tests/universe_server.rs

@@ -88,6 +88,9 @@ pub trait MassaDBController: Send + Sync + Debug {
         last_change_id: Option<Slot>,
     ) -> Result<StreamBatch<Slot>, MassaDBError>;
 
+    /// Get the total size of the change history and the change versioning history respectively
+    fn get_change_history_sizes(&self) -> (usize, usize);
+
     /// Used in test to compare a prebuilt ledger with a ledger that has been built by the code
     #[cfg(feature = "test-exports")]
     fn get_entire_database(&self) -> Vec<BTreeMap<Vec<u8>, Vec<u8>>>;

massa-db-exports/src/controller.rs

@@ -15,4 +15,6 @@ pub struct MassaDBConfig {
     pub thread_count: u8,
     /// Maximum number of ledger backups to keep
     pub max_ledger_backups: u64,
+    /// Enable metrics
+    pub enable_metrics: bool,
 }

massa-db-exports/src/settings.rs

@@ -54,6 +54,8 @@ pub struct RawMassaDB<
     pub change_id_deserializer: ChangeIDDeserializer,
     /// The current RocksDB batch of the database, in a Mutex to share it
     pub current_batch: Arc<Mutex<WriteBatch>>,
+    /// If metrics are enabled, we keep track of the size of the changes associated to each change_id
+    pub change_history_sizes: BTreeMap<ChangeID, (usize, usize)>,
 }
 
 impl<ChangeID, ChangeIDSerializer, ChangeIDDeserializer> std::fmt::Debug
@@ -389,15 +391,51 @@ where
             }
         }
 
-        // in versioning_changes, we have the data that we do not want to include in hash
-        // e.g everything that is not in 'Active' state (so hashes remain compatibles)
-        for (key, value) in versioning_changes.iter() {
-            if let Some(value) = value {
-                self.current_batch
-                    .lock()
-                    .put_cf(handle_versioning, key, value);
-            } else {
-                self.current_batch.lock().delete_cf(handle_versioning, key);
+        // If metrics are enabled, we keep track of the size of the changes (for state changes, and then for versioning changes)
+        if self.config.enable_metrics {
+            let changes_size;
+            let changes_versioning_size;
+            {
+                let mut current_batch_guard = self.current_batch.lock();
+                changes_size = current_batch_guard.size_in_bytes();
+
+                // in versioning_changes, we have the data that we do not want to include in hash
+                // e.g everything that is not in 'Active' state (so hashes remain compatibles)
+                for (key, value) in versioning_changes.iter() {
+                    if let Some(value) = value {
+                        current_batch_guard.put_cf(handle_versioning, key, value);
+                    } else {
+                        current_batch_guard.delete_cf(handle_versioning, key);
+                    }
+                }
+
+                changes_versioning_size = current_batch_guard
+                    .size_in_bytes()
+                    .saturating_sub(changes_size);
+            }
+
+            match self
+                .change_history_sizes
+                .entry(self.get_change_id().expect(CHANGE_ID_DESER_ERROR))
+            {
+                std::collections::btree_map::Entry::Vacant(entry) => {
+                    entry.insert((changes_size, changes_versioning_size));
+                }
+                std::collections::btree_map::Entry::Occupied(mut entry) => {
+                    entry.get_mut().0 += changes_size;
+                    entry.get_mut().1 += changes_versioning_size;
+                }
+            }
+        } else {
+            let mut current_batch_guard = self.current_batch.lock();
+            // in versioning_changes, we have the data that we do not want to include in hash
+            // e.g everything that is not in 'Active' state (so hashes remain compatibles)
+            for (key, value) in versioning_changes.iter() {
+                if let Some(value) = value {
+                    current_batch_guard.put_cf(handle_versioning, key, value);
+                } else {
+                    current_batch_guard.delete_cf(handle_versioning, key);
+                }
             }
         }
 
@@ -447,10 +485,12 @@ where
         if reset_history {
             self.change_history.clear();
             self.change_history_versioning.clear();
+            self.change_history_sizes.clear();
         }
 
         while self.change_history.len() > self.config.max_history_length {
             self.change_history.pop_first();
+            self.change_history_sizes.pop_first();
         }
 
         while self.change_history_versioning.len() > self.config.max_history_length {
@@ -614,6 +654,7 @@ impl RawMassaDB<Slot, SlotSerializer, SlotDeserializer> {
             config,
             change_history: BTreeMap::new(),
             change_history_versioning: BTreeMap::new(),
+            change_history_sizes: BTreeMap::new(),
             change_id_serializer: SlotSerializer::new(),
             change_id_deserializer,
             current_batch,
@@ -851,6 +892,15 @@ impl MassaDBController for RawMassaDB<Slot, SlotSerializer, SlotDeserializer> {
         self.get_versioning_batch_to_stream(last_versioning_step, last_change_id)
     }
 
+    fn get_change_history_sizes(&self) -> (usize, usize) {
+        self.change_history_sizes.values().fold(
+            (0, 0),
+            |(acc_state, acc_version), &(state, version)| {
+                (acc_state + state, acc_version + version)
+            },
+        )
+    }
+
     #[cfg(feature = "test-exports")]
     fn get_entire_database(&self) -> Vec<BTreeMap<Vec<u8>, Vec<u8>>> {
         let handle_state = self.db.cf_handle(STATE_CF).expect(CF_ERROR);
@@ -933,6 +983,7 @@ mod test {
             max_versioning_elements_size: 100,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 10,
+            enable_metrics: false,
         };
         let mut db_opts = MassaDB::default_db_opts();
         // Additional checks (only for testing)
@@ -963,6 +1014,7 @@ mod test {
             max_versioning_elements_size: 100,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 10,
+            enable_metrics: false,
         };
         let mut db_opts = MassaDB::default_db_opts();
         // Additional checks (only for testing)
@@ -1047,6 +1099,7 @@ mod test {
             max_versioning_elements_size: 100,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 10,
+            enable_metrics: false,
         };
         let mut db_opts = MassaDB::default_db_opts();
         // Additional checks (only for testing)
@@ -1132,6 +1185,7 @@ mod test {
             max_versioning_elements_size: 100,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 10,
+            enable_metrics: false,
         };
         let mut db_opts = MassaDB::default_db_opts();
         // Additional checks (only for testing)
@@ -1181,6 +1235,7 @@ mod test {
                 max_versioning_elements_size: 100,
                 thread_count: THREAD_COUNT,
                 max_ledger_backups: 10,
+                enable_metrics: false,
             };
             let mut db_backup_1_opts = MassaDB::default_db_opts();
             db_backup_1_opts.create_if_missing(false);
@@ -1205,6 +1260,7 @@ mod test {
                 max_versioning_elements_size: 100,
                 thread_count: THREAD_COUNT,
                 max_ledger_backups: 10,
+                enable_metrics: false,
             };
             let mut db_backup_2_opts = MassaDB::default_db_opts();
             db_backup_2_opts.create_if_missing(false);
@@ -1238,6 +1294,7 @@ mod test {
             max_versioning_elements_size: 100,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 10,
+            enable_metrics: false,
         };
         let mut db_opts = MassaDB::default_db_opts();
         // Additional checks (only for testing)
@@ -1285,6 +1342,7 @@ mod test {
                 max_versioning_elements_size: 100,
                 thread_count: THREAD_COUNT,
                 max_ledger_backups: 10,
+                enable_metrics: false,
             };
             // let db_backup_2_opts = MassaDB::default_db_opts();
 
@@ -1334,6 +1392,7 @@ mod test {
             max_versioning_elements_size: 100,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 10,
+            enable_metrics: false,
         };
         let mut db_opts = MassaDB::default_db_opts();
         // Additional checks (only for testing)
@@ -1427,6 +1486,7 @@ mod test {
             max_versioning_elements_size: 100,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 10,
+            enable_metrics: false,
         };
         let mut db_opts = MassaDB::default_db_opts();
         // Additional checks (only for testing)
@@ -1515,6 +1575,7 @@ mod test {
             max_versioning_elements_size: 10,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 10,
+            enable_metrics: false,
         };
         let mut db_opts = MassaDB::default_db_opts();
         // Additional checks (only for testing)
@@ -1600,6 +1661,7 @@ mod test {
             max_versioning_elements_size: 20,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 10,
+            enable_metrics: false,
         };
         let mut db_opts = MassaDB::default_db_opts();
         // Additional checks (only for testing)
@@ -1712,6 +1774,7 @@ mod test {
             max_versioning_elements_size: 20,
             thread_count: THREAD_COUNT,
             max_ledger_backups: 10,
+            enable_metrics: false,
         };
 
         let slot_1 = Slot::new(1, 0);

massa-db-worker/src/massa_db.rs

@@ -23,6 +23,7 @@ fn call_registry_apply_changes() {
         max_versioning_elements_size: 100,
         thread_count: THREAD_COUNT,
         max_ledger_backups: 100,
+        enable_metrics: false,
     };
     let call_id_serializer = DeferredCallIdSerializer::new();
     let db: ShareableMassaDBController = Arc::new(RwLock::new(
@@ -75,6 +76,7 @@ fn call_registry_get_slot_calls() {
         max_versioning_elements_size: 100,
         thread_count: THREAD_COUNT,
         max_ledger_backups: 100,
+        enable_metrics: false,
     };
     let call_id_serializer = DeferredCallIdSerializer::new();
     let db: ShareableMassaDBController = Arc::new(RwLock::new(

massa-deferred-calls/src/tests/mod.rs

@@ -4,30 +4,27 @@ version = "28.10.0"
 edition = "2021"
 
 [features]
-test-exports = [
-    "massa_models/test-exports",
-    "mockall",
-    "mockall_wrap"
-]
+test-exports = ["massa_models/test-exports", "mockall", "mockall_wrap"]
 
 
 [dependencies]
-nom = {workspace = true}
-rocksdb = {workspace = true}
-tracing = {workspace = true}
+nom = { workspace = true }
+rocksdb = { workspace = true }
+tracing = { workspace = true }
 parking_lot = { workspace = true }
 num_enum = { workspace = true }
-massa_models = {workspace = true}
-massa_serialization = {workspace = true}
-massa_time = {workspace = true}
-mockall = {workspace = true, optional = true}
-mockall_wrap = {workspace = true, optional = true}
+massa_metrics = { workspace = true }
+massa_models = { workspace = true }
+massa_serialization = { workspace = true }
+massa_time = { workspace = true }
+mockall = { workspace = true, optional = true }
+mockall_wrap = { workspace = true, optional = true }
 
 [dev-dependencies]
-tempfile = {workspace = true}
-serial_test = {workspace = true}
-more-asserts = {workspace = true}
-rand = {workspace = true}
-mockall = {workspace = true}
-mockall_wrap = {workspace = true}
+tempfile = { workspace = true }
+serial_test = { workspace = true }
+more-asserts = { workspace = true }
+rand = { workspace = true }
+mockall = { workspace = true }
+mockall_wrap = { workspace = true }
 massa_models = { workspace = true, features = ["test-exports"] }

massa-event-cache/Cargo.toml

@@ -59,6 +59,7 @@ impl EventCacheController for EventCacheControllerImpl {
         // lock input data
         let mut input_data = self.input_data.1.lock();
         input_data.events.extend(events);
+        massa_metrics::set_event_cache_vec_len(input_data.events.len());
         // Wake up the condvar in EventCacheWriterThread waiting for events
         self.input_data.0.notify_all();
     }