Merge branch 'master' into thlorenz/subscription-metrics

* master:
  fix: move delete onto separate thread (#629)
  feat: return ledger + accountsdb metrics (#624)

Author: thlorenz

magicblock-api/src/tickers.rs

@@ -107,35 +107,6 @@ pub fn init_system_metrics_ticker(
     ledger: &Arc<Ledger>,
     accountsdb: &Arc<AccountsDb>,
     token: CancellationToken,
-) -> tokio::task::JoinHandle<()> {
-    fn try_set_ledger_storage_size(ledger: &Ledger) {
-        match ledger.storage_size() {
-            Ok(byte_size) => metrics::set_ledger_size(byte_size),
-            Err(err) => warn!("Failed to get ledger storage size: {:?}", err),
-        }
-    }
-    let ledger = ledger.clone();
-    tokio::task::spawn(async move {
-        loop {
-            tokio::select! {
-                _ = tokio::time::sleep(tick_duration) => {
-                    try_set_ledger_storage_size(&ledger);
-                },
-                _ = token.cancelled() => {
-                    break;
-                }
-            }
-        }
-    })
-}
-
-/*
-#[allow(unused_variables)]
-pub fn init_system_metrics_ticker_old(
-    tick_duration: Duration,
-    ledger: &Arc<Ledger>,
-    accountsdb: &Arc<AccountsDb>,
-    token: CancellationToken,
 ) -> tokio::task::JoinHandle<()> {
     fn try_set_ledger_counts(ledger: &Ledger) {
         macro_rules! try_set_ledger_count {
@@ -173,23 +144,28 @@ pub fn init_system_metrics_ticker_old(
             Err(err) => warn!("Failed to get ledger storage size: {:?}", err),
         }
     }
-    fn set_accounts_storage_size(bank: &Bank) {
-        let byte_size = bank.accounts_db_storage_size();
-        metrics::set_accounts_size(byte_size);
+    fn set_accounts_storage_size(accounts_db: &AccountsDb) {
+        let byte_size = accounts_db.storage_size();
+        metrics::set_accounts_size(byte_size.try_into().unwrap_or(i64::MAX));
     }
-    fn set_accounts_count(bank: &Bank) {
-        metrics::set_accounts_count(bank.accounts_db.get_accounts_count());
+    fn set_accounts_count(accounts_db: &AccountsDb) {
+        metrics::set_accounts_count(
+            accounts_db
+                .get_accounts_count()
+                .try_into()
+                .unwrap_or(i64::MAX),
+        );
     }
 
     let ledger = ledger.clone();
-    let bank = bank.clone();
+    let bank = accountsdb.clone();
     tokio::task::spawn(async move {
         loop {
             tokio::select! {
                 _ = tokio::time::sleep(tick_duration) => {
                     try_set_ledger_storage_size(&ledger);
                     set_accounts_storage_size(&bank);
-                    try_set_ledger_counts(&ledger);
+                    metrics::observe_columns_count_duration(|| try_set_ledger_counts(&ledger));
                     set_accounts_count(&bank);
                 },
                 _ = token.cancelled() => {
@@ -198,7 +174,4 @@ pub fn init_system_metrics_ticker_old(
             }
         }
     })
-
-    tokio::task::spawn(async move {})
 }
-*/

magicblock-ledger/src/database/columns.rs

@@ -129,6 +129,9 @@ pub trait Column {
     // first item in the key.
     fn as_index(slot: Slot) -> Self::Index;
     fn slot(index: Self::Index) -> Slot;
+    fn keep_all_on_compaction() -> bool {
+        false
+    }
 }
 
 pub trait ColumnName {
@@ -651,6 +654,11 @@ impl Column for AccountModDatas {
     fn as_index(slot: Slot) -> Self::Index {
         slot
     }
+
+    /// We don't clean AccountModData on compaction as it isn't slot based
+    fn keep_all_on_compaction() -> bool {
+        true
+    }
 }
 
 impl TypedColumn for AccountModDatas {

magicblock-ledger/src/database/compaction_filter.rs

@@ -7,7 +7,6 @@ use std::{
     },
 };
 
-use log::trace;
 use rocksdb::{
     compaction_filter::CompactionFilter,
     compaction_filter_factory::{
@@ -82,21 +81,17 @@ pub(crate) struct PurgedSlotFilter<C: Column + ColumnName> {
 impl<C: Column + ColumnName> CompactionFilter for PurgedSlotFilter<C> {
     fn filter(
         &mut self,
-        level: u32,
+        _level: u32,
         key: &[u8],
         _value: &[u8],
     ) -> CompactionDecision {
         use rocksdb::CompactionDecision::*;
-        trace!("CompactionFilter: triggered!");
+        if C::keep_all_on_compaction() {
+            return Keep;
+        }
 
         let slot_in_key = C::slot(C::index(key));
         if slot_in_key < self.oldest_slot {
-            trace!(
-                "CompactionFilter: removing key. level: {}, slot: {}",
-                level,
-                slot_in_key
-            );
-
             // It is safe to delete this key
             // since those slots were truncated anyway
             Remove

magicblock-ledger/src/database/ledger_column.rs

@@ -7,9 +7,9 @@ use std::{
 };
 
 use bincode::{deserialize, serialize};
-use log::{error, warn};
+use log::warn;
 use prost::Message;
-use rocksdb::{properties as RocksProperties, ColumnFamily};
+use rocksdb::{properties as RocksProperties, CStrLike, ColumnFamily};
 use serde::de::DeserializeOwned;
 
 use super::{
@@ -228,7 +228,7 @@ where
     /// [here](https://github.com/facebook/rocksdb/blob/08809f5e6cd9cc4bc3958dd4d59457ae78c76660/include/rocksdb/db.h#L654-L689).
     pub fn get_int_property(
         &self,
-        name: &'static std::ffi::CStr,
+        name: impl CStrLike,
     ) -> Result<i64, LedgerError> {
         self.backend.get_int_property_cf(self.handle(), name)
     }
@@ -277,23 +277,45 @@ where
         self.backend.flush_cf(self.handle())
     }
 
+    #[inline(always)]
+    fn is_sequential_cf<T: ColumnName>() -> bool {
+        matches!(
+            T::NAME,
+            crate::database::columns::Blocktime::NAME
+                | crate::database::columns::Blockhash::NAME
+                | crate::database::columns::PerfSamples::NAME
+                | crate::database::columns::AccountModDatas::NAME
+        )
+    }
+
+    /// Initialize value when current one is `DIRTY_VALUE`
+    /// Value isn't set if current doesn't equal `DIRTY_VALUE`
+    fn init_column_count_cache(&self) -> LedgerResult<()> {
+        let count = if Self::is_sequential_cf::<C>() {
+            get_column_count_sequential_column(self)? as i64
+        } else {
+            get_column_count_complex_column(self)? as i64
+        };
+
+        // We can ignore error here since it means value already initialized
+        let _ = self.entry_counter.compare_exchange(
+            DIRTY_COUNT,
+            count,
+            Ordering::AcqRel,
+            Ordering::Relaxed,
+        );
+
+        Ok(())
+    }
+
     pub fn count_column_using_cache(&self) -> LedgerResult<i64> {
         let cached = self.entry_counter.load(Ordering::Relaxed);
         if cached != DIRTY_COUNT {
-            return Ok(cached);
+            Ok(cached)
+        } else {
+            self.init_column_count_cache()?;
+            Ok(self.entry_counter.load(Ordering::Acquire))
         }
-
-        self
-            .iter(IteratorMode::Start)
-            .map(Iterator::count)
-            .map(|val| if val > i64::MAX as usize {
-                // NOTE: this value is only used for metrics/diagnostics and
-                // aside from the fact that we will never encounter this case,
-                // it is good enough to return i64::MAX
-                error!("Column {} count is too large: {} for metrics, returning max.", C::NAME, val);
-                i64::MAX
-            } else { val as i64 })
-            .inspect(|updated| self.entry_counter.store(*updated, Ordering::Relaxed))
     }
 
     /// Increases entries counter if it's not [`DIRTY_COUNT`]
@@ -540,6 +562,38 @@ where
     }
 }
 
+/// When column key format is sequentially incremented key, like: `SlotColumn`
+/// We can simplify extraction of count by calculating `LastKey - FirstKey + 1`
+fn get_column_count_sequential_column<C: Column + ColumnName>(
+    ledger_column: &LedgerColumn<C>,
+) -> LedgerResult<u64> {
+    let last_key = ledger_column.iter(IteratorMode::End)?.next();
+    let start_key = ledger_column.iter(IteratorMode::Start)?.next();
+    let count = match (start_key, last_key) {
+        (Some((start_key, _)), Some((last_key, _))) => {
+            let last_slot = C::slot(last_key);
+            let start_slot = C::slot(start_key);
+
+            last_slot - start_slot + 1
+        }
+        // Empty ColumnFamily
+        _ => 0,
+    };
+
+    Ok(count)
+}
+
+/// For complex columns, like: `AddressSignatures`
+/// We get column count using rocksdb's "estimate-num-keys" proprty
+/// Due to properies of how we use DB this value shall be ~correct on start
+fn get_column_count_complex_column<C: Column + ColumnName>(
+    ledger_column: &LedgerColumn<C>,
+) -> LedgerResult<u64> {
+    const ESTIMATE_NUM_KEYS: &str = "rocksdb.estimate-num-keys";
+
+    Ok(ledger_column.get_int_property(ESTIMATE_NUM_KEYS)? as u64)
+}
+
 /// Increases entries counter if it's not [`DIRTY_COUNT`]
 /// Otherwise just skips it until it is set
 pub fn try_increase_entry_counter(entry_counter: &AtomicI64, by: u64) {

magicblock-ledger/src/database/rocks_db.rs

@@ -8,7 +8,7 @@ use std::{
 };
 
 use rocksdb::{
-    AsColumnFamilyRef, ColumnFamily, DBIterator, DBPinnableSlice,
+    AsColumnFamilyRef, CStrLike, ColumnFamily, DBIterator, DBPinnableSlice,
     DBRawIterator, FlushOptions, IteratorMode as RocksIteratorMode, LiveFile,
     Options, WriteBatch as RWriteBatch, DB,
 };
@@ -237,7 +237,7 @@ impl Rocks {
     pub fn get_int_property_cf(
         &self,
         cf: &ColumnFamily,
-        name: &'static std::ffi::CStr,
+        name: impl CStrLike,
     ) -> LedgerResult<i64> {
         match self.db.property_int_value_cf(cf, name) {
             Ok(Some(value)) => Ok(value.try_into().unwrap()),

magicblock-metrics/src/metrics/mod.rs

@@ -82,6 +82,20 @@ lazy_static::lazy_static! {
     static ref LEDGER_ACCOUNT_MOD_DATA_GAUGE: IntGauge = IntGauge::new(
         "ledger_account_mod_data_gauge", "Ledger Account Mod Data Gauge",
     ).unwrap();
+    pub static ref LEDGER_COLUMNS_COUNT_DURATION_SECONDS: Histogram = Histogram::with_opts(
+        HistogramOpts::new(
+            "ledger_columns_count_duration_seconds",
+            "Time taken to compute ledger columns counts"
+        )
+        .buckets(
+            MICROS_10_90.iter().chain(
+            MICROS_100_900.iter()).chain(
+            MILLIS_1_9.iter()).chain(
+            MILLIS_10_90.iter()).chain(
+            MILLIS_100_900.iter()).chain(
+            SECONDS_1_9.iter()).cloned().collect()
+        ),
+    ).unwrap();
 
     // -----------------
     // Accounts
@@ -235,14 +249,11 @@ lazy_static::lazy_static! {
 
     static ref COMMITTOR_INTENT_EXECUTION_TIME_HISTOGRAM: HistogramVec = HistogramVec::new(
         HistogramOpts::new(
-            "committor_intent_execution_time_histogram",
+            "committor_intent_execution_time_histogram_v2",
             "Time in seconds spent on intent execution"
         )
         .buckets(
-            MILLIS_1_9.iter()
-            .chain(MILLIS_10_90.iter())
-            .chain(MILLIS_100_900.iter())
-            .chain(SECONDS_1_9.iter()).cloned().collect(),
+            vec![0.01, 0.1, 1.0, 3.0, 5.0, 10.0, 15.0, 20.0, 25.0]
         ),
         &["intent_kind", "outcome_kind"],
     ).unwrap();
@@ -276,6 +287,7 @@ pub(crate) fn register() {
         register!(LEDGER_TRANSACTION_MEMOS_GAUGE);
         register!(LEDGER_PERF_SAMPLES_GAUGE);
         register!(LEDGER_ACCOUNT_MOD_DATA_GAUGE);
+        register!(LEDGER_COLUMNS_COUNT_DURATION_SECONDS);
         register!(ACCOUNTS_SIZE_GAUGE);
         register!(ACCOUNTS_COUNT_GAUGE);
         register!(PENDING_ACCOUNT_CLONES_GAUGE);
@@ -358,6 +370,21 @@ pub fn set_ledger_account_mod_data_count(count: i64) {
     LEDGER_ACCOUNT_MOD_DATA_GAUGE.set(count);
 }
 
+pub fn observe_columns_count_duration<F, T>(f: F) -> T
+where
+    F: FnOnce() -> T,
+{
+    LEDGER_COLUMNS_COUNT_DURATION_SECONDS.observe_closure_duration(f)
+}
+
+pub fn set_accounts_size(value: i64) {
+    ACCOUNTS_SIZE_GAUGE.set(value)
+}
+
+pub fn set_accounts_count(value: i64) {
+    ACCOUNTS_COUNT_GAUGE.set(value)
+}
+
 pub fn inc_pending_clone_requests() {
     PENDING_ACCOUNT_CLONES_GAUGE.inc()
 }