diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index 6119f0ca4ecae..ef931df25781a 100644 --- a/doc/user/data/metrics.yml +++ b/doc/user/data/metrics.yml @@ -844,6 +844,18 @@ metrics: - response_type source: src/compute-client/src/metrics.rs visibility: internal +- name: mz_compute_shared_arrangement_held_count + help: How many of the shared arrangements this worker publishes are kept from compacting by the importing runtime or a reader. + labels: + - worker_id + source: src/compute/src/metrics.rs + visibility: internal +- name: mz_compute_shared_arrangement_hold_gap_ms + help: The largest gap, over the shared arrangements this worker publishes, between the compaction frontier it was told to apply and the one it applied. + labels: + - worker_id + source: src/compute/src/metrics.rs + visibility: internal - name: mz_connection_status help: Count of completed network connections, by status labels: diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index c1aeae24ef33f..0560466e39393 100644 --- a/src/compute/src/compute_state.rs +++ b/src/compute/src/compute_state.rs @@ -1380,6 +1380,25 @@ impl<'a> ActiveComputeState<'a> { .replica_expiration_remaining_seconds .set(remaining) } + + // Only the publishing runtime is held back, and only it has slots in the registry under its + // own worker ordinal. Reporting from the interactive runtime as well would repeat the same + // numbers under a second `role` label, so its series stays at zero: an extra zero leaves a + // `sum` or a `max` over the label correct, where a duplicate would not. + if self.compute_state.role() != ComputeRuntimeRole::Interactive { + let (gap, held) = self + .compute_state + .sharing_registry + .hold_gaps(self.timely_worker.index()); + self.compute_state + .metrics + .shared_arrangement_hold_gap_ms + .set(gap); + self.compute_state + .metrics + .shared_arrangement_held_count + .set(u64::cast_from(held)); + } } /// Gives `peek` a turn on the worker if this activation's budget has one left, and queues it diff --git a/src/compute/src/metrics.rs b/src/compute/src/metrics.rs index 7586c181f5762..b57d39447f597 100644 --- a/src/compute/src/metrics.rs +++ b/src/compute/src/metrics.rs @@ -89,6 +89,10 @@ pub struct ComputeMetrics { // subscribes subscribe_snapshots_skipped_total: IntCounter, + + // shared arrangements + shared_arrangement_hold_gap_ms: raw::UIntGaugeVec, + shared_arrangement_held_count: raw::UIntGaugeVec, } /// Applies the per-role const label to `opts`, unless `role` is `Solo`. @@ -305,6 +309,16 @@ impl ComputeMetrics { name: "mz_subscribe_snapshots_skipped_total", help: "The number of collection snapshots that were skipped by the subscribe snapshot optimization.", ), role)), + shared_arrangement_hold_gap_ms: registry.register(with_role(metric!( + name: "mz_compute_shared_arrangement_hold_gap_ms", + help: "The largest gap, over the shared arrangements this worker publishes, between the compaction frontier it was told to apply and the one it applied.", + var_labels: ["worker_id"], + ), role)), + shared_arrangement_held_count: registry.register(with_role(metric!( + name: "mz_compute_shared_arrangement_held_count", + help: "How many of the shared arrangements this worker publishes are kept from compacting by the importing runtime or a reader.", + var_labels: ["worker_id"], + ), role)), } } @@ -357,6 +371,12 @@ impl ComputeMetrics { let shared_row_heap_capacity_bytes = self .shared_row_heap_capacity_bytes .with_label_values(&[&worker]); + let shared_arrangement_hold_gap_ms = self + .shared_arrangement_hold_gap_ms + .with_label_values(&[&worker]); + let shared_arrangement_held_count = self + .shared_arrangement_held_count + .with_label_values(&[&worker]); WorkerMetrics { worker_label: worker, @@ -385,6 +405,8 @@ impl ComputeMetrics { replica_expiration_timestamp_seconds, replica_expiration_remaining_seconds, shared_row_heap_capacity_bytes, + shared_arrangement_hold_gap_ms, + shared_arrangement_held_count, } } } @@ -454,6 +476,10 @@ pub struct WorkerMetrics { pub(crate) replica_expiration_remaining_seconds: raw::Gauge, /// Heap capacity of the shared row. shared_row_heap_capacity_bytes: UIntGauge, + /// The largest compaction gap over the shared arrangements this worker publishes. + pub(crate) shared_arrangement_hold_gap_ms: UIntGauge, + /// How many of those arrangements are held back at all. + pub(crate) shared_arrangement_held_count: UIntGauge, } impl WorkerMetrics { diff --git a/src/compute/src/shared_trace/publish.rs b/src/compute/src/shared_trace/publish.rs index 049b70c68f612..1aea4b7082281 100644 --- a/src/compute/src/shared_trace/publish.rs +++ b/src/compute/src/shared_trace/publish.rs @@ -121,6 +121,15 @@ where self.shared.upper() } + /// The `(applied, requested)` logical compaction frontiers of this point. + /// + /// `applied` is what the publishing runtime's trace compacted to, `requested` what its + /// controller stream asked for. They differ by exactly what the importing runtime's standing + /// hold, or a live reader, is keeping the publisher from shedding. + pub(crate) fn logical_frontiers(&self) -> (Antichain, Antichain) { + (self.shared.since(), self.shared.writer_since()) + } + /// Why this point would refuse an `as_of`. See [`Diagnostics`]. pub(crate) fn diagnostics(&self) -> Diagnostics { Diagnostics { diff --git a/src/compute/src/shared_trace/tests.rs b/src/compute/src/shared_trace/tests.rs index 1ba5ee5d0f537..6dbcc138aab92 100644 --- a/src/compute/src/shared_trace/tests.rs +++ b/src/compute/src/shared_trace/tests.rs @@ -637,6 +637,69 @@ fn handle_at_mints_at_as_of_or_refuses() { }); } +/// A point reports both the frontier its writer was told to compact to and the one the holds let +/// it apply. +/// +/// The two differ by exactly what the importing runtime is withholding, which is the coupling this +/// publication introduces, and the pair is the only place that coupling is observable. +#[mz_ore::test] +fn logical_frontiers_report_what_the_holds_withhold() { + timely::execute_directly(move |worker| { + let (mut writer, published, mut input) = worker.dataflow::(|scope| { + let (input, collection) = scope.new_collection::<(Row, Row), Diff>(); + let arranged = collection.mz_arrange::< + ColumnationChunker<_>, + RowRowBatcher<_, _>, + RowRowBuilder<_, _>, + RowRowSpine<_, _>, + >("gap oks"); + let writer = arranged.trace.clone(); + let published = adopt_fresh(&arranged); + (writer, published, input) + }); + for t in 0..5 { + tick( + worker, + &mut input, + Timestamp::from(t), + Timestamp::from(t + 1), + ); + } + + // The importing runtime is four behind the controller's request, so the writer applies + // four and reports ten. + let held = Antichain::from_elem(Timestamp::from(4_u64)); + let target = Antichain::from_elem(Timestamp::from(10_u64)); + published.note_standing_hold(&held); + writer.set_logical_compaction(target.borrow()); + tick( + worker, + &mut input, + Timestamp::from(10_u64), + Timestamp::from(11_u64), + ); + assert_eq!( + published.logical_frontiers(), + (held, target.clone()), + "the applied frontier must report the hold and the requested one the controller" + ); + + // The importing runtime catches up and the gap closes. + published.note_standing_hold(&target); + tick( + worker, + &mut input, + Timestamp::from(11_u64), + Timestamp::from(12_u64), + ); + assert_eq!( + published.logical_frontiers(), + (target.clone(), target), + "a caught-up importer must leave no gap" + ); + }); +} + /// A consumer forwarding an empty input frontier releases its hold rather than recording an /// empty one. /// diff --git a/src/compute/src/sharing.rs b/src/compute/src/sharing.rs index ed6a4f484077f..b37822db0fd6c 100644 --- a/src/compute/src/sharing.rs +++ b/src/compute/src/sharing.rs @@ -327,6 +327,42 @@ impl ArrangementSharingRegistry { Some((slot.oks.handle(), slot.errs.handle())) } + /// How far the readers' holds keep the publishing runtime from compacting, over every + /// arrangement published on `worker_index`. + /// + /// Returns the largest gap in milliseconds between an arrangement's requested and applied + /// logical compaction frontiers, and how many arrangements have any gap at all. Aliases share + /// their target's points, so points are counted once rather than once per id. A point whose + /// requested frontier is empty is counted as held but contributes no gap: its collection is + /// being dropped and has no finite frontier left to subtract from. + pub(crate) fn hold_gaps(&self, worker_index: usize) -> (u64, usize) { + let inner = self.lock(); + let mut seen = BTreeSet::new(); + let mut max_gap = 0; + let mut held = 0; + for slots in inner.map.values() { + let Some(Some(slot)) = slots.get(worker_index) else { + continue; + }; + if !seen.insert(Arc::as_ptr(slot)) { + continue; + } + let (applied, requested) = slot.oks.logical_frontiers(); + match (applied.as_option(), requested.as_option()) { + (Some(applied), Some(requested)) => { + let gap = u64::from(*requested).saturating_sub(u64::from(*applied)); + if gap > 0 { + held += 1; + max_gap = max_gap.max(gap); + } + } + (Some(_), None) => held += 1, + _ => {} + } + } + (max_gap, held) + } + /// The accumulated `oks` logical holds registered against `id` on `worker_index`, if published. /// /// Test-only. Minting a handle to observe the published frontiers cannot distinguish a live diff --git a/src/timely-util/src/shared_trace.rs b/src/timely-util/src/shared_trace.rs index 5f4dbeab3352d..23a199b3664d0 100644 --- a/src/timely-util/src/shared_trace.rs +++ b/src/timely-util/src/shared_trace.rs @@ -87,6 +87,12 @@ struct State { upper: Antichain, /// The inner trace's logical compaction frontier. Reads at times not beyond it are not accurate. logical: Antichain, + /// The frontier the writer asked for, before meeting the readers' holds. + /// + /// Equal to `logical` when nothing holds the trace back. Above it by exactly the amount a + /// reader is keeping the writer from compacting, which is the coupling this publication + /// introduces and the only place it is observable. + writer_logical: Antichain, /// The inner trace's physical compaction frontier. physical: Antichain, /// Readers' logical holds. @@ -147,6 +153,7 @@ impl Shared { chain: Vec::new(), upper: minimum.clone(), logical: minimum.clone(), + writer_logical: minimum.clone(), physical: minimum, remote_logical: MutableAntichain::new(), remote_physical: MutableAntichain::new(), @@ -171,6 +178,12 @@ impl Shared { self.lock().logical.clone() } + /// The logical compaction frontier the writer asked for, before the readers' holds were met + /// into it. At or beyond [`Shared::since`], and equal to it when no reader holds this point. + pub fn writer_since(&self) -> Antichain { + self.lock().writer_logical.clone() + } + /// The published `(since, upper)`, read together. pub fn frontiers(&self) -> (Antichain, Antichain) { let state = self.lock(); @@ -300,6 +313,7 @@ impl SharedSpine { // The frontiers the `TraceReader` impl below last saw. Reading them from `inner` here // would need `&mut`, and they are equal. state.logical = self.local_logical.clone(); + state.writer_logical = self.local_logical.clone(); state.physical = self.local_physical.clone(); state.live_queues() }; @@ -437,6 +451,7 @@ impl SharedSpine { let logical = self.local_logical.meet(&remote_logical); for state in guards.iter_mut() { state.logical = logical.clone(); + state.writer_logical = self.local_logical.clone(); } drop(guards); drop(attachment);