feat(metrics): RocksDB write-stall gauges

crates/hotblocks/src/cli.rs

@@ -10,7 +10,7 @@ use sqd_storage::db::{DatabaseSettings, DatasetId};
 use crate::{
     data_service::{DataService, DataServiceRef},
     dataset_config::{DatasetConfig, RetentionConfig},
-    metrics::DatasetMetricsCollector,
+    metrics::{DatasetMetricsCollector, RocksDbCollector},
     query::{QueryService, QueryServiceRef},
     types::DBRef
 };
@@ -125,6 +125,7 @@ impl CLI {
             db: db.clone(),
             datasets: datasets.keys().copied().collect()
         }));
+        metrics_registry.register_collector(Box::new(RocksDbCollector { db: db.clone() }));
 
         let api_controlled_datasets = datasets
             .iter()

crates/hotblocks/src/metrics.rs

@@ -12,7 +12,7 @@ use prometheus_client::{
     },
     registry::Registry
 };
-use sqd_storage::db::{DatasetId, ReadSnapshot};
+use sqd_storage::db::{CF_CHUNKS, CF_TABLES, DatasetId, ReadSnapshot};
 use tracing::error;
 
 use crate::{query::QueryExecutorCollector, types::DBRef};
@@ -168,6 +168,105 @@ fn collect_dataset_metrics(
     Ok(())
 }
 
