linkrs_metrics/manager/
storage.rs

1//! Storage engine observability: I/O, bloom filters and space health.
2use serde::{Deserialize, Serialize};
3
4use super::core::StatsManager;
5use super::metric_type::MetricType;
6
7/// Storage subsystem snapshot for handlers.
8#[derive(Debug, Clone, Serialize, Deserialize)]
9pub struct StorageSnapshot {
10    pub read_ops: u64,
11    pub write_ops: u64,
12    pub avg_read_latency_us: f64,
13    pub avg_write_latency_us: f64,
14    pub storage_errors: u64,
15    pub index_memory_bytes: u64,
16    pub tombstone_count: u64,
17    pub tombstone_memory_bytes: u64,
18    pub fragmentation_permille: u64,
19    pub wasted_bytes: u64,
20    pub bloom_queries: u64,
21    pub bloom_hits: u64,
22    pub bloom_hit_rate: f64,
23    pub dirty_pages: u64,
24    pub dirty_pages_total: u64,
25    pub dirty_page_ratio_permille: u64,
26    pub checkpoint_strategy_incremental: u64,
27    pub checkpoint_strategy_hybrid: u64,
28    pub checkpoint_strategy_full: u64,
29    pub checkpoint_incremental_duration_us: u64,
30    pub checkpoint_incremental_bytes_flushed: u64,
31}
32
33fn avg(total: u64, count: u64) -> f64 {
34    if count == 0 {
35        0.0
36    } else {
37        total as f64 / count as f64
38    }
39}
40
41impl StatsManager {
42    pub fn record_storage_read(&self, latency_us: u64) {
43        self.add_value(MetricType::StorageReadOps);
44        self.add_value_with_amount(MetricType::StorageReadLatencyUs, latency_us);
45    }
46
47    /// Record a storage write with latency so means are computable.
48    pub fn record_storage_write(&self, latency_us: u64) {
49        self.add_value(MetricType::StorageWriteOps);
50        self.add_value_with_amount(MetricType::StorageWriteLatencyUs, latency_us);
51    }
52
53    /// Record a storage error.
54    pub fn record_storage_error(&self) {
55        self.add_value(MetricType::StorageErrors);
56    }
57
58    /// Record a bloom pre-check query and whether it was positive.
59    /// Hit rate derives from `BloomHits / BloomQueries`; permille is exported
60    /// for dashboards via `BloomHitRatePermille`. The triple update holds
61    /// `bloom_lock` so queries/hits/permille stay atomic under concurrency.
62    pub fn record_bloom_query(&self, hit: bool) {
63        let _guard = self.bloom_lock.lock();
64        self.add_value(MetricType::BloomQueries);
65        if hit {
66            self.add_value(MetricType::BloomHits);
67        }
68        if let (Some(queries), Some(hits)) = (
69            self.get_value(MetricType::BloomQueries),
70            self.get_value(MetricType::BloomHits),
71        ) {
72            if queries > 0 {
73                let permille = hits.saturating_mul(1000) / queries.max(1);
74                self.set_value(MetricType::BloomHitRatePermille, permille);
75            }
76        }
77    }
78
79    /// Snapshot-export bloom counters from a shard runtime.
80    /// Holds `bloom_lock` for the same atomicity as `record_bloom_query`.
81    /// Callers must pass deltas since the last export, not cumulative
82    /// totals, to avoid double counting.
83    pub fn record_bloom_snapshot(&self, queries: u64, hits: u64) {
84        let _guard = self.bloom_lock.lock();
85        self.add_value_with_amount(MetricType::BloomQueries, queries);
86        self.add_value_with_amount(MetricType::BloomHits, hits);
87        let total_queries = self.get_value(MetricType::BloomQueries).unwrap_or(0);
88        let total_hits = self.get_value(MetricType::BloomHits).unwrap_or(0);
89        if total_queries > 0 {
90            self.set_value(
91                MetricType::BloomHitRatePermille,
92                total_hits.saturating_mul(1000) / total_queries.max(1),
93            );
94        }
95    }
96
97    pub fn set_index_memory_usage(&self, bytes: u64) {
98        self.set_value(MetricType::IndexMemoryUsage, bytes);
99    }
100
101    pub fn storage_snapshot(&self) -> StorageSnapshot {
102        let v = |m: MetricType| self.get_value(m).unwrap_or(0);
103        let read_ops = v(MetricType::StorageReadOps);
104        let write_ops = v(MetricType::StorageWriteOps);
105        let bloom_queries = v(MetricType::BloomQueries);
106        let bloom_hits = v(MetricType::BloomHits);
107        StorageSnapshot {
108            read_ops,
109            write_ops,
110            avg_read_latency_us: avg(v(MetricType::StorageReadLatencyUs), read_ops),
111            avg_write_latency_us: avg(v(MetricType::StorageWriteLatencyUs), write_ops),
112            storage_errors: v(MetricType::StorageErrors),
113            index_memory_bytes: v(MetricType::IndexMemoryUsage),
114            tombstone_count: v(MetricType::TombstoneCount),
115            tombstone_memory_bytes: v(MetricType::TombstoneMemoryBytes),
116            fragmentation_permille: v(MetricType::FragmentationRatioPermille),
117            wasted_bytes: v(MetricType::TopologyWastedBytes),
118            bloom_queries,
119            bloom_hits,
120            bloom_hit_rate: if bloom_queries == 0 {
121                0.0
122            } else {
123                bloom_hits as f64 / bloom_queries as f64
124            },
125            dirty_pages: v(MetricType::DirtyPagesCount),
126            dirty_pages_total: v(MetricType::DirtyPagesTotal),
127            dirty_page_ratio_permille: v(MetricType::DirtyPageRatioPermille),
128            checkpoint_strategy_incremental: v(MetricType::CheckpointStrategyIncremental),
129            checkpoint_strategy_hybrid: v(MetricType::CheckpointStrategyHybrid),
130            checkpoint_strategy_full: v(MetricType::CheckpointStrategyFull),
131            checkpoint_incremental_duration_us: v(MetricType::CheckpointIncrementalDurationUs),
132            checkpoint_incremental_bytes_flushed: v(MetricType::CheckpointIncrementalBytesFlushed),
133        }
134    }
135
136    /// Record tombstone statistics for MVCC observability
137    pub fn record_tombstone_stats(
138        &self,
139        count: u64,
140        memory_bytes: u64,
141        oldest_ts_min: Option<u32>,
142        newest_ts_max: Option<u32>,
143        active_snapshots: u64,
144    ) {
145        self.set_value(MetricType::TombstoneCount, count);
146        self.set_value(MetricType::TombstoneMemoryBytes, memory_bytes);
147        self.set_value(MetricType::TombstoneActiveSnapshots, active_snapshots);
148
149        if let Some(ts) = oldest_ts_min {
150            self.set_value(MetricType::TombstoneOldestTsMin, ts as u64);
151        }
152        if let Some(ts) = newest_ts_max {
153            self.set_value(MetricType::TombstoneNewestTsMax, ts as u64);
154        }
155    }
156
157    /// Record storage fragmentation on the single wasted-share caliber.
158    pub fn record_fragmentation_stats(&self, ratio: f32, wasted_bytes: u64) {
159        let clamped = ratio.clamp(0.0, 1.0);
160        self.set_value(
161            MetricType::FragmentationRatioPermille,
162            (clamped * 1000.0) as u64,
163        );
164        self.set_value(MetricType::TopologyWastedBytes, wasted_bytes);
165    }
166
167    /// Record one background TTL-purge sweep: how many sweep cycles ran and
168    /// how many expired rows each cycle physically deleted.
169    pub fn record_ttl_purge(&self, cycles: u64, removed_rows: u64) {
170        if cycles > 0 {
171            self.add_value_with_amount(MetricType::TtlPurgeCycles, cycles);
172        }
173        if removed_rows > 0 {
174            self.add_value_with_amount(MetricType::TtlPurgeRows, removed_rows);
175        }
176    }
177}
linkrs_metrics/manager/storage.rs

linkrs_metrics/manager/
storage.rs
