[Storage] Add contiguous journal and QMDB metrics (#3721)

Author: danlaine

runtime/src/telemetry/metrics/histogram.rs

@@ -1,8 +1,8 @@
 //! Utilities for working with histograms.
 
-use super::{raw, Histogram};
-use crate::Clock;
-use std::time::SystemTime;
+use super::{raw, Histogram, MetricsExt as _};
+use crate::{Clock, Metrics};
+use std::{sync::Arc, time::SystemTime};
 
 /// Convenience methods for Prometheus histograms.
 pub trait HistogramExt {
@@ -101,3 +101,89 @@ impl Timer {
         self.histogram.observe_between(self.start, clock.current());
     }
 }
+
+/// A timer guard that observes its duration when dropped.
+///
+/// Built on top of [`Timer`]. Useful for `?`-heavy async code where every early-return path
+/// would otherwise need to remember to call [`Timer::observe`]. Validation failures after the
+/// guard is created are still part of the recorded duration; if a code path should not record
+/// a sample, call [`ScopedTimer::cancel`] before the guard is dropped.
+pub struct ScopedTimer<C: Clock> {
+    timer: Option<Timer>,
+    clock: Arc<C>,
+}
+
+impl<C: Clock> ScopedTimer<C> {
+    /// Cancel the guard so it does not observe a sample on drop.
+    pub fn cancel(mut self) {
+        self.timer = None;
+    }
+}
+
+impl<C: Clock> Drop for ScopedTimer<C> {
+    fn drop(&mut self) {
+        if let Some(timer) = self.timer.take() {
+            timer.observe(self.clock.as_ref());
+        }
+    }
+}
+
+impl Timed {
+    /// Start a timer guard that observes the elapsed duration when dropped.
+    pub fn scoped<C: Clock>(&self, clock: &Arc<C>) -> ScopedTimer<C> {
+        ScopedTimer {
+            timer: Some(self.timer(clock.as_ref())),
+            clock: clock.clone(),
+        }
+    }
+}
+
+/// Register a duration histogram using [`Buckets::LOCAL`] (storage-style work).
+pub fn duration_histogram<M: Metrics>(
+    context: &M,
+    name: &'static str,
+    help: &'static str,
+) -> Histogram {
+    context.histogram(name, help, Buckets::LOCAL)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{deterministic, Runner as _, Supervisor as _};
+    use std::time::Duration;
+
+    #[test]
+    fn duration_records_all_calls() {
+        deterministic::Runner::default().start(|context| async move {
+            let histogram = duration_histogram(&context, "test_duration", "test duration");
+            let timed = Timed::new(histogram);
+            let clock = Arc::new(context.child("timer"));
+
+            {
+                let _timer = timed.scoped(&clock);
+                context.sleep(Duration::from_millis(1)).await;
+                let result: Result<(), ()> = Ok(());
+                assert!(result.is_ok());
+            }
+
+            {
+                let _timer = timed.scoped(&clock);
+                context.sleep(Duration::from_millis(1)).await;
+                let result: Result<(), ()> = Err(());
+                assert!(result.is_err());
+            }
+
+            {
+                let _timer = timed.scoped(&clock);
+                context.sleep(Duration::from_millis(1)).await;
+            }
+
+            let metrics = context.encode();
+            assert!(
+                metrics.contains("test_duration_count 3"),
+                "unexpected metrics: {metrics}"
+            );
+        });
+    }
+}

storage/src/journal/contiguous/fixed.rs

@@ -57,7 +57,7 @@
 use super::Reader as _;
 use crate::{
     journal::{
-        contiguous::{Many, Mutable},
+        contiguous::{metrics::FixedMetrics as Metrics, Many, Mutable},
         segmented::fixed::{Config as SegmentedConfig, Journal as SegmentedJournal},
         Error,
     },
@@ -198,12 +198,16 @@ pub struct Journal<E: Context, A: CodecFixedShared> {
 
     /// The maximum number of items per blob (section).
     items_per_blob: u64,
+
+    /// Metrics for monitoring journal state and activity.
+    metrics: Metrics<E>,
 }
 
 /// A reader guard that holds a consistent snapshot of the journal's bounds.
 pub struct Reader<'a, E: Context, A: CodecFixedShared> {
     guard: AsyncRwLockReadGuard<'a, Inner<E, A>>,
     items_per_blob: u64,
+    metrics: &'a Metrics<E>,
 }
 
 impl<E: Context, A: CodecFixedShared> super::Reader for Reader<'_, E, A> {
@@ -214,16 +218,27 @@ impl<E: Context, A: CodecFixedShared> super::Reader for Reader<'_, E, A> {
     }
 
     async fn read(&self, pos: u64) -> Result<A, Error> {
-        self.guard.read(pos, self.items_per_blob).await
+        let _timer = self.metrics.read_timer();
+        self.metrics.read_calls.inc();
+        let result = match self.guard.read(pos, self.items_per_blob).await {
+            Ok(item) => {
+                self.metrics.items_read.inc();
+                Ok(item)
+            }
+            Err(error) => Err(error),
+        };
+        result
     }
 
     async fn read_many(&self, positions: &[u64]) -> Result<Vec<A>, Error> {
         if positions.is_empty() {
             return Ok(Vec::new());
         }
+        let _timer = self.metrics.read_many_timer();
+        self.metrics.read_many_calls.inc();
         debug_assert!(
             positions.windows(2).all(|w| w[0] < w[1]),
-            "positions must be sorted and unique"
+            "positions must be strictly increasing"
         );
         // Validate all positions.
         for &pos in positions {
@@ -259,8 +274,14 @@ impl<E: Context, A: CodecFixedShared> super::Reader for Reader<'_, E, A> {
         }
 
         if miss_positions.is_empty() {
+            self.metrics.record_cache_hits(positions.len() as u64);
+            self.metrics.items_read.inc_by(positions.len() as u64);
             return Ok(result.into_iter().map(|r| r.unwrap()).collect());
         }
+        self.metrics
+            .record_cache_hits((positions.len() - miss_positions.len()) as u64);
+        self.metrics
+            .record_cache_misses(miss_positions.len() as u64);
 
         // Phase 2: Read cache misses grouped by section (sequential).
         let mut reusable_buf = vec![0u8; miss_positions.len() * chunk_size];
@@ -308,11 +329,25 @@ impl<E: Context, A: CodecFixedShared> super::Reader for Reader<'_, E, A> {
             group_start = group_end;
         }
 