+/// RocksDB write-stall / LSM-backlog gauges, read from intrinsic properties on scrape.
+/// One shared WriteController: any CF's stall stops the whole DB, so DB-wide gauges are global.
+#[derive(Debug)]
+pub struct RocksDbCollector {
+    pub db: DBRef
+}
+
+// (metric, help, property) -- DB-global, queried on one CF.
+const ROCKSDB_DB_WIDE: &[(&str, &str, &str)] = &[
+    (
+        "hotblocks_rocksdb_write_stopped",
+        "1 while RocksDB has hard-stopped all writes",
+        "rocksdb.is-write-stopped"
+    ),
+    (
+        "hotblocks_rocksdb_actual_delayed_write_rate",
+        "Throttled write rate in bytes/s while writes are delayed (soft stall); 0 when not delayed",
+        "rocksdb.actual-delayed-write-rate"
+    ),
+    (
+        "hotblocks_rocksdb_running_compactions",
+        "Compactions currently running",
+        "rocksdb.num-running-compactions"
+    ),
+    (
+        "hotblocks_rocksdb_running_flushes",
+        "Memtable flushes currently running",
+        "rocksdb.num-running-flushes"
+    )
+];
+
+// (metric, help, property) -- per-cf backlog that trips a stall.
+const ROCKSDB_PER_CF: &[(&str, &str, &str)] = &[
+    (
+        "hotblocks_rocksdb_num_files_at_level0",
+        "SST files at L0 (stalls writes once it reaches level0_stop_writes_trigger)",
+        "rocksdb.num-files-at-level0"
+    ),
+    (
+        "hotblocks_rocksdb_estimate_pending_compaction_bytes",
+        "Estimated bytes pending compaction (stalls writes at the soft/hard limit)",
+        "rocksdb.estimate-pending-compaction-bytes"
+    ),
+    (
+        "hotblocks_rocksdb_num_immutable_mem_table",
+        "Immutable memtables not yet flushed",
+        "rocksdb.num-immutable-mem-table"
+    ),
+    (
+        "hotblocks_rocksdb_mem_table_flush_pending",
+        "1 while a memtable flush is pending",
+        "rocksdb.mem-table-flush-pending"
+    )
+];
+
+const ROCKSDB_DATA_CFS: &[&str] = &[CF_CHUNKS, CF_TABLES];
+
+#[derive(Clone, Hash, Debug, Eq, PartialEq, EncodeLabelSet)]
+struct CfLabel {
+    cf: &'static str
+}
+
+impl Collector for RocksDbCollector {
+    fn encode(&self, mut encoder: DescriptorEncoder) -> Result<(), std::fmt::Error> {
+        if let Err(err) = collect_rocksdb_metrics(&mut encoder, &self.db) {
+            return if err.is::<std::fmt::Error>() {
+                Err(err.downcast().unwrap())
+            } else {
+                error!(err =? err, "failed to collect rocksdb metrics");
+                Ok(())
+            };
+        }
+        Ok(())
+    }
+}
+
+fn collect_rocksdb_metrics(encoder: &mut DescriptorEncoder, db: &DBRef) -> anyhow::Result<()> {
+    for (metric, help, prop) in ROCKSDB_DB_WIDE {
+        if let Some(value) = db.get_int_property(CF_TABLES, prop)? {
+            encoder
+                .encode_descriptor(metric, help, None, MetricType::Gauge)?
+                .encode_gauge(&value)?;
+        }
+    }
+
+    for (metric, help, prop) in ROCKSDB_PER_CF {
+        for &cf in ROCKSDB_DATA_CFS {
+            if let Some(value) = db.get_int_property(cf, prop)? {
+                encoder
+                    .encode_descriptor(metric, help, None, MetricType::Gauge)?
+                    .encode_family(&CfLabel { cf })?
+                    .encode_gauge(&value)?;
+            }
+        }
+    }
+
+    Ok(())
+}
+
 pub fn build_metrics_registry() -> Registry {
     let mut top_registry = Registry::default();
     let registry = top_registry.sub_registry_with_prefix("hotblocks");

crates/storage/src/db/db.rs

@@ -396,6 +396,14 @@ impl Database {
         let val = self.db.property_value_cf(cf_handle, name)?;
         Ok(val)
     }
+
+    pub fn get_int_property(&self, cf: &str, name: &str) -> anyhow::Result<Option<u64>> {
+        let Some(cf_handle) = self.db.cf_handle(cf) else {
+            return Ok(None);
+        };
+        let val = self.db.property_int_value_cf(cf_handle, name)?;
+        Ok(val)
+    }
 }
 
 impl std::fmt::Debug for Database {

crates/storage/tests/rocksdb_properties.rs

@@ -0,0 +1,50 @@
+//! `get_int_property` backs the `hotblocks_rocksdb_*` gauges; it must work without
+//! `enable_statistics()` (prod default). Pins the property names -- a typo => silent None.
+
+use sqd_storage::db::{DatabaseSettings, CF_CHUNKS, CF_TABLES};
+
+const DB_WIDE_PROPS: &[&str] = &[
+    "rocksdb.is-write-stopped",
+    "rocksdb.actual-delayed-write-rate",
+    "rocksdb.num-running-compactions",
+    "rocksdb.num-running-flushes"
+];
+
+const PER_CF_PROPS: &[&str] = &[
+    "rocksdb.num-files-at-level0",
+    "rocksdb.estimate-pending-compaction-bytes",
+    "rocksdb.num-immutable-mem-table",
+    "rocksdb.mem-table-flush-pending"
+];
+
+#[test]
+fn int_properties_available_without_stats() {
+    let dir = tempfile::tempdir().unwrap();
+    // Default settings => rocksdb stats disabled, mirroring production.
+    let db = DatabaseSettings::default().open(dir.path()).unwrap();
+
+    for prop in DB_WIDE_PROPS {
+        assert!(
+            db.get_int_property(CF_TABLES, prop).unwrap().is_some(),
+            "db-wide property {prop} returned None"
+        );
+    }
+
+    for cf in [CF_CHUNKS, CF_TABLES] {
+        for prop in PER_CF_PROPS {
+            assert!(
+                db.get_int_property(cf, prop).unwrap().is_some(),
+                "property {prop} on cf {cf} returned None"
+            );
+        }
+    }
+
+    // A fresh database is not stalled.
+    assert_eq!(
+        db.get_int_property(CF_TABLES, "rocksdb.is-write-stopped").unwrap(),
+        Some(0)
+    );
+
+    // An unknown column family yields None rather than erroring.
+    assert_eq!(db.get_int_property("NOPE", "rocksdb.is-write-stopped").unwrap(), None);
+}