compute,storage: record wallclock lag histograms

This commit extends the two controllers' `refresh_wallclock_lag` methods
to also record `WallclockLagHistogram` introspection data. Both the
histogram refresh time (i.e. cadence of persistent writes) and the
period length can be configured through dyncfg flags.

Author: teskje

Cargo.lock

@@ -1081,7 +1081,7 @@ where
                     let op = StorageWriteOp::Append { updates };
                     storage.update_introspection_collection(type_, op);
                 }
-                WallclockLagHistory => {
+                WallclockLagHistory | WallclockLagHistogram => {
                     storage.append_introspection_updates(type_, updates);
                 }
                 _ => panic!("unexpected introspection type: {type_:?}"),

src/compute-client/src/controller.rs

@@ -26,7 +26,9 @@ use mz_compute_types::sinks::{
 };
 use mz_compute_types::sources::SourceInstanceDesc;
 use mz_compute_types::ComputeInstanceId;
-use mz_controller_types::dyncfgs::WALLCLOCK_LAG_HISTORY_REFRESH_INTERVAL;
+use mz_controller_types::dyncfgs::{
+    WALLCLOCK_LAG_HISTOGRAM_REFRESH_INTERVAL, WALLCLOCK_LAG_HISTORY_REFRESH_INTERVAL,
+};
 use mz_dyncfg::ConfigSet;
 use mz_expr::RowSetFinishing;
 use mz_ore::cast::CastFrom;
@@ -35,9 +37,10 @@ use mz_ore::now::NowFn;
 use mz_ore::tracing::OpenTelemetryContext;
 use mz_ore::{soft_assert_or_log, soft_panic_or_log};
 use mz_repr::adt::interval::Interval;
+use mz_repr::adt::timestamp::CheckedTimestamp;
 use mz_repr::refresh_schedule::RefreshSchedule;
-use mz_repr::{Datum, Diff, GlobalId, Row};
-use mz_storage_client::controller::IntrospectionType;
+use mz_repr::{Datum, DatumMap, Diff, GlobalId, Row};
+use mz_storage_client::controller::{IntrospectionType, WallclockLagHistogramPeriod};
 use mz_storage_types::read_holds::{self, ReadHold};
 use mz_storage_types::read_policy::ReadPolicy;
 use serde::Serialize;
@@ -291,6 +294,8 @@ pub(super) struct Instance<T: ComputeControllerTimestamp> {
     wallclock_lag: WallclockLagFn<T>,
     /// The last time `WallclockLagHistory` introspection was refreshed.
     wallclock_lag_history_last_refresh: Instant,
+    /// The last time `WallclockLagHistogram` introspection was refreshed.
+    wallclock_lag_histogram_last_refresh: Instant,
 
     /// Sender for updates to collection read holds.
     ///
@@ -555,17 +560,20 @@ impl<T: ComputeControllerTimestamp> Instance<T> {
         let refresh_history = !self.read_only
             && self.wallclock_lag_history_last_refresh.elapsed()
                 >= WALLCLOCK_LAG_HISTORY_REFRESH_INTERVAL.get(&self.dyncfg);
+        let refresh_histogram = !self.read_only
+            && self.wallclock_lag_histogram_last_refresh.elapsed()
+                >= WALLCLOCK_LAG_HISTOGRAM_REFRESH_INTERVAL.get(&self.dyncfg);
 
-        let mut history_updates = Vec::new();
-
-        let now = mz_ore::now::to_datetime((self.now)());
-        let now_tz = now.try_into().expect("must fit");
+        let now_ms = (self.now)();
+        let now_dt = mz_ore::now::to_datetime(now_ms);
+        let now_ts: CheckedTimestamp<_> = now_dt.try_into().expect("must fit");
 
         let frontier_lag = |frontier: &Antichain<_>| match frontier.as_option() {
             Some(ts) => (self.wallclock_lag)(ts),
             None => Duration::ZERO,
         };
 
+        let mut history_updates = Vec::new();
         for (replica_id, replica) in &mut self.replicas {
             for (collection_id, collection) in &mut replica.collections {
                 let lag = frontier_lag(&collection.write_frontier);
@@ -580,7 +588,7 @@ impl<T: ComputeControllerTimestamp> Instance<T> {
                             Datum::String(&collection_id.to_string()),
                             Datum::String(&replica_id.to_string()),
                             Datum::Interval(Interval::new(0, 0, max_lag_us)),
-                            Datum::TimestampTz(now_tz),
+                            Datum::TimestampTz(now_ts),
                         ]);
                         history_updates.push((row, 1));
                     }
@@ -591,14 +599,46 @@ impl<T: ComputeControllerTimestamp> Instance<T> {
                 };
             }
         }
-
         if !history_updates.is_empty() {
             self.deliver_introspection_updates(
                 IntrospectionType::WallclockLagHistory,
                 history_updates,
             );
             self.wallclock_lag_history_last_refresh = Instant::now();
         }
+
+        let histogram_period = WallclockLagHistogramPeriod::from_epoch_millis(now_ms, &self.dyncfg);
+
+        let mut histogram_updates = Vec::new();
+        for (collection_id, collection) in &mut self.collections {
+            if let Some(stash) = &mut collection.wallclock_lag_histogram_stash {
+                let lag = collection.shared.lock_write_frontier(|f| frontier_lag(f));
+                let bucket = lag.as_secs().next_power_of_two();
+
+                let key = (histogram_period, bucket);
+                *stash.entry(key).or_default() += 1;
+
+                if refresh_histogram {
+                    for ((period, lag), count) in std::mem::take(stash) {
+                        let row = Row::pack_slice(&[
+                            Datum::String(&collection_id.to_string()),
+                            Datum::UInt64(lag),
+                            Datum::TimestampTz(period.start),
+                            Datum::TimestampTz(period.end),
+                            Datum::Map(DatumMap::empty()),
+                        ]);
+                        histogram_updates.push((row, count));
+                    }
+                }
+            }
+        }
+        if !histogram_updates.is_empty() {
+            self.deliver_introspection_updates(
+                IntrospectionType::WallclockLagHistogram,
+                histogram_updates,
+            );
+            self.wallclock_lag_histogram_last_refresh = Instant::now();
+        }
     }
 
     /// Report updates (inserts or retractions) to the identified collection's dependencies.