+        self.metrics.items_read.inc_by(positions.len() as u64);
         Ok(result.into_iter().map(|r| r.unwrap()).collect())
     }
 
     fn try_read_sync(&self, pos: u64) -> Option<A> {
-        self.guard.try_read_sync(pos, self.items_per_blob)
+        self.guard
+            .try_read_sync(pos, self.items_per_blob)
+            .map_or_else(
+                || {
+                    self.metrics.record_cache_misses(1);
+                    None
+                },
+                |item| {
+                    self.metrics.record_cache_hits(1);
+                    self.metrics.try_read_sync_hits.inc();
+                    self.metrics.items_read.inc();
+                    Some(item)
+                },
+            )
     }
 
     async fn replay(
@@ -467,6 +502,9 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
         let tail_section = size / items_per_blob;
         journal.ensure_section_exists(tail_section).await?;
 
+        let metrics = Metrics::new(context);
+        metrics.update(size, pruning_boundary, items_per_blob);
+
         Ok(Self {
             inner: UpgradableAsyncRwLock::new(Inner {
                 journal,
@@ -475,6 +513,7 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
                 pruning_boundary,
             }),
             items_per_blob,
+            metrics,
         })
     }
 
@@ -676,6 +715,9 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
             metadata.sync().await?;
         }
 
+        let metrics = Metrics::new(context);
+        metrics.update(size, size, items_per_blob);
+
         Ok(Self {
             inner: UpgradableAsyncRwLock::new(Inner {
                 journal,
@@ -684,6 +726,7 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
                 pruning_boundary: size, // No data exists yet
             }),
             items_per_blob,
+            metrics,
         })
     }
 
@@ -700,6 +743,8 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
     /// Only the tail section can have pending updates since historical sections are synced
     /// when they become full.
     pub async fn sync(&self) -> Result<(), Error> {
+        let _timer = self.metrics.sync_timer();
+        self.metrics.sync_calls.inc();
         // Serialize with append/prune/rewind to ensure section selection is stable, while still allowing
         // concurrent readers.
         let inner = self.inner.upgradable_read().await;
@@ -750,6 +795,7 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
         Reader {
             guard: self.inner.read().await,
             items_per_blob: self.items_per_blob,
+            metrics: &self.metrics,
         }
     }
 
@@ -762,7 +808,9 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
     /// Append a new item to the journal. Return the item's position in the journal, or error if the
     /// operation fails.
     pub async fn append(&self, item: &A) -> Result<u64, Error> {
-        self.append_many(Many::Flat(std::slice::from_ref(item)))
+        let _timer = self.metrics.append_timer();
+        self.metrics.append_calls.inc();
+        self.append_many_inner(Many::Flat(std::slice::from_ref(item)))
             .await
     }
 
