diff --git a/crates/hotblocks/src/cli.rs b/crates/hotblocks/src/cli.rs index 1db864a..6a06d5d 100644 --- a/crates/hotblocks/src/cli.rs +++ b/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() diff --git a/crates/hotblocks/src/metrics.rs b/crates/hotblocks/src/metrics.rs index d5e08c0..a36ef82 100644 --- a/crates/hotblocks/src/metrics.rs +++ b/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::() { + 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"); diff --git a/crates/storage/src/db/db.rs b/crates/storage/src/db/db.rs index 0773636..3cc8afe 100644 --- a/crates/storage/src/db/db.rs +++ b/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> { + 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 { diff --git a/crates/storage/tests/rocksdb_properties.rs b/crates/storage/tests/rocksdb_properties.rs new file mode 100644 index 0000000..93f8702 --- /dev/null +++ b/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); +}