@@ -819,6 +859,7 @@ impl<T: ComputeControllerTimestamp> Instance<T> {
             now: _,
             wallclock_lag: _,
             wallclock_lag_history_last_refresh,
+            wallclock_lag_histogram_last_refresh,
             read_hold_tx: _,
             replica_tx: _,
             replica_rx: _,
@@ -854,6 +895,8 @@ impl<T: ComputeControllerTimestamp> Instance<T> {
             .map(|(id, epoch)| (id.to_string(), epoch))
             .collect();
         let wallclock_lag_history_last_refresh = format!("{wallclock_lag_history_last_refresh:?}");
+        let wallclock_lag_histogram_last_refresh =
+            format!("{wallclock_lag_histogram_last_refresh:?}");
 
         let map = serde_json::Map::from_iter([
             field("initialized", initialized)?,
@@ -869,6 +912,10 @@ impl<T: ComputeControllerTimestamp> Instance<T> {
                 "wallclock_lag_history_last_refresh",
                 wallclock_lag_history_last_refresh,
             )?,
+            field(
+                "wallclock_lag_histogram_last_refresh",
+                wallclock_lag_histogram_last_refresh,
+            )?,
         ]);
         Ok(serde_json::Value::Object(map))
     }
@@ -934,6 +981,7 @@ where
             now,
             wallclock_lag,
             wallclock_lag_history_last_refresh: Instant::now(),
+            wallclock_lag_histogram_last_refresh: Instant::now(),
             read_hold_tx,
             replica_tx,
             replica_rx,
@@ -2155,6 +2203,12 @@ struct CollectionState<T: ComputeControllerTimestamp> {
 
     /// Introspection state associated with this collection.
     introspection: CollectionIntrospection<T>,
+
+    /// Frontier wallclock lag measurements stashed until the next `WallclockLagHistogram`
+    /// introspection update.
+    ///
+    /// If this is `None`, wallclock lag is not tracked for this collection.
+    wallclock_lag_histogram_stash: Option<BTreeMap<(WallclockLagHistogramPeriod, u64), i64>>,
 }
 
 impl<T: ComputeControllerTimestamp> CollectionState<T> {
@@ -2189,6 +2243,14 @@ impl<T: ComputeControllerTimestamp> CollectionState<T> {
             c.update_iter(updates);
         });
 
+        // In an effort to keep the produced wallclock lag introspection data small and
+        // predictable, we disable wallclock lag tracking for transient collections, i.e. slow-path
+        // select indexes and subscribes.
+        let wallclock_lag_histogram_stash = match collection_id.is_transient() {
+            true => None,
+            false => Some(Default::default()),
+        };
+
         Self {
             log_collection: false,
             dropped: false,
@@ -2200,6 +2262,7 @@ impl<T: ComputeControllerTimestamp> CollectionState<T> {
             storage_dependencies,
             compute_dependencies,
             introspection,
+            wallclock_lag_histogram_stash,
         }
     }
 

src/compute-client/src/controller/instance.rs

@@ -32,6 +32,18 @@ pub const WALLCLOCK_LAG_HISTORY_REFRESH_INTERVAL: Config<Duration> = Config::new
     "The interval at which to refresh `WallclockLagHistory` introspection.",
 );
 
+pub const WALLCLOCK_LAG_HISTOGRAM_REFRESH_INTERVAL: Config<Duration> = Config::new(
+    "wallclock_lag_histogram_refresh_interval",
+    Duration::from_secs(60),
+    "The interval at which to refresh `WallclockLagHistogram` introspection.",
+);
+
+pub const WALLCLOCK_LAG_HISTOGRAM_PERIOD_LENGTH: Config<Duration> = Config::new(
+    "wallclock_lag_histogram_period_length",
+    Duration::from_secs(24 * 60 * 60),
+    "The period length of histograms in `WallclockLagHistogram` introspection.",
+);
+
 pub const ENABLE_TIMELY_ZERO_COPY: Config<bool> = Config::new(
     "enable_timely_zero_copy",
     false,
@@ -56,6 +68,8 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&CONTROLLER_PAST_GENERATION_REPLICA_CLEANUP_RETRY_INTERVAL)
         .add(&ENABLE_0DT_DEPLOYMENT_SOURCES)
         .add(&WALLCLOCK_LAG_HISTORY_REFRESH_INTERVAL)
+        .add(&WALLCLOCK_LAG_HISTOGRAM_REFRESH_INTERVAL)
+        .add(&WALLCLOCK_LAG_HISTOGRAM_PERIOD_LENGTH)
         .add(&ENABLE_TIMELY_ZERO_COPY)
         .add(&ENABLE_TIMELY_ZERO_COPY_LGALLOC)
         .add(&TIMELY_ZERO_COPY_LIMIT)

src/controller-types/src/dyncfgs.rs

@@ -20,6 +20,8 @@ itertools = { version = "0.12.1" }
 maplit = "1.0.2"
 mz-ccsr = { path = "../ccsr" }
 mz-cluster-client = { path = "../cluster-client" }
+mz-controller-types = { path = "../controller-types" }
+mz-dyncfg = { path = "../dyncfg" }
 mz-dyncfgs = { path = "../dyncfgs" }
 mz-kafka-util = { path = "../kafka-util" }
 mz-ore = { path = "../ore", features = ["async", "process", "tracing"] }

src/storage-client/Cargo.toml

@@ -27,11 +27,16 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use async_trait::async_trait;
+use chrono::{DateTime, Utc};
 use differential_dataflow::lattice::Lattice;
 use mz_cluster_client::client::ClusterReplicaLocation;
 use mz_cluster_client::ReplicaId;
+use mz_controller_types::dyncfgs::WALLCLOCK_LAG_HISTOGRAM_PERIOD_LENGTH;
+use mz_dyncfg::ConfigSet;
+use mz_ore::soft_panic_or_log;
 use mz_persist_client::batch::ProtoBatch;
 use mz_persist_types::{Codec64, Opaque, ShardId};
+use mz_repr::adt::timestamp::CheckedTimestamp;
 use mz_repr::{Diff, GlobalId, RelationDesc, RelationVersion, Row};
 use mz_storage_types::configuration::StorageConfiguration;
 use mz_storage_types::connections::inline::InlinedConnection;
@@ -842,6 +847,36 @@ impl<T> MonotonicAppender<T> {
     }
 }
 
+/// The period covered by a wallclock lag histogram, represented as a `[start, end)` range.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub struct WallclockLagHistogramPeriod {
+    pub start: CheckedTimestamp<DateTime<Utc>>,
+    pub end: CheckedTimestamp<DateTime<Utc>>,
+}
+
+impl WallclockLagHistogramPeriod {
+    /// Construct a `WallclockLagHistogramPeriod` from the given epoch timestamp and dyncfg.
+    pub fn from_epoch_millis(epoch_ms: u64, dyncfg: &ConfigSet) -> Self {
+        let length = WALLCLOCK_LAG_HISTOGRAM_PERIOD_LENGTH.get(dyncfg);
+        let length_ms = u64::try_from(length.as_millis()).unwrap_or_else(|_| {
+            soft_panic_or_log!("excessive wallclock lag histogram period length: {length:?}");
+            let default = WALLCLOCK_LAG_HISTOGRAM_PERIOD_LENGTH.default();
+            u64::try_from(default.as_millis()).unwrap()
+        });
+        let length_ms = std::cmp::max(length_ms, 1);
+
+        let start_ms = epoch_ms - (epoch_ms % length_ms);
+        let start_dt = mz_ore::now::to_datetime(start_ms);
+        let start = start_dt.try_into().expect("must fit");
+
+        let end_ms = start_ms + length_ms;
+        let end_dt = mz_ore::now::to_datetime(end_ms);
+        let end = end_dt.try_into().expect("must fit");
+
+        Self { start, end }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;