@@ -771,6 +819,13 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
     /// Acquires the write lock once for all items instead of per-item.
     /// Returns [Error::EmptyAppend] if items is empty.
     pub async fn append_many<'a>(&'a self, items: Many<'a, A>) -> Result<u64, Error> {
+        let _timer = self.metrics.append_many_timer();
+        self.metrics.append_many_calls.inc();
+        self.append_many_inner(items).await
+    }
+
+    // Shared implementation for `append` and `append_many`; public wrappers record metrics.
+    async fn append_many_inner<'a>(&'a self, items: Many<'a, A>) -> Result<u64, Error> {
         if items.is_empty() {
             return Err(Error::EmptyAppend);
         }
@@ -825,6 +880,8 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
             }
         }
 
+        self.metrics
+            .update(inner.size, inner.pruning_boundary, self.items_per_blob);
         Ok(inner.size - 1)
     }
 
@@ -859,6 +916,8 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
 
         inner.journal.rewind(section, byte_offset).await?;
         inner.size = size;
+        self.metrics
+            .update(inner.size, inner.pruning_boundary, self.items_per_blob);
 
         Ok(())
     }
@@ -898,6 +957,8 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
             // Pruning boundary only moves forward
             assert!(inner.pruning_boundary < new_oldest * self.items_per_blob);
             inner.pruning_boundary = new_oldest * self.items_per_blob;
+            self.metrics
+                .update(inner.size, inner.pruning_boundary, self.items_per_blob);
         }
 
         Ok(pruned)
@@ -947,6 +1008,8 @@ impl<E: Context, A: CodecFixedShared> Journal<E, A> {
             inner.metadata.sync().await?;
         }
 
+        self.metrics
+            .update(inner.size, inner.pruning_boundary, self.items_per_blob);
         Ok(())
     }
 
@@ -1060,7 +1123,7 @@ mod tests {
     use commonware_macros::test_traced;
     use commonware_runtime::{
         deterministic::{self, Context},
-        Blob, BufferPooler, Error as RuntimeError, Runner, Storage, Supervisor as _,
+        Blob, BufferPooler, Error as RuntimeError, Metrics as _, Runner, Storage, Supervisor as _,
     };
     use commonware_utils::{NZUsize, NZU16, NZU64};
     use futures::{pin_mut, StreamExt};
@@ -3090,4 +3153,52 @@ mod tests {
             journal.destroy().await.unwrap();
         });
     }
+
+    #[test_traced]
+    fn test_fixed_journal_metrics() {
+        let executor = deterministic::Runner::default();
+        executor.start(|context| async move {
+            let cfg = test_cfg(&context, NZU64!(2));
+            let journal = Journal::<_, Digest>::init(context.child("fixed_metrics"), cfg.clone())
+                .await
+                .unwrap();
+
+            let items: Vec<_> = (0..5).map(test_digest).collect();
+            journal.append_many(Many::Flat(&items)).await.unwrap();
+            journal.append(&test_digest(5)).await.unwrap();
+            journal.sync().await.unwrap();
+            journal.reader().await.read(0).await.unwrap();
+            journal.reader().await.try_read_sync(0).unwrap();
+            journal.reader().await.read_many(&[1, 2, 4]).await.unwrap();
+            journal.prune(2).await.unwrap();
+            journal.rewind(4).await.unwrap();
+
+            let buffer = context.encode();
+            for expected in [
+                "fixed_metrics_size 4",
+                "fixed_metrics_pruning_boundary 2",
+                "fixed_metrics_retained 2",
+                "fixed_metrics_tail_items 2",
+                "fixed_metrics_append_calls_total 1",
+                "fixed_metrics_append_many_calls_total 1",
+                "fixed_metrics_read_calls_total 1",
+                "fixed_metrics_read_many_calls_total 1",
+                "fixed_metrics_try_read_sync_hits_total 1",
+                "fixed_metrics_items_read_total 5",
+                "fixed_metrics_sync_calls_total 1",
+                "fixed_metrics_append_duration_count 1",
+                "fixed_metrics_append_many_duration_count 1",
+                "fixed_metrics_read_duration_count 1",
+                "fixed_metrics_read_many_duration_count 1",
+                "fixed_metrics_sync_duration_count 1",
+                "fixed_metrics_cache_hits_total",
+                "fixed_metrics_cache_misses_total",
+                "fixed_metrics_blobs_tracked",
+            ] {
+                assert!(buffer.contains(expected), "{expected}\n{buffer}");
+            }
+
+            journal.destroy().await.unwrap();
+        });
+    }
 }