(LedgerStore) Perf Metric for RocksDB Writes (#23951)

#### Summary of Changes
This PR implements the reporting of RocksDB write perf metrics to blockstore_rocksdb_write_perf
based on RocksDB's PerfContext.  The default sample rate is 10 in 1000, and the env arg SOLANA_METRICS_ROCKSDB_PERF_SAMPLES_IN_1K can control the sample rate.

ledger/src/blockstore_db.rs

@@ -780,6 +780,7 @@ pub trait ColumnMetrics {
         column_options: &Arc<LedgerColumnOptions>,
     );
     fn rocksdb_get_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str;
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str;
 }
 
 pub trait ColumnName {
@@ -884,6 +885,13 @@ impl ColumnMetrics for columns::TransactionStatus {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "transaction_status",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::TransactionStatus {
     const NAME: &'static str = TRANSACTION_STATUS_CF;
@@ -943,6 +951,13 @@ impl ColumnMetrics for columns::AddressSignatures {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "address_signatures",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::AddressSignatures {
     const NAME: &'static str = ADDRESS_SIGNATURES_CF;
@@ -992,6 +1007,13 @@ impl ColumnMetrics for columns::TransactionMemos {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "transaction_memos",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::TransactionMemos {
     const NAME: &'static str = TRANSACTION_MEMOS_CF;
@@ -1041,6 +1063,13 @@ impl ColumnMetrics for columns::TransactionStatusIndex {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "transaction_status_index",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::TransactionStatusIndex {
     const NAME: &'static str = TRANSACTION_STATUS_INDEX_CF;
@@ -1065,6 +1094,13 @@ impl ColumnMetrics for columns::Rewards {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "rewards",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::Rewards {
     const NAME: &'static str = REWARDS_CF;
@@ -1092,6 +1128,13 @@ impl ColumnMetrics for columns::Blocktime {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "blocktime",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::Blocktime {
     const NAME: &'static str = BLOCKTIME_CF;
@@ -1119,6 +1162,13 @@ impl ColumnMetrics for columns::PerfSamples {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "perf_samples",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::PerfSamples {
     const NAME: &'static str = PERF_SAMPLES_CF;
@@ -1146,6 +1196,13 @@ impl ColumnMetrics for columns::BlockHeight {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "block_height",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::BlockHeight {
     const NAME: &'static str = BLOCK_HEIGHT_CF;
@@ -1172,6 +1229,13 @@ impl ColumnMetrics for columns::ProgramCosts {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "program_costs",
+            column_options
+        )
+    }
 }
 
 impl ColumnName for columns::ProgramCosts {
@@ -1245,6 +1309,13 @@ impl ColumnMetrics for columns::ShredCode {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "shred_code",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::ShredCode {
     const NAME: &'static str = CODE_SHRED_CF;
@@ -1293,6 +1364,13 @@ impl ColumnMetrics for columns::ShredData {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "shred_data",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::ShredData {
     const NAME: &'static str = DATA_SHRED_CF;
@@ -1317,6 +1395,13 @@ impl ColumnMetrics for columns::Index {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "index",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::Index {
     const NAME: &'static str = INDEX_CF;
@@ -1344,6 +1429,13 @@ impl ColumnMetrics for columns::DeadSlots {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "dead_slots",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::DeadSlots {
     const NAME: &'static str = DEAD_SLOTS_CF;
@@ -1371,6 +1463,13 @@ impl ColumnMetrics for columns::DuplicateSlots {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "duplicate_slots",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::DuplicateSlots {
     const NAME: &'static str = DUPLICATE_SLOTS_CF;
@@ -1398,6 +1497,13 @@ impl ColumnMetrics for columns::Orphans {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "orphans",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::Orphans {
     const NAME: &'static str = ORPHANS_CF;
@@ -1425,6 +1531,13 @@ impl ColumnMetrics for columns::BankHash {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "bank_hash",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::BankHash {
     const NAME: &'static str = BANK_HASH_CF;
@@ -1452,6 +1565,13 @@ impl ColumnMetrics for columns::Root {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "root",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::Root {
     const NAME: &'static str = ROOT_CF;
@@ -1479,6 +1599,13 @@ impl ColumnMetrics for columns::SlotMeta {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "slot_meta",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::SlotMeta {
     const NAME: &'static str = META_CF;
@@ -1531,6 +1658,13 @@ impl ColumnMetrics for columns::ErasureMeta {
             column_options
         )
     }
+    fn rocksdb_put_perf_metric_header(column_options: &Arc<LedgerColumnOptions>) -> &'static str {
+        rocksdb_metric_header!(
+            "blockstore_rocksdb_write_perf,op=put",
+            "erasure_meta",
+            column_options
+        )
+    }
 }
 impl ColumnName for columns::ErasureMeta {
     const NAME: &'static str = ERASURE_META_CF;
@@ -2149,9 +2283,74 @@ mod rocks_metrics_utils {
             );
         });
     }
+    /// Reports the collected PerfContext and disables the PerfContext after
+    /// reporting.
+    pub fn report_write_perf_context(metric_header: &'static str) {
+        PER_THREAD_ROCKS_PERF_CONTEXT.with(|perf_context_cell| {
+            set_perf_stats(PerfStatsLevel::Disable);
+            let perf_context = perf_context_cell.borrow();
+            datapoint_info!(
+                metric_header,
+                // total nanos spent on writing to WAL
+                (
+                    "write_wal_nanos",
+                    perf_context.metric(PerfMetric::WriteWalTime) as i64,
+                    i64
+                ),
+                // total nanos spent on writing to mem tables
+                (
+                    "write_memtable_nanos",
+                    perf_context.metric(PerfMetric::WriteMemtableTime) as i64,
+                    i64
+                ),
+                // total nanos spent on delaying or throttling write
+                (
+                    "write_delay_nanos",
+                    perf_context.metric(PerfMetric::WriteDelayTime) as i64,
+                    i64
+                ),
+                // total nanos spent on writing a record, excluding the above four things
+                (
+                    "write_pre_and_post_process_nanos",
+                    perf_context.metric(PerfMetric::WritePreAndPostProcessTime) as i64,
+                    i64
+                ),
+                // time spent on acquiring DB mutex.
+                (
+                    "db_mutex_lock_nanos",
+                    perf_context.metric(PerfMetric::DbMutexLockNanos) as i64,
+                    i64
+                ),
+                // Time spent on waiting with a condition variable created with DB mutex.
+                (
+                    "db_condition_wait_nanos",
+                    perf_context.metric(PerfMetric::DbConditionWaitNanos) as i64,
+                    i64
+                ),
+                // Time spent on merge operator.
+                (
+                    "merge_operator_nanos_nanos",
+                    perf_context.metric(PerfMetric::MergeOperatorTimeNanos) as i64,
+                    i64
+                ),
+                // Time spent waiting on key locks in transaction lock manager.
+                (
+                    "key_lock_wait_nanos",
+                    perf_context.metric(PerfMetric::KeyLockWaitTime) as i64,
+                    i64
+                ),
+                // number of times acquiring a lock was blocked by another transaction.
+                (
+                    "key_lock_wait_count",
+                    perf_context.metric(PerfMetric::KeyLockWaitCount) as i64,
+                    i64
+                ),
+            );
+        });
+    }
 }
 use crate::blockstore_db::rocks_metrics_utils::{
-    maybe_collect_perf_context, report_read_perf_context,
+    maybe_collect_perf_context, report_read_perf_context, report_write_perf_context,
 };
 
 impl<C> LedgerColumn<C>
@@ -2174,10 +2373,17 @@ where
     }
 
     pub fn put(&self, key: C::Index, value: &C::Type) -> Result<()> {
+        let is_perf_context_enabled = maybe_collect_perf_context();
         let serialized_value = serialize(value)?;
 
-        self.backend
+        let result = self
-            .put_cf(self.handle(), &C::key(key), &serialized_value)
+            .backend
+            .put_cf(self.handle(), &C::key(key), &serialized_value);
+
+        if is_perf_context_enabled {
+            report_write_perf_context(C::rocksdb_put_perf_metric_header(&self.column_options));
+        }
+        result
     }
 
     pub fn delete(&self, key: C::Index) -> Result<()> {