diff --git a/nodedb-mem/src/governor/metrics.rs b/nodedb-mem/src/governor/metrics.rs index 15a1ba696e..ddd470c132 100644 --- a/nodedb-mem/src/governor/metrics.rs +++ b/nodedb-mem/src/governor/metrics.rs @@ -5,11 +5,25 @@ use std::sync::atomic::Ordering; +use nodedb_types::DatabaseId; + use super::core::MemoryGovernor; use crate::engine::EngineId; use crate::pressure::{PressureLevel, PressureThresholds}; impl MemoryGovernor { + /// Current allocated memory in bytes for a database. + /// + /// Returns 0 if the database has no active allocations. + pub fn database_usage_bytes(&self, db: DatabaseId) -> usize { + self.database_budgets + .read() + .unwrap_or_else(|p| p.into_inner()) + .get(&db) + .map(|b| b.allocated.load(Ordering::Relaxed)) + .unwrap_or(0) + } + /// Total memory allocated across all engines (engine-layer sum). A /// separate aggregate from [`global_utilization_percent`](Self::global_utilization_percent), /// which reads the global counter admission enforces the ceiling against. @@ -164,4 +178,16 @@ mod tests { assert_eq!(gov.engine_pressure(EngineId::Vector), PressureLevel::Normal); assert_eq!(gov.worst_engine_pressure(), PressureLevel::Critical); } + + #[test] + fn database_usage_bytes_tracks_allocation() { + let gov = MemoryGovernor::new(test_config()).unwrap(); + gov.set_database_budget(db(), 10_000); + assert_eq!(gov.database_usage_bytes(db()), 0); + + let _tok = gov + .try_reserve(db(), tenant(), EngineId::Vector, 512) + .unwrap(); + assert_eq!(gov.database_usage_bytes(db()), 512); + } } diff --git a/nodedb-mem/src/governor/reserve.rs b/nodedb-mem/src/governor/reserve.rs index 1b3cd5dfb7..ea63288f04 100644 --- a/nodedb-mem/src/governor/reserve.rs +++ b/nodedb-mem/src/governor/reserve.rs @@ -13,6 +13,7 @@ use crate::error::{MemError, Result}; use crate::over_release::ReleaseIdentity; use crate::reservation_token::{ReservationParams, ReservationToken}; use crate::reserve_scope::{ReserveScope, ReservedLayers}; +use crate::scoped_budget::ScopedBudget; /// Build the token both entry points return. They differ in how they reach /// a committed [`ReservedLayers`], never in what they build from one. @@ -36,7 +37,28 @@ fn token_from_layers( } impl MemoryGovernor { - // ── 4-arity reservation ─────────────────────────────────────────────────── + /// Run `f` with the database's `ScopedBudget`, inserting an uncapped one if absent. + /// + /// Takes a read lock for existing entries to avoid contending with writer locks. + /// Takes a write lock only on missing entries. + fn with_database_budget(&self, db: DatabaseId, f: impl FnOnce(&ScopedBudget) -> R) -> R { + let read_map = self + .database_budgets + .read() + .unwrap_or_else(|p| p.into_inner()); + if let Some(budget) = read_map.get(&db) { + return f(budget); + } + drop(read_map); + let mut write_map = self + .database_budgets + .write() + .unwrap_or_else(|p| p.into_inner()); + let budget = write_map + .entry(db) + .or_insert_with(|| ScopedBudget::new(None)); + f(budget) + } /// Reserve `size` bytes for one (database, tenant, engine) triple. /// @@ -61,25 +83,18 @@ impl MemoryGovernor { scope.try_credit_global()?; - { - let map = self - .database_budgets - .read() - .unwrap_or_else(|p| p.into_inner()); - if let Some(budget) = map.get(&db) { - match budget.try_reserve(size) { - Ok(arc) => scope.credit_database(arc), - Err(denied) => { - return Err(MemError::DatabaseBudgetExhausted { - db, - requested: size, - available: budget.available(), - limit: denied.limit, - }); - } - } + self.with_database_budget(db, |budget| match budget.try_reserve(size) { + Ok(arc) => { + scope.credit_database(arc); + Ok(()) } - } + Err(denied) => Err(MemError::DatabaseBudgetExhausted { + db, + requested: size, + available: budget.available(), + limit: denied.limit, + }), + })?; { let map = self @@ -137,15 +152,9 @@ impl MemoryGovernor { scope.credit_global_unchecked(); - { - let map = self - .database_budgets - .read() - .unwrap_or_else(|p| p.into_inner()); - if let Some(budget) = map.get(&db) { - scope.credit_database(budget.credit(size)); - } - } + self.with_database_budget(db, |budget| { + scope.credit_database(budget.credit(size)); + }); { let map = self @@ -493,4 +502,18 @@ mod tests { tok.err() ); } + + #[test] + fn uncapped_database_usage_tracks_allocation() { + let gov = MemoryGovernor::new(test_config()).unwrap(); + assert_eq!(gov.database_usage_bytes(db()), 0); + + let tok = gov + .try_reserve(db(), tenant(), EngineId::Vector, 512) + .unwrap(); + assert_eq!(gov.database_usage_bytes(db()), 512); + + drop(tok); + assert_eq!(gov.database_usage_bytes(db()), 0); + } } diff --git a/nodedb-mem/src/scoped_budget.rs b/nodedb-mem/src/scoped_budget.rs index d756898edc..f1034c6e6f 100644 --- a/nodedb-mem/src/scoped_budget.rs +++ b/nodedb-mem/src/scoped_budget.rs @@ -25,7 +25,7 @@ pub(crate) struct BudgetDenied { } impl ScopedBudget { - fn new(limit: Option) -> Self { + pub(crate) fn new(limit: Option) -> Self { Self { limit, allocated: Arc::new(AtomicUsize::new(0)), diff --git a/nodedb/src/bootstrap/background_loops/spawn.rs b/nodedb/src/bootstrap/background_loops/spawn.rs index 99c615ab6d..4f90f5eee7 100644 --- a/nodedb/src/bootstrap/background_loops/spawn.rs +++ b/nodedb/src/bootstrap/background_loops/spawn.rs @@ -60,6 +60,7 @@ pub fn spawn_background_loops( maintenance::spawn_pending_leave_cleanup(shared); maintenance::spawn_orphaned_drain_sweep(shared); timers::spawn_tenant_timers(shared); + crate::bootstrap::database_metrics_sampler::spawn_database_metrics_sampler(shared); spawn_security_loops(shared); // Checkpoint manager. It is handed the Event Plane's watermark store diff --git a/nodedb/src/bootstrap/database_metrics_sampler.rs b/nodedb/src/bootstrap/database_metrics_sampler.rs new file mode 100644 index 0000000000..6644167cc0 --- /dev/null +++ b/nodedb/src/bootstrap/database_metrics_sampler.rs @@ -0,0 +1,200 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! Background sampler for per-database Prometheus metrics. +//! +//! Samples live gauges every 10 seconds across memory, admission, storage, +//! virtual queue depths, and rolling-window WAL fsync latency. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use tracing::info; + +use crate::control::metrics::histogram::HistogramSnapshot; +use crate::control::state::SharedState; + +/// Spawn the 10-second periodic database metrics sampler. +pub fn spawn_database_metrics_sampler(shared: &Arc) { + let shared_sampler = Arc::clone(shared); + crate::control::shutdown::spawn_loop( + &shared.loop_registry, + &shared.shutdown, + "database_metrics_sampler", + crate::control::shutdown::ShutdownPhase::DrainingControlPlane, + move |mut shutdown| async move { + let mut tick = tokio::time::interval(Duration::from_secs(10)); + let mut last_fsync_snapshot: HistogramSnapshot = shared_sampler + .system_metrics + .as_ref() + .map(|s| s.wal_fsync_seconds.snapshot_counts()) + .unwrap_or_default(); + + loop { + tokio::select! { + _ = shutdown.wait_cancelled() => break, + _ = tick.tick() => {} + } + if shutdown.is_cancelled() { + break; + } + let catalog = shared_sampler.credentials.catalog(); + let databases = match catalog.list_databases() { + Ok(d) => d, + Err(e) => { + tracing::warn!(error = %e, "database_metrics_sampler: catalog list error"); + continue; + } + }; + + // Compute rolling P99 fsync latency across the 10s tick window. + let curr_fsync_snapshot = shared_sampler + .system_metrics + .as_ref() + .map(|s| s.wal_fsync_seconds.snapshot_counts()) + .unwrap_or_default(); + let wal_latency_p99 = + last_fsync_snapshot.delta_percentile(&curr_fsync_snapshot, 0.99); + last_fsync_snapshot = curr_fsync_snapshot; + + // Lock dispatcher ONCE per sampler tick to read virtual-queue depths for all databases. + let queue_depths: HashMap = { + let d = shared_sampler + .dispatcher + .lock() + .unwrap_or_else(|p| p.into_inner()); + databases + .iter() + .map(|db| (db.id.as_u64(), d.virtual_queue_depth(db.id.as_u64()))) + .collect() + }; + + for db in databases { + let depth = queue_depths.get(&db.id.as_u64()).copied().unwrap_or(0); + sample_database(&shared_sampler, db.id, &db.name, depth, wal_latency_p99); + } + } + }, + ); + info!("database metrics sampler running"); +} + +/// Sample and record gauges for a single database (internal helper, not pub). +fn sample_database( + shared: &SharedState, + db_id: nodedb_types::DatabaseId, + db_name: &str, + bridge_queue_depth: u64, + wal_latency_p99: u64, +) { + // 1. connections: active connections holding an admission permit for this database + let connections = shared + .admission_registry + .database_live_connections(db_id) + .map(u64::from) + .unwrap_or(0); + shared + .database_metrics + .set_connections(db_id, db_name, connections); + + // 2. memory: resident memory in bytes allocated by this database in governor + let memory_bytes = shared.governor.database_usage_bytes(db_id) as u64; + shared + .database_metrics + .set_memory_bytes(db_id, db_name, memory_bytes); + + // 3. storage: real on-disk storage usage from system metrics if present + if let Some(m) = &shared.system_metrics { + let storage_bytes = m.database_storage_bytes(db_name); + shared + .database_metrics + .set_storage_bytes(db_id, db_name, storage_bytes); + } + + // 4. bridge_queue_depth: sum of SPSC bridge virtual-queue depths across cores + shared + .database_metrics + .set_bridge_queue_depth(db_id, db_name, bridge_queue_depth); + + // 5. wal_latency_p99: node-wide rolling 10s P99 group-commit fsync latency + shared + .database_metrics + .set_wal_latency_p99(db_id, db_name, wal_latency_p99); + + // 6. maintenance: cumulative maintenance CPU seconds tracked by MaintenanceBudgetTracker + let maintenance_cpu = shared.maintenance_budget.cumulative_cpu_secs(db_id); + shared + .database_metrics + .set_maintenance_cpu_secs(db_id, db_name, maintenance_cpu); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::bridge::dispatch::Dispatcher; + use crate::control::metrics::histogram::HistogramSnapshot; + use crate::control::state::SharedState; + use crate::wal::WalManager; + use nodedb_mem::engine::EngineId; + use nodedb_types::{DatabaseId, TenantId}; + + #[tokio::test] + async fn sampler_records_memory_connections_and_wal_fsync() { + let dir = tempfile::tempdir().expect("tempdir"); + let wal = Arc::new( + WalManager::open_for_testing(&dir.path().join("sampler-test.wal")).expect("open wal"), + ); + let (dispatcher, _data_sides) = Dispatcher::new(1, 64); + let shared = SharedState::new(dispatcher, wal).expect("construct state"); + + if let Some(metrics) = shared.system_metrics.as_ref() { + for _ in 0..90 { + metrics.record_wal_fsync(100); + } + for _ in 0..10 { + metrics.record_wal_fsync(500); + } + } + + let db_id = DatabaseId::new(42); + let db_name = "test_analytics"; + + // Reserve memory in governor for db_id. + let _token = shared + .governor + .try_reserve(db_id, TenantId::new(1), EngineId::Vector, 8192) + .expect("reserve memory"); + + // Acquire connection permit for db_id. + let _permit = shared + .admission_registry + .try_acquire_database(db_id) + .expect("acquire permit"); + + let curr_snapshot = shared + .system_metrics + .as_ref() + .map(|s| s.wal_fsync_seconds.snapshot_counts()) + .unwrap_or_default(); + let delta_p99 = HistogramSnapshot::default().delta_percentile(&curr_snapshot, 0.99); + + // Run sample_database. + sample_database(&shared, db_id, db_name, 5, delta_p99); + + // Verify gauge outputs in DatabaseMetricsRegistry. + let mut text = String::new(); + shared.database_metrics.render_prometheus(&mut text); + assert!(text.contains(&format!( + "nodedb_database_connections{{database=\"{db_name}\"}} 1" + ))); + assert!(text.contains(&format!( + "nodedb_database_memory_used_bytes{{database=\"{db_name}\"}} 8192" + ))); + assert!(text.contains(&format!( + "nodedb_database_bridge_queue_depth{{database=\"{db_name}\"}} 5" + ))); + assert!(text.contains(&format!( + "nodedb_database_wal_commit_latency_p99_us{{database=\"{db_name}\"}} 500" + ))); + } +} diff --git a/nodedb/src/bootstrap/mod.rs b/nodedb/src/bootstrap/mod.rs index b1a5177f9b..85a5fef43f 100644 --- a/nodedb/src/bootstrap/mod.rs +++ b/nodedb/src/bootstrap/mod.rs @@ -9,6 +9,7 @@ pub mod core_stall_monitor; pub mod credentials; pub mod data_group_recovery; pub mod data_plane; +pub mod database_metrics_sampler; pub mod diagnostics; pub mod jwks_registry; pub mod listeners; diff --git a/nodedb/src/bridge/dispatch/dispatcher.rs b/nodedb/src/bridge/dispatch/dispatcher.rs index 984165374c..8d38d0ade9 100644 --- a/nodedb/src/bridge/dispatch/dispatcher.rs +++ b/nodedb/src/bridge/dispatch/dispatcher.rs @@ -259,6 +259,14 @@ impl Dispatcher { .unwrap_or(PressureState::Normal) } + /// Sum of bridge virtual-queue depths for a database across all cores. + pub fn virtual_queue_depth(&self, database_id: u64) -> u64 { + self.cores + .iter() + .map(|c| c.wfq.depth_for(database_id) as u64) + .sum() + } + /// Number of Data Plane cores. pub fn num_cores(&self) -> usize { self.cores.len() @@ -276,3 +284,55 @@ impl Dispatcher { &self.router } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::bridge::dispatch::test_requests::make_request_for_db; + + #[test] + fn virtual_queue_depth_reporting() { + let (mut dispatcher, data_sides) = Dispatcher::new(2, 2); + assert_eq!(dispatcher.virtual_queue_depth(1), 0); + + // Fill physical rings on core 0 and core 1 so subsequent requests park in WFQ. + dispatcher + .dispatch_to_core(0, make_request_for_db(0, 1, 1)) + .unwrap(); + dispatcher + .dispatch_to_core(0, make_request_for_db(0, 1, 2)) + .unwrap(); + assert_eq!(data_sides[0].request_rx.len(), 2); + + dispatcher + .dispatch_to_core(1, make_request_for_db(1, 1, 10)) + .unwrap(); + dispatcher + .dispatch_to_core(1, make_request_for_db(1, 1, 11)) + .unwrap(); + assert_eq!(data_sides[1].request_rx.len(), 2); + + // Park 2 requests in core 0's WFQ for db 1. + dispatcher + .dispatch_to_core(0, make_request_for_db(0, 1, 3)) + .unwrap(); + dispatcher + .dispatch_to_core(0, make_request_for_db(0, 1, 4)) + .unwrap(); + + // Park 3 requests in core 1's WFQ for db 1. + dispatcher + .dispatch_to_core(1, make_request_for_db(1, 1, 12)) + .unwrap(); + dispatcher + .dispatch_to_core(1, make_request_for_db(1, 1, 13)) + .unwrap(); + dispatcher + .dispatch_to_core(1, make_request_for_db(1, 1, 14)) + .unwrap(); + + // Virtual queue depth must sum across all cores. + assert_eq!(dispatcher.virtual_queue_depth(1), 5); + assert_eq!(dispatcher.virtual_queue_depth(2), 0); + } +} diff --git a/nodedb/src/control/catalog_entry/post_apply/database.rs b/nodedb/src/control/catalog_entry/post_apply/database.rs index 16a8879cf6..2a3f1fd7ba 100644 --- a/nodedb/src/control/catalog_entry/post_apply/database.rs +++ b/nodedb/src/control/catalog_entry/post_apply/database.rs @@ -1,10 +1,10 @@ // SPDX-License-Identifier: BUSL-1.1 -//! Post-apply side effects for database catalog entries. +//! Synchronous post-apply side effects for database catalog entries. //! //! Database descriptors and grants are read directly from redb on the hot //! path (no separate in-memory registry), so most arms are no-ops. The -//! synchronous `DeleteDatabase` arm releases the dropped scope's quota caps. +//! `DeleteDatabase` arm releases the dropped scope's quota caps on every node. use std::sync::Arc; @@ -16,10 +16,18 @@ use crate::control::state::SharedState; /// Post-apply for `PutDatabase` — no in-memory cache to update. pub fn put(_descriptor: DatabaseDescriptor, _shared: Arc) {} -/// Synchronous post-apply for `DeleteDatabase` — release the quota caps of -/// the dropped scope, its tenants' caps included. +/// Post-apply for `DeleteDatabase` — release the quota caps of the dropped +/// scope, its tenants' caps included, unregister from maintenance budget, +/// and remove any metrics handles from `DatabaseMetricsRegistry`. pub fn delete(db_id: u64, shared: Arc) { - super::quota::release_database_scope(DatabaseId::new(db_id), &shared); + let db = DatabaseId::new(db_id); + if let Some((name, _)) = shared.database_metrics.remove(db) { + if let Some(m) = &shared.system_metrics { + m.remove_database(&name); + } + } + shared.maintenance_budget.remove_database(db); + super::quota::release_database_scope(db, &shared); } /// Post-apply for `PutDatabaseGrant`. diff --git a/nodedb/src/control/maintenance/budget.rs b/nodedb/src/control/maintenance/budget.rs index 6971649480..df7c691488 100644 --- a/nodedb/src/control/maintenance/budget.rs +++ b/nodedb/src/control/maintenance/budget.rs @@ -31,6 +31,8 @@ struct DbWindow { buckets: [Bucket; 60], /// UNIX-style second index of the last bucket written (wraps at 60). last_bucket_secs: u64, + /// Cumulative CPU-seconds consumed since process start. + cumulative_cpu_secs: f64, } impl DbWindow { @@ -38,6 +40,7 @@ impl DbWindow { Self { buckets: std::array::from_fn(|_| Bucket::default()), last_bucket_secs: 0, + cumulative_cpu_secs: 0.0, } } @@ -64,6 +67,9 @@ impl DbWindow { self.advance(now_secs); let idx = (now_secs % 60) as usize; self.buckets[idx].consumed_secs += secs; + if secs.is_finite() && secs > 0.0 { + self.cumulative_cpu_secs += secs; + } } } @@ -115,6 +121,23 @@ impl MaintenanceBudgetTracker { } } + /// Cumulative maintenance CPU-seconds consumed by `db`. + pub fn cumulative_cpu_secs(&self, db: DatabaseId) -> f64 { + let inner = self.inner.lock().unwrap_or_else(|p| p.into_inner()); + inner + .windows + .get(&db) + .map(|w| w.cumulative_cpu_secs) + .unwrap_or(0.0) + } + + /// Remove a database's budget window and cap (e.g. on database drop). + pub fn remove_database(&self, db: DatabaseId) { + let mut inner = self.inner.lock().unwrap_or_else(|p| p.into_inner()); + inner.caps.remove(&db); + inner.windows.remove(&db); + } + /// Install or replace the maintenance CPU cap for `db`. /// /// `maintenance_cpu_pct` is the `QuotaRecord` field (0–100). @@ -129,6 +152,11 @@ impl MaintenanceBudgetTracker { inner.caps.insert(db, cap); } + /// Install or replace the maintenance CPU cap and register the database name for `db`. + pub fn set_cap_named(&self, db: DatabaseId, _name: impl Into, maintenance_cpu_pct: u8) { + self.set_cap(db, maintenance_cpu_pct); + } + /// Attempt to acquire a maintenance lease for `db`. /// /// Returns `Some(MaintenanceLease)` when `consumed + estimated_secs ≤ cap` @@ -171,13 +199,20 @@ impl Default for MaintenanceBudgetTracker { /// RAII lease returned by [`MaintenanceBudgetTracker::try_acquire`]. /// /// On drop, the actual elapsed wall-clock seconds are recorded into the -/// sliding window for the database. +/// sliding window and cumulative total for the database. pub struct MaintenanceLease { tracker: Arc, db: DatabaseId, start: Instant, } +impl MaintenanceLease { + /// Database id for which the lease was acquired. + pub fn db(&self) -> DatabaseId { + self.db + } +} + impl Drop for MaintenanceLease { fn drop(&mut self) { let elapsed = self.start.elapsed().as_secs_f64(); @@ -270,6 +305,7 @@ mod tests { assert!(total > 0.0, "lease drop should have recorded elapsed time"); assert!(total < 1.0, "elapsed should be under 1s"); } + assert!(t.cumulative_cpu_secs(db) > 0.0); } #[test] @@ -294,4 +330,21 @@ mod tests { "old consumption should have expired out of the window" ); } + + #[test] + fn cumulative_cpu_secs_increments() { + let t = tracker(); + let db = DatabaseId::new(42); + t.set_cap(db, 100); + + assert_eq!(t.cumulative_cpu_secs(db), 0.0); + { + let _lease = t.try_acquire(db, 1.0).expect("acquire should succeed"); + std::thread::sleep(std::time::Duration::from_millis(15)); + } + assert!( + t.cumulative_cpu_secs(db) > 0.0, + "cumulative maintenance CPU time should be > 0 after lease drop" + ); + } } diff --git a/nodedb/src/control/metrics/database.rs b/nodedb/src/control/metrics/database.rs index 75abd4a5d4..68b5583712 100644 --- a/nodedb/src/control/metrics/database.rs +++ b/nodedb/src/control/metrics/database.rs @@ -30,9 +30,10 @@ pub struct DatabaseCounters { pub connections: AtomicU64, /// SPSC bridge virtual-queue depth snapshot. pub bridge_queue_depth: AtomicU64, - /// WAL commit latency P99 in microseconds (updated by WAL group-commit path). + /// WAL commit latency P99 in microseconds. Node-wide metric sampled from the node's + /// single WalManager; every database reports the same node-wide value. pub wal_commit_latency_p99_us: AtomicU64, - /// Cumulative maintenance CPU-seconds consumed by this database. + /// Cumulative maintenance CPU-seconds consumed by this database (in microseconds). pub maintenance_cpu_seconds_total: AtomicU64, /// Mirror replication lag in milliseconds (`now_ms - last_apply_ms`). /// Zero for non-mirror databases or a promoted mirror. @@ -46,11 +47,10 @@ pub struct DatabaseCounters { /// control-plane-side collection point for Prometheus scrapes. #[derive(Debug, Default)] pub struct DatabaseMetricsRegistry { - /// Map from database display name to its live counter handle. + /// Map from `DatabaseId` to its display name and live counter handle. /// - /// Display names are resolved from `DatabaseId` via the catalog name cache. - /// The registry is write-locked only on first registration of a new database. - counters: RwLock>>, + /// The registry is write-locked only on first registration or removal of a database. + counters: RwLock)>>, } impl DatabaseMetricsRegistry { @@ -59,94 +59,101 @@ impl DatabaseMetricsRegistry { Self::default() } - /// Get or create the counter handle for `db_name`. + /// Get or create the counter handle for `(db_id, db_name)`. /// /// The returned `Arc` is stable for the process lifetime. - pub fn get_or_create(&self, db_name: &str) -> Arc { + pub fn get_or_create(&self, db_id: DatabaseId, db_name: &str) -> Arc { // Fast path: read lock. { let r = self.counters.read().unwrap_or_else(|p| p.into_inner()); - if let Some(c) = r.get(db_name) { + if let Some((_, c)) = r.get(&db_id) { return Arc::clone(c); } } // Slow path: write lock. let mut w = self.counters.write().unwrap_or_else(|p| p.into_inner()); - w.entry(db_name.to_string()) - .or_insert_with(|| Arc::new(DatabaseCounters::default())) + w.entry(db_id) + .or_insert_with(|| (db_name.to_string(), Arc::new(DatabaseCounters::default()))) + .1 .clone() } - /// Increment the QPS counter for `db_name` by 1. - pub fn record_qps(&self, db_name: &str) { - self.get_or_create(db_name) + /// Remove all metrics for `db_id` from the registry (e.g. on DROP DATABASE). + pub fn remove(&self, db_id: DatabaseId) -> Option<(String, Arc)> { + let mut w = self.counters.write().unwrap_or_else(|p| p.into_inner()); + w.remove(&db_id) + } + + /// Increment the QPS counter for `db_id` by 1. + pub fn record_qps(&self, db_id: DatabaseId, db_name: &str) { + self.get_or_create(db_id, db_name) .qps_total .fetch_add(1, Ordering::Relaxed); } - /// Set the memory-bytes gauge for `db_name`. - pub fn set_memory_bytes(&self, db_name: &str, bytes: u64) { - self.get_or_create(db_name) + /// Set the memory-bytes gauge for `db_id`. + pub fn set_memory_bytes(&self, db_id: DatabaseId, db_name: &str, bytes: u64) { + self.get_or_create(db_id, db_name) .memory_bytes .store(bytes, Ordering::Relaxed); } - /// Set the storage-bytes gauge for `db_name`. - pub fn set_storage_bytes(&self, db_name: &str, bytes: u64) { - self.get_or_create(db_name) + /// Set the storage-bytes gauge for `db_id`. + pub fn set_storage_bytes(&self, db_id: DatabaseId, db_name: &str, bytes: u64) { + self.get_or_create(db_id, db_name) .storage_bytes .store(bytes, Ordering::Relaxed); } - /// Set the active-connections gauge for `db_name`. - pub fn set_connections(&self, db_name: &str, count: u64) { - self.get_or_create(db_name) + /// Set the active-connections gauge for `db_id`. + pub fn set_connections(&self, db_id: DatabaseId, db_name: &str, count: u64) { + self.get_or_create(db_id, db_name) .connections .store(count, Ordering::Relaxed); } - /// Set the bridge queue-depth snapshot for `db_name`. - pub fn set_bridge_queue_depth(&self, db_name: &str, depth: u64) { - self.get_or_create(db_name) + /// Set the bridge queue-depth snapshot for `db_id`. + pub fn set_bridge_queue_depth(&self, db_id: DatabaseId, db_name: &str, depth: u64) { + self.get_or_create(db_id, db_name) .bridge_queue_depth .store(depth, Ordering::Relaxed); } - /// Set the WAL commit latency P99 (microseconds) for `db_name`. - pub fn set_wal_latency_p99(&self, db_name: &str, us: u64) { - self.get_or_create(db_name) + /// Set the WAL commit latency P99 (microseconds) for `db_id`. + /// + /// Note: WAL latency is a node-wide metric (one WalManager per node); all databases + /// report the same node-wide latency. + pub fn set_wal_latency_p99(&self, db_id: DatabaseId, db_name: &str, us: u64) { + self.get_or_create(db_id, db_name) .wal_commit_latency_p99_us .store(us, Ordering::Relaxed); } - /// Set the mirror replication lag in milliseconds for `db_name`. + /// Set the mirror replication lag in milliseconds for `db_id`. /// /// Computed as `now_ms - last_apply_ms` using wall-clock time. The /// bounded-staleness read rejection path uses the same wall-clock basis, /// so the metric and the rejection gate are always consistent. - pub fn set_mirror_lag_ms(&self, db_name: &str, lag_ms: u64) { - self.get_or_create(db_name) + pub fn set_mirror_lag_ms(&self, db_id: DatabaseId, db_name: &str, lag_ms: u64) { + self.get_or_create(db_id, db_name) .mirror_lag_ms .store(lag_ms, Ordering::Relaxed); } - /// Add `secs` fractional CPU-seconds to the maintenance counter for `db_name`. + /// Set cumulative maintenance CPU-seconds for `db_id`. /// /// Converts to integer microseconds (rounded) to avoid `AtomicF64` complexity. /// Negative or non-finite inputs are clamped to zero so accidental underflow /// in upstream timing arithmetic cannot subtract from the cumulative counter. - pub fn add_maintenance_cpu_secs(&self, db_name: &str, secs: f64) { + pub fn set_maintenance_cpu_secs(&self, db_id: DatabaseId, db_name: &str, secs: f64) { let us = if secs.is_finite() && secs > 0.0 { (secs * 1_000_000.0).round() as u64 } else { 0 }; - if us == 0 { - return; - } - self.get_or_create(db_name) + self.get_or_create(db_id, db_name) .maintenance_cpu_seconds_total - .fetch_add(us, Ordering::Relaxed); + .store(us, Ordering::Relaxed); } /// Render all registered databases as Prometheus text format 0.0.4. @@ -160,7 +167,7 @@ impl DatabaseMetricsRegistry { return; } - let mut pairs: Vec<_> = r.iter().collect(); + let mut pairs: Vec<_> = r.values().map(|(name, c)| (name.as_str(), c)).collect(); pairs.sort_by(|a, b| a.0.cmp(b.0)); macro_rules! emit_counter { @@ -226,7 +233,7 @@ impl DatabaseMetricsRegistry { ); emit_gauge!( "nodedb_database_wal_commit_latency_p99_us", - "WAL commit latency P99 in microseconds per database", + "WAL commit latency P99 in microseconds (node-wide; shared across all databases)", wal_commit_latency_p99_us ); emit_counter!( @@ -280,8 +287,9 @@ mod tests { #[test] fn registry_get_or_create_idempotent() { let reg = DatabaseMetricsRegistry::new(); - let a = reg.get_or_create("mydb"); - let b = reg.get_or_create("mydb"); + let db = DatabaseId::new(1); + let a = reg.get_or_create(db, "mydb"); + let b = reg.get_or_create(db, "mydb"); // Both arcs point to the same allocation. assert!(Arc::ptr_eq(&a, &b)); } @@ -289,17 +297,18 @@ mod tests { #[test] fn record_qps_increments() { let reg = DatabaseMetricsRegistry::new(); - reg.record_qps("db1"); - reg.record_qps("db1"); - let c = reg.get_or_create("db1"); + let db = DatabaseId::new(1); + reg.record_qps(db, "db1"); + reg.record_qps(db, "db1"); + let c = reg.get_or_create(db, "db1"); assert_eq!(c.qps_total.load(Ordering::Relaxed), 2); } #[test] fn render_prometheus_contains_labels() { let reg = DatabaseMetricsRegistry::new(); - reg.record_qps("alpha"); - reg.record_qps("beta"); + reg.record_qps(DatabaseId::new(1), "alpha"); + reg.record_qps(DatabaseId::new(2), "beta"); let mut out = String::new(); reg.render_prometheus(&mut out); assert!(out.contains(r#"database="alpha""#)); @@ -326,4 +335,39 @@ mod tests { }; assert!(m.is_over_quota()); } + + #[test] + fn gauge_setters_update_counters() { + let reg = DatabaseMetricsRegistry::new(); + let db = DatabaseId::new(1); + reg.set_connections(db, "db1", 42); + reg.set_memory_bytes(db, "db1", 1024 * 1024); + reg.set_storage_bytes(db, "db1", 10 * 1024 * 1024); + reg.set_bridge_queue_depth(db, "db1", 7); + reg.set_wal_latency_p99(db, "db1", 1500); + + let c = reg.get_or_create(db, "db1"); + assert_eq!(c.connections.load(Ordering::Relaxed), 42); + assert_eq!(c.memory_bytes.load(Ordering::Relaxed), 1024 * 1024); + assert_eq!(c.storage_bytes.load(Ordering::Relaxed), 10 * 1024 * 1024); + assert_eq!(c.bridge_queue_depth.load(Ordering::Relaxed), 7); + assert_eq!(c.wal_commit_latency_p99_us.load(Ordering::Relaxed), 1500); + } + + #[test] + fn remove_clears_database_metrics() { + let reg = DatabaseMetricsRegistry::new(); + let db = DatabaseId::new(99); + reg.set_connections(db, "db_to_drop", 10); + assert_eq!( + reg.get_or_create(db, "db_to_drop") + .connections + .load(Ordering::Relaxed), + 10 + ); + reg.remove(db); + let mut out = String::new(); + reg.render_prometheus(&mut out); + assert!(!out.contains(r#"database="db_to_drop""#)); + } } diff --git a/nodedb/src/control/metrics/histogram.rs b/nodedb/src/control/metrics/histogram.rs index 48a228efaf..66ebb45e7f 100644 --- a/nodedb/src/control/metrics/histogram.rs +++ b/nodedb/src/control/metrics/histogram.rs @@ -168,6 +168,19 @@ impl AtomicHistogram { snap } + /// Create a point-in-time snapshot of bucket counts for rolling delta calculations. + pub fn snapshot_counts(&self) -> HistogramSnapshot { + HistogramSnapshot { + boundaries: self.boundaries, + buckets: self + .buckets + .iter() + .map(|b| b.load(Ordering::Relaxed)) + .collect(), + count: self.count.load(Ordering::Relaxed), + } + } + /// Merge another histogram's counts into this one. /// /// Both histograms must share the same bucket boundaries — if they do @@ -186,6 +199,71 @@ impl AtomicHistogram { } } +/// Point-in-time snapshot of bucket counts for rolling-window percentile calculations. +#[derive(Debug, Clone, Default)] +pub struct HistogramSnapshot { + /// Upper bounds in microseconds. + pub boundaries: &'static [u64], + /// Bucket counters at snapshot time. + pub buckets: Vec, + /// Total observations at snapshot time. + pub count: u64, +} + +impl HistogramSnapshot { + /// Compute percentile from the bucket count delta between `self` (earlier) and `newer`. + /// + /// `p` is a fraction between `0.0` and `1.0` (e.g. `0.99` for P99). + /// Returns `0` if no new observations occurred in the interval. + pub fn delta_percentile(&self, newer: &HistogramSnapshot, p: f64) -> u64 { + let is_empty_baseline = self.boundaries.is_empty() && self.count == 0; + if !is_empty_baseline && (self.boundaries != newer.boundaries || newer.count <= self.count) + { + return 0; + } + if newer.count == 0 { + return 0; + } + let delta_count = if is_empty_baseline { + newer.count + } else { + newer.count - self.count + }; + if delta_count == 0 { + return 0; + } + let target = (p * delta_count as f64) as u64; + let mut cumulative = 0u64; + let mut prev_boundary = 0u64; + + for (i, &boundary) in newer.boundaries.iter().enumerate() { + let older_bucket = if is_empty_baseline { + 0 + } else { + self.buckets.get(i).copied().unwrap_or(0) + }; + let newer_bucket = newer.buckets.get(i).copied().unwrap_or(0); + let bucket_delta = newer_bucket.saturating_sub(older_bucket); + cumulative += bucket_delta; + if cumulative >= target { + let bucket_start = prev_boundary; + let bucket_width = boundary - bucket_start; + if bucket_delta == 0 { + return boundary; + } + let fraction = if cumulative > target { + (bucket_delta - (cumulative - target)) as f64 / bucket_delta as f64 + } else { + 1.0 + }; + return bucket_start + (fraction * bucket_width as f64) as u64; + } + prev_boundary = boundary; + } + newer.boundaries.last().copied().unwrap_or(0) + } +} + impl Default for AtomicHistogram { fn default() -> Self { Self::new() @@ -229,6 +307,67 @@ mod tests { assert!((500..=1000).contains(&p50), "p50={p50}"); } + #[test] + fn percentile_p99_numerically_correct() { + // WAL buckets: [100, 500, 1000, 5000, 10000, 50000, 100000, 500000, 1000000] + let h = AtomicHistogram::with_buckets(WAL_FSYNC_BUCKETS_US); + // Observe 90 items at 80us (falls in <=100us bucket) + for _ in 0..90 { + h.observe(80); + } + // Observe 9 items at 400us (falls in <=500us bucket) + for _ in 0..9 { + h.observe(400); + } + // Observe 1 item at 800us (falls in <=1000us bucket) + h.observe(800); + + assert_eq!(h.count(), 100); + + // p50 is rank 50 (within first bucket 0..100us) + let p50 = h.percentile(0.50); + assert!(p50 <= 100, "expected p50 <= 100, got {p50}"); + + // p99 is rank 99 (90 in bucket0 + 9 in bucket1 = 99 -> top of <=500us bucket) + let p99 = h.percentile(0.99); + assert!( + (100..=500).contains(&p99), + "expected p99 in [100, 500], got {p99}" + ); + } + + #[test] + fn rolling_window_delta_percentile() { + let h = AtomicHistogram::with_buckets(WAL_FSYNC_BUCKETS_US); + + // Window 1: 100 observations at 80us (<=100us) + for _ in 0..100 { + h.observe(80); + } + let snap1 = h.snapshot_counts(); + + // Window 2: 90 observations at 80us, 10 observations at 4000us (<=5000us) + for _ in 0..90 { + h.observe(80); + } + for _ in 0..10 { + h.observe(4000); + } + let snap2 = h.snapshot_counts(); + + // Delta p99 in Window 2 alone (100 new observations: 90 at 80us, 10 at 4000us) + let delta_p99 = snap1.delta_percentile(&snap2, 0.99); + assert!( + (1000..=5000).contains(&delta_p99), + "expected delta p99 in [1000, 5000], got {delta_p99}" + ); + + // Window 3: No new observations + let snap3 = h.snapshot_counts(); + let delta_p99_empty = snap2.delta_percentile(&snap3, 0.99); + assert_eq!(delta_p99_empty, 0); + } + #[test] fn prometheus_output() { let h = AtomicHistogram::new(); diff --git a/nodedb/src/control/metrics/system/render.rs b/nodedb/src/control/metrics/system/render.rs index fa79406ff1..de9edafd7c 100644 --- a/nodedb/src/control/metrics/system/render.rs +++ b/nodedb/src/control/metrics/system/render.rs @@ -240,6 +240,28 @@ impl SystemMetrics { } } + /// Remove all metric entries for a dropped database. + pub fn remove_database(&self, db_name: &str) { + if let Ok(mut m) = self.database_collections_by_name.write() { + m.remove(db_name); + } + if let Ok(mut m) = self.database_tenants_by_name.write() { + m.remove(db_name); + } + if let Ok(mut m) = self.database_memory_bytes_by_name.write() { + m.remove(db_name); + } + if let Ok(mut m) = self.database_storage_bytes_by_name.write() { + m.remove(db_name); + } + if let Ok(mut m) = self.database_queries_by_name.write() { + m.remove(db_name); + } + if let Ok(mut m) = self.database_errors_by_name.write() { + m.remove(db_name); + } + } + pub(super) fn prometheus_backpressure(&self, out: &mut String) { use std::fmt::Write as _; let critical = self diff --git a/nodedb/src/control/mirror/observer.rs b/nodedb/src/control/mirror/observer.rs index 86699e4445..cf9f11d2a2 100644 --- a/nodedb/src/control/mirror/observer.rs +++ b/nodedb/src/control/mirror/observer.rs @@ -183,7 +183,7 @@ pub fn update_lag_status( let lag_ms = now.saturating_sub(lag_record.last_apply_ms); // Publish the metric: now_ms - last_apply_ms. - db_metrics.set_mirror_lag_ms(db_name, lag_ms); + db_metrics.set_mirror_lag_ms(db_id, db_name, lag_ms); // Fall back to apply-time when the registry has no entry. let last_received = last_entry_received_ms.unwrap_or(lag_record.last_apply_ms); diff --git a/nodedb/src/control/server/admission/registry.rs b/nodedb/src/control/server/admission/registry.rs index 3e62578884..5a8f066582 100644 --- a/nodedb/src/control/server/admission/registry.rs +++ b/nodedb/src/control/server/admission/registry.rs @@ -139,7 +139,7 @@ impl LimitEntry { let owed = u32::try_from(owed).unwrap_or(u32::MAX); let _ = self .pending_shrink - .try_update(Ordering::AcqRel, Ordering::Acquire, |current| { + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| { Some(current.saturating_add(owed)) }); } @@ -273,13 +273,23 @@ impl AdmissionRegistry { /// Attempt to acquire a database-level permit for a new connection. /// - /// Returns `Ok(Some(permit))` if an entry exists and a slot was acquired, - /// `Ok(None)` if the database has no entry, or `Err(AdmissionError)` if the - /// database is at capacity. + /// Returns `Ok(Some(permit))` when a slot was acquired, or `Err(AdmissionError)` + /// if the database is at capacity. pub fn try_acquire_database( &self, db: DatabaseId, ) -> Result, AdmissionError> { + let has_entry = { + let map = self.db_semaphores.read().unwrap_or_else(|p| p.into_inner()); + map.contains_key(&db) + }; + if !has_entry { + let mut map = self + .db_semaphores + .write() + .unwrap_or_else(|p| p.into_inner()); + map.entry(db).or_insert_with(|| LimitEntry::new(None)); + } let map = self.db_semaphores.read().unwrap_or_else(|p| p.into_inner()); let permit = try_acquire_entry(map.get(&db), |limit| { AdmissionError::DatabaseCapExhausted { db, limit } @@ -372,10 +382,12 @@ mod tests { #[test] fn no_database_cap_allows_unlimited() { let reg = AdmissionRegistry::new(); - // No entry configured → Ok(None). - let r = reg.try_acquire_database(db(0)); - assert!(r.unwrap().is_none()); - assert_eq!(reg.database_live_connections(db(0)), None); + // An uncapped database admits and tracks live connections. + let p = reg.try_acquire_database(db(0)).unwrap(); + assert!(p.is_some()); + assert_eq!(reg.database_live_connections(db(0)), Some(1)); + drop(p); + assert_eq!(reg.database_live_connections(db(0)), Some(0)); } #[test] @@ -563,7 +575,7 @@ mod tests { reg.set_database_limit(db(5), 0); assert_eq!(reg.database_live_connections(db(5)), None); - assert!(reg.try_acquire_database(db(5)).unwrap().is_none()); + assert!(reg.try_acquire_database(db(5)).unwrap().is_some()); } #[test] diff --git a/nodedb/src/control/server/pgwire/handler/sql_exec.rs b/nodedb/src/control/server/pgwire/handler/sql_exec.rs index 1d83cfc581..1e6372e571 100644 --- a/nodedb/src/control/server/pgwire/handler/sql_exec.rs +++ b/nodedb/src/control/server/pgwire/handler/sql_exec.rs @@ -413,7 +413,7 @@ impl NodeDbPgHandler { if let Some(ref m) = self.state.system_metrics { m.record_database_query(&desc.name); } - self.state.database_metrics.record_qps(&desc.name); + self.state.database_metrics.record_qps(desc.id, &desc.name); } // Request-admission gate: internal-service exemption, blacklist, diff --git a/nodedb/src/control/server/shared/ddl/neutral/database/drop.rs b/nodedb/src/control/server/shared/ddl/neutral/database/drop.rs index 8d97dc8f33..871ad43824 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/database/drop.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/database/drop.rs @@ -185,15 +185,7 @@ pub async fn drop_database( // Remove per-database metrics entries on drop. if let Some(m) = &state.system_metrics { - if let Ok(mut map) = m.database_collections_by_name.write() { - map.remove(name); - } - if let Ok(mut map) = m.database_queries_by_name.write() { - map.remove(name); - } - if let Ok(mut map) = m.database_errors_by_name.write() { - map.remove(name); - } + m.remove_database(name); } Ok(status("DROP DATABASE")) diff --git a/nodedb/src/control/server/shared/ddl/neutral/observability.rs b/nodedb/src/control/server/shared/ddl/neutral/observability.rs index e5e9e47f23..c7fa02d3dc 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/observability.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/observability.rs @@ -14,8 +14,10 @@ //! `SHOW MEMORY` reports per-engine memory budgets and current //! utilisation from `nodedb_mem::MemoryGovernor`. //! -//! The metric source reads, ordering, and the tenant-admin gate run here. The -//! result is the protocol-neutral `DdlResult` over `ShapedRows`. +//! Ported from the pgwire `ddl::observability` handlers. The metric source +//! reads, ordering, and the tenant-admin gate are preserved verbatim; only the +//! result construction changed from pgwire `Response` / `QueryResponse` to the +//! protocol-neutral `DdlResult` over `ShapedRows`. use serde_json::{Map, Value as JsonValue}; @@ -169,19 +171,19 @@ pub fn show_metrics( if let Some(sys) = state.system_metrics.as_ref() { rows.push(( "wal_fsync_p50_us".into(), - sys.wal_fsync_seconds.percentile(50.0).to_string(), + sys.wal_fsync_seconds.percentile(0.5).to_string(), )); rows.push(( "wal_fsync_p99_us".into(), - sys.wal_fsync_seconds.percentile(99.0).to_string(), + sys.wal_fsync_seconds.percentile(0.99).to_string(), )); rows.push(( "query_latency_p50_us".into(), - sys.query_latency.percentile(50.0).to_string(), + sys.query_latency.percentile(0.5).to_string(), )); rows.push(( "query_latency_p99_us".into(), - sys.query_latency.percentile(99.0).to_string(), + sys.query_latency.percentile(0.99).to_string(), )); } diff --git a/nodedb/src/control/state/init.rs b/nodedb/src/control/state/init.rs index ac4cc2f9d8..693ef40fee 100644 --- a/nodedb/src/control/state/init.rs +++ b/nodedb/src/control/state/init.rs @@ -436,6 +436,7 @@ impl SharedState { startup: Arc::clone(&startup_gate), }); Self::wire_session_handle_audit(&state); + state.wire_wal_and_metrics(); Ok(state) } } diff --git a/nodedb/src/control/state/init_prod/post_init.rs b/nodedb/src/control/state/init_prod/post_init.rs index cbf37066b9..6ba27201d6 100644 --- a/nodedb/src/control/state/init_prod/post_init.rs +++ b/nodedb/src/control/state/init_prod/post_init.rs @@ -17,6 +17,7 @@ use super::super::SharedState; pub(super) fn hydrate_caches(state: &Arc) { SharedState::wire_session_handle_audit(state); super::catalog_registries::load_catalog_caches(state, state.credentials.catalog()); + state.wire_wal_and_metrics(); } /// Spawn the array GC background task. The handle is stored by the caller diff --git a/nodedb/src/control/state/methods.rs b/nodedb/src/control/state/methods.rs index a4b97b76da..99d5dd6c9b 100644 --- a/nodedb/src/control/state/methods.rs +++ b/nodedb/src/control/state/methods.rs @@ -13,6 +13,13 @@ use crate::types::TenantId; use super::SharedState; impl SharedState { + /// Wire system metrics to the WAL manager if system metrics are present. + pub(super) fn wire_wal_and_metrics(&self) { + if let Some(metrics) = &self.system_metrics { + self.wal.set_metrics(Arc::clone(metrics)); + } + } + /// Bundle of the auth-adjacent stores every `RequestAuthScope` /// construction needs (`scope_grants`, `quota_manager`), for callers /// that build a scope from `&SharedState` directly rather than pulling diff --git a/nodedb/src/wal/manager/core.rs b/nodedb/src/wal/manager/core.rs index 571c97f62c..31a26a0090 100644 --- a/nodedb/src/wal/manager/core.rs +++ b/nodedb/src/wal/manager/core.rs @@ -59,6 +59,8 @@ pub struct WalManager { pub(super) ordinal_clock: Arc, /// Floors truncation keeps for holders that read records back at boot. pub(super) floor_holds: super::floor_holds::WalFloorHolds, + /// System metrics registry handle for recording group-commit fsync latency. + pub(super) metrics: std::sync::RwLock>>, } impl WalManager { @@ -91,6 +93,20 @@ impl WalManager { /// rewrapped on rotation, so offline signatures survive chained key /// rotations and restarts. No root exists when WAL encryption is disabled, /// which prevents `SIGNED_DELTAS` from being enabled. + /// Wire or replace the system metrics handle used to record group-commit fsync latency. + pub fn set_metrics(&self, metrics: Arc) { + let mut w = self.metrics.write().unwrap_or_else(|p| p.into_inner()); + *w = Some(metrics); + } + + /// Read the optional system metrics handle. + pub fn metrics(&self) -> Option> { + self.metrics + .read() + .unwrap_or_else(|p| p.into_inner()) + .clone() + } + pub fn crdt_signing_root(&self) -> crate::Result> { Ok(self.crdt_signing_root) } @@ -231,6 +247,7 @@ impl WalManager { time_anchors, ordinal_clock: Arc::new(nodedb_types::OrdinalClock::new()), floor_holds: super::floor_holds::WalFloorHolds::default(), + metrics: std::sync::RwLock::new(None), }) } diff --git a/nodedb/src/wal/manager/durable_commit.rs b/nodedb/src/wal/manager/durable_commit.rs index feb92eb382..f563e00d0b 100644 --- a/nodedb/src/wal/manager/durable_commit.rs +++ b/nodedb/src/wal/manager/durable_commit.rs @@ -63,9 +63,15 @@ impl WalManager { // advances to exactly what the fsync made durable, never past // it. let wal = std::sync::Arc::clone(&self.wal); + let metrics = self.metrics(); let join = tokio::task::spawn_blocking(move || -> crate::Result { let mut guard = wal.lock().unwrap_or_else(|p| p.into_inner()); + let start = std::time::Instant::now(); guard.sync().map_err(crate::Error::Wal)?; + let elapsed_us = start.elapsed().as_micros() as u64; + if let Some(m) = &metrics { + m.record_wal_fsync(elapsed_us); + } // `next_lsn()` is the next LSN to assign; the highest LSN // this sync made durable is one below it. Ok(guard.next_lsn().saturating_sub(1)) @@ -113,6 +119,8 @@ mod tests { use super::*; use crate::types::{DatabaseId, TenantId, VShardId}; use crate::wal::manager::NO_APPLY_KEY; + use std::sync::Arc; + use std::sync::atomic::Ordering; fn open_wal(dir: &std::path::Path) -> WalManager { WalManager::open_for_testing(&dir.join("test.wal")).expect("open wal") @@ -140,6 +148,8 @@ mod tests { async fn wait_durable_fast_path_when_already_durable() { let dir = tempfile::tempdir().expect("tempdir"); let wal = open_wal(dir.path()); + let metrics = Arc::new(crate::control::metrics::SystemMetrics::new()); + wal.set_metrics(Arc::clone(&metrics)); let lsn = wal .appender(NO_APPLY_KEY) .with_event_source(crate::event::EventSource::User) @@ -151,8 +161,11 @@ mod tests { ) .expect("append"); wal.wait_durable(lsn).await.expect("first"); + assert_eq!(metrics.wal_fsync_count.load(Ordering::Relaxed), 1); + assert_eq!(metrics.wal_fsync_seconds.count(), 1); // Second call is the fast path — no further fsync required. wal.wait_durable(lsn).await.expect("second"); + assert_eq!(metrics.wal_fsync_count.load(Ordering::Relaxed), 1); } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] diff --git a/nodedb/tests/inproc/cases/quota_drop_cleanup.rs b/nodedb/tests/inproc/cases/quota_drop_cleanup.rs index 63a7b7d57d..f834b3e618 100644 --- a/nodedb/tests/inproc/cases/quota_drop_cleanup.rs +++ b/nodedb/tests/inproc/cases/quota_drop_cleanup.rs @@ -245,11 +245,11 @@ async fn dropped_database_id_does_not_inherit_cap() { None, "the dead cap must not survive for a database reusing this id" ); + let permit = registry + .try_acquire_database(db) + .expect("an uncapped database admits"); assert!( - registry - .try_acquire_database(db) - .expect("an uncapped database admits") - .is_none(), - "an uncapped database hands out no permit" + permit.is_some(), + "an uncapped database hands out an unconstrained permit" ); } diff --git a/nodedb/tests/inproc/cases/quota_live_enforcement_apply.rs b/nodedb/tests/inproc/cases/quota_live_enforcement_apply.rs index 49bcfff32f..29f3a3eb21 100644 --- a/nodedb/tests/inproc/cases/quota_live_enforcement_apply.rs +++ b/nodedb/tests/inproc/cases/quota_live_enforcement_apply.rs @@ -90,12 +90,12 @@ async fn alter_database_quota_zero_clears_connection_cap() { None, "zero drops the entry, so the database is uncapped again" ); + let permit = registry + .try_acquire_database(db) + .expect("an uncapped database admits"); assert!( - registry - .try_acquire_database(db) - .expect("an uncapped database admits") - .is_none(), - "an uncapped database hands out no permit" + permit.is_some(), + "an uncapped database hands out an unconstrained permit" ); } diff --git a/nodedb/tests/inproc/cases/quota_three_level_denial.rs b/nodedb/tests/inproc/cases/quota_three_level_denial.rs index dc927dd0ab..967379d09a 100644 --- a/nodedb/tests/inproc/cases/quota_three_level_denial.rs +++ b/nodedb/tests/inproc/cases/quota_three_level_denial.rs @@ -216,15 +216,23 @@ fn maintenance_budget_tracker_three_level() { #[test] fn database_metrics_registry_independent_counters() { let reg = DatabaseMetricsRegistry::new(); + let db_alpha = nodedb_types::DatabaseId::new(1); + let db_beta = nodedb_types::DatabaseId::new(2); for _ in 0..5 { - reg.record_qps("alpha"); + reg.record_qps(db_alpha, "alpha"); } for _ in 0..3 { - reg.record_qps("beta"); + reg.record_qps(db_beta, "beta"); } use std::sync::atomic::Ordering; - let alpha = reg.get_or_create("alpha").qps_total.load(Ordering::Relaxed); - let beta = reg.get_or_create("beta").qps_total.load(Ordering::Relaxed); + let alpha = reg + .get_or_create(db_alpha, "alpha") + .qps_total + .load(Ordering::Relaxed); + let beta = reg + .get_or_create(db_beta, "beta") + .qps_total + .load(Ordering::Relaxed); assert_eq!(alpha, 5); assert_eq!(beta, 3); diff --git a/nodedb/tests/wire/cases/database_metrics.rs b/nodedb/tests/wire/cases/database_metrics.rs index 3f9b0ca5b0..44d09fc8ed 100644 --- a/nodedb/tests/wire/cases/database_metrics.rs +++ b/nodedb/tests/wire/cases/database_metrics.rs @@ -168,3 +168,38 @@ async fn tenant_quota_metrics_non_zero_under_load() { "at least one tenant should have tracked requests: {body}" ); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn drop_database_removes_metrics_series() { + let (server, db_a) = TestServer::with_database("metrics_drop_test").await; + + // Issue a query so db_a creates a series in DatabaseMetricsRegistry + server + .exec("CREATE COLLECTION col_drop (id STRING PRIMARY KEY) WITH (engine='kv')") + .await + .unwrap(); + + let body = fetch_metrics_body(server.http_port).await; + assert!( + body.contains(&format!("database=\"{db_a}\"")), + "database '{db_a}' must appear in /metrics after queries" + ); + + // Drop database via SQL through real post-apply path + server + .client + .simple_query("USE DATABASE default") + .await + .unwrap(); + server + .client + .simple_query(&format!("DROP DATABASE {db_a}")) + .await + .unwrap(); + + let body_after = fetch_metrics_body(server.http_port).await; + assert!( + !body_after.contains(&format!("database=\"{db_a}\"")), + "database '{db_a}' series must be removed from /metrics after DROP DATABASE: {body_after}" + ); +}