reader-map: Report Values timings to prometheus

Record the `Values` timings data as prometheus histograms. This CL
captures the duration between updates to a given key, as well as the
lifetime duration of the key in the `reader-map` before
eviction.

Fixes: REA-4639
Change-Id: If3e322ffe98306352ac9fc941fc9cdd1ed4e2db6
Release-Note-Core: Add metrics for updates and
  lifetimes of entries in the reader cache.
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/7883
Tested-by: Buildkite CI
Reviewed-by: Sidney Cammeresi <[email protected]>

Author: jasobrown-rs

Cargo.lock

@@ -6,17 +6,18 @@ authors = ["ReadySet Technology, Inc. <[email protected]>"]
 edition = "2021"
 
 [dependencies]
-triomphe = { workspace = true }
-smallvec = { workspace = true, features = ["union"] }
-rand = { workspace = true }
-left-right = { workspace = true }
 itertools = { workspace = true }
+iter-enum = { workspace = true }
+left-right = { workspace = true }
+metrics = { workspace = true }
+rand = { workspace = true }
+smallvec = { workspace = true, features = ["union"] }
+thiserror = { workspace = true }
+triomphe = { workspace = true }
 
+partial-map = { path = "../partial-map" }
 readyset-client = { path = "../readyset-client" }
 readyset-util = { path = "../readyset-util" }
-partial-map = { path = "../partial-map" }
-thiserror = { workspace = true }
-iter-enum = { workspace = true }
 
 [dev-dependencies]
 proptest = { workspace = true }

reader-map/Cargo.toml

@@ -5,12 +5,14 @@ use std::hash::{BuildHasher, Hash};
 
 use iter_enum::{ExactSizeIterator, Iterator};
 use itertools::Either;
+use metrics::{register_histogram, Histogram};
 use partial_map::PartialMap;
 use readyset_client::internal::IndexType;
 use readyset_util::ranges::{Bound, RangeBounds};
 
 use crate::eviction::{EvictionMeta, EvictionStrategy};
-use crate::values::Values;
+use crate::recorded::{READER_MAP_LIFETIMES, READER_MAP_UPDATES};
+use crate::values::{Metrics, Values};
 
 /// Represents a miss when looking up a range.
 ///
@@ -155,18 +157,15 @@ impl<K, V, S> Data<K, V, S> {
         }
     }
 
-    pub(crate) fn remove_range<R>(&mut self, range: R)
+    pub(crate) fn remove_range<R, F>(&mut self, range: R, f: F)
     where
         R: RangeBounds<K>,
         K: Ord + Clone,
+        F: Fn(&Metrics),
     {
         match self {
             Self::BTreeMap(map, ..) => {
-                // Returns an iterator, but we don't actually care about the elements. Dropping
-                // the iterator does not automatically remove the elements from the BTreeMap (as of
-                // nightly 2023-06-16), so we need to iterate by collect()'ing them
-                // first.
-                let _: Vec<_> = map.remove_range(range).collect();
+                map.remove_range(range).for_each(|(_, v)| f(v.metrics()));
             }
             Self::HashMap(..) => panic!("remove_range called on a HashMap reader_map"),
         }
@@ -312,6 +311,39 @@ where
     }
 }
 
+#[derive(Clone)]
+pub(crate) struct WriteMetrics {
+    // Captures the duration between updates at a given key (entry).
+    entry_updated: Histogram,
+    // Captures the lifetime of an entry, from the time it was first added
+    // until it's eviction.
+    lifetime_evict: Histogram,
+}
+
+impl WriteMetrics {
+    fn new() -> Self {
+        let entry_updated = register_histogram!(READER_MAP_UPDATES);
+        let lifetime_evict = register_histogram!(READER_MAP_LIFETIMES);
+
+        Self {
+            entry_updated,
+            lifetime_evict,
+        }
+    }
+
+    pub(crate) fn record_updated(&self, values: &Metrics) {
+        if let Some(duration) = values.last_update_interval() {
+            self.entry_updated.record(duration.as_millis() as f64);
+        }
+    }
+
+    pub(crate) fn record_evicted(&self, values: &Metrics) {
+        if let Some(lifetime) = values.lifetime() {
+            self.lifetime_evict.record(lifetime.as_millis() as f64);
+        }
+    }
+}
+
 pub(crate) struct Inner<K, V, M, T, S, I> {
     pub(crate) data: Data<K, V, S>,
     pub(crate) meta: M,
@@ -320,6 +352,7 @@ pub(crate) struct Inner<K, V, M, T, S, I> {
     pub(crate) hasher: S,
     pub(crate) eviction_strategy: EvictionStrategy,
     pub(crate) insertion_order: Option<I>,
+    pub(crate) metrics: WriteMetrics,
 }
 
 impl<K, V, M, T, S, I> fmt::Debug for Inner<K, V, M, T, S, I>
@@ -358,6 +391,7 @@ where
             hasher: self.hasher.clone(),
             eviction_strategy: self.eviction_strategy.clone(),
             insertion_order: self.insertion_order.clone(),
+            metrics: self.metrics.clone(),
         }
     }
 }
@@ -384,6 +418,7 @@ where
             hasher,
             eviction_strategy,
             insertion_order,
+            metrics: WriteMetrics::new(),
         }
     }
 

reader-map/src/inner.rs

@@ -241,6 +241,7 @@ mod error;
 mod eviction;
 mod inner;
 mod read;
+mod recorded;
 mod values;
 mod write;
 

reader-map/src/lib.rs

@@ -0,0 +1,18 @@
+//! Documents the set of metrics that are currently being recorded within
+//! instances of a reader-map.
+
+/// Histogram: The time interval in milliseconds between updates to a given key in the reader-map.
+/// Updates include inserts, updates, and deletions from the key's results set.
+///
+/// | Tag | Description |
+/// | --- | ----------- |
+/// | node_index | The node index of the reader node in the dataflow graph |
+pub const READER_MAP_UPDATES: &str = "readyset_reader.entry_updates_ms";
+
+/// Histogram: The time in milliseconds that an entry was "live" in the `reader-map`, as measured
+/// from first insert to eviction.
+///
+/// | Tag | Description |
+/// | --- | ----------- |
+/// | node_index | The node index of the reader node in the dataflow graph |
+pub const READER_MAP_LIFETIMES: &str = "readyset_reader.entry_lifetimes_ms";

reader-map/src/recorded.rs

@@ -191,6 +191,10 @@ impl<T> Values<T> {
     {
         Arc::make_mut(&mut self.values.0).retain(f)
     }
+
+    pub(crate) fn metrics(&self) -> &Metrics {
+        &self.metrics
+    }
 }
 
 impl<'a, T: 'a> IntoIterator for &'a Values<T> {

reader-map/src/values.rs

@@ -384,6 +384,7 @@ where
                 insertion_index,
                 timestamp,
             } => {
+                let metrics = self.metrics.clone();
                 let values = self.data_entry(key.clone(), eviction_meta);
                 // Always insert values in sorted order, even if no ordering method is provided,
                 // otherwise it will require a linear scan to remove a value
@@ -395,6 +396,7 @@ where
                 .unwrap_or_else(|i| i);
                 values.insert(insert_idx, value.clone(), *timestamp);
                 *insertion_index = Some(insert_idx);
+                metrics.record_updated(values.metrics());
             }
             Operation::RemoveValue {
                 key,
@@ -414,18 +416,33 @@ where
                         e.remove(remove_idx, *timestamp);
                         *removal_index = Some(remove_idx);
                     }
+                    // removing a value from a key is just "updating" that key
+                    self.metrics.record_updated(e.metrics());
                 }
             }
             Operation::AddRange(range) => self.data.add_range(range.clone()),
             Operation::AddFullRange => self.data.add_full_range(),
             Operation::Clear(key, eviction_meta) => {
                 self.data_entry(key.clone(), eviction_meta).clear()
+                // `clear()` is invoked on replay when filling a hole, will be followed
+                // by a call to `add()`. hence don't capture metrics on clear().
             }
             Operation::RemoveEntry(key) => {
-                self.data.remove(key);
+                let v = self.data.remove(key);
+
+                // upstream deletes flow through `RemoveValue`, thus these removes are either
+                // upstream evictions or a call to "mark a hole" (mostly likely an eviction).
+                if let Some(values) = v {
+                    self.metrics.record_evicted(values.metrics());
+                }
             }
             Operation::Purge => self.data.clear(),
-            Operation::RemoveRange(range) => self.data.remove_range(range.clone()),
+            Operation::RemoveRange(range) => {
+                // RemoveRange is only called on evictions (via marking a hole).
+                self.data.remove_range(range.clone(), |metrics| {
+                    self.metrics.record_evicted(metrics);
+                });
+            }
             Operation::Retain(key, predicate) => {
                 if let Some(e) = self.data.get_mut(key) {
                     let mut first = true;
@@ -503,7 +520,7 @@ where
             Operation::RemoveEntry(key) => {
                 self.data.remove(&key);
             }
-            Operation::RemoveRange(range) => self.data.remove_range(range),
+            Operation::RemoveRange(range) => self.data.remove_range(range, |_| {}),
             Operation::Purge => self.data.clear(),
             Operation::Retain(key, mut predicate) => {
                 if let Some(e) = self.data.get_mut(&key) {

reader-map/src/write.rs

@@ -187,24 +187,6 @@ fn mapref() {
     }
 }
 
-#[test]
-#[cfg_attr(miri, ignore)]
-// https://github.com/rust-lang/miri/issues/658
-fn panicked_reader_doesnt_block_writer() {
-    let (mut w, r) = reader_map::new();
-    w.insert(1, "a");
-    w.publish();
-
-    // reader panics
-    let r = std::panic::catch_unwind(move || r.get(&1).map(|_| panic!()));
-    r.unwrap_err();
-
-    // writer should still be able to continue
-    w.insert(1, "b");
-    w.publish();
-    w.publish();
-}
-
 #[test]
 fn read_after_drop() {
     let x = ('x', 42);