diff --git a/.agents/skills/mz-release-signoff/SKILL.md b/.agents/skills/mz-release-signoff/SKILL.md index b8aec0715ca49..892542107e4cd 100644 --- a/.agents/skills/mz-release-signoff/SKILL.md +++ b/.agents/skills/mz-release-signoff/SKILL.md @@ -290,4 +290,4 @@ The per-area references list the dead panels found so far. Check for new ones wh **An implausibly constant quantile is a bucket artifact, not stability.** A `histogram_quantile` landing inside one wide bucket returns the bucket boundary and cannot move, which reads as a rock-steady latency. Two adapter panels do this. When a quantile is stable to five significant figures while its counters advance, switch to `rate(_sum) / rate(_count)`. -**Counting series is sometimes the measurement.** Several metrics are per-entity gauges whose only useful aggregate is a series count: `mz_persist_shard_upper` for shards, `mz_balancer_metadata_seconds` for balancers, `mz_persist_metadata_seconds` by `version` for persist client builds. Similarly `mz_source_progress` is a millisecond frontier timestamp, so a healthy series contributes exactly 1000 to its rate and the panel's absolute value is really a series count in disguise. +**Counting series is sometimes the measurement.** Several metrics are per-entity gauges whose only useful aggregate is a series count: `mz_balancer_metadata_seconds` for balancers, `mz_persist_metadata_seconds` by `version` for persist client builds. Similarly `mz_source_progress` is a millisecond frontier timestamp, so a healthy series contributes exactly 1000 to its rate and the panel's absolute value is really a series count in disguise. diff --git a/.agents/skills/mz-release-signoff/references/persist.md b/.agents/skills/mz-release-signoff/references/persist.md index 3c17f59e9a984..85bb9ab094037 100644 --- a/.agents/skills/mz-release-signoff/references/persist.md +++ b/.agents/skills/mz-release-signoff/references/persist.md @@ -34,7 +34,7 @@ Worth adding to the sweep even though the panel omits them: `mz_persist_compacti | `mz_persist_compaction_seconds`, `_requested`, `_applied`, `_bytes`, `_goodbytes` | counter | Compaction. Requested minus applied equals the noop and dropped counts. | | `mz_persist_gc_seconds`, `_started`, `_finished`, `_noop`, `_skipped`, `_merged_reqs` | counter | Garbage collection. | | `mz_persist_retry_retries_count`, `_started_count`, `_finished_count`, `_sleep_seconds` | counter | The panel excludes `op="next_listen_batch"`, which has its own panel because it retries by design. | -| `mz_persist_shard_upper` | gauge, per shard | Counting distinct `shard` labels gives the shard count. | +| `mz_persist_shard_count` | gauge, per process | Shards the process holds a handle to. environmentd holds one per collection, so its value is the environment's shard count. | | `mz_persist_read_batch_part_bytes`, `_count` | counter, by `op` | The `op="unindexed"` slice is the `unindexed reads` panel. Rises by roughly 3x during rehydration. | | `mz_persist_pushdown_parts_filtered_count`, `_fetched_count`, `_inline_count`, `_faked_count` and their `_bytes` variants | counter | The ratio panels divide one by the sum of all four. | | `mz_persist_blob_cache_hits_bytes`, `_hits_blobs`, `_evictions` | counter | | @@ -65,7 +65,7 @@ Each entry states a property that holds at any fleet size, followed by the measu **Persist reports its own build independently of `mz_version`.** `mz_persist_metadata_seconds` carries a `version` label, and the `# processes by version` panel counts by it. Use it to confirm a rollout reached the persist clients. -**Counting distinct `shard` labels on `mz_persist_shard_upper` is how the shard count is obtained.** There is no shard-count gauge. +**`mz_persist_shard_count` on the environmentd pod is the shard count.** Summing it across pods overcounts, because clusterd pods hold handles to the same shards. ## Order of magnitude diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index 5648f85b5a690..c1e6844d41b36 100644 --- a/doc/user/data/metrics.yml +++ b/doc/user/data/metrics.yml @@ -1646,24 +1646,15 @@ metrics: source: src/persist-client/src/internal/metrics.rs visibility: internal - name: mz_persist_backpressure_emitted_bytes - help: A counter with the number of emitted bytes. - labels: - - name - - shard + help: bytes emitted by backpressure operators source: src/persist-client/src/internal/metrics.rs visibility: internal - name: mz_persist_backpressure_last_backpressured_bytes - help: The last count of bytes we are waiting to be retired in the operator. This cannot be directly compared to `retired_bytes`, but CAN indicate that backpressure is happening. - labels: - - name - - shard + help: sum over backpressure operators of the inflight bytes each last stalled on; not comparable to retired bytes, but nonzero growth indicates backpressure is happening source: src/persist-client/src/internal/metrics.rs visibility: internal - name: mz_persist_backpressure_retired_bytes - help: A counter with the number of bytes retired by downstream processing. - labels: - - name - - shard + help: bytes retired by processing downstream of backpressure operators source: src/persist-client/src/internal/metrics.rs visibility: internal - name: mz_persist_blob_cache_evictions @@ -2198,6 +2189,18 @@ metrics: help: count of grpc errors received source: src/persist-client/src/internal/metrics.rs visibility: internal +- name: mz_persist_pubsub_client_receiver_diff_applied + help: number of diffs received via pubsub that applied + source: src/persist-client/src/internal/metrics.rs + visibility: internal +- name: mz_persist_pubsub_client_receiver_diff_not_applied_out_of_order + help: number of diffs received via pubsub that did not apply due to out-of-order delivery + source: src/persist-client/src/internal/metrics.rs + visibility: internal +- name: mz_persist_pubsub_client_receiver_diff_not_applied_stale + help: number of diffs received via pubsub that did not apply due to staleness + source: src/persist-client/src/internal/metrics.rs + visibility: internal - name: mz_persist_pubsub_client_receiver_state_push_diff_fast_path help: count fast-path state push_diff calls source: src/persist-client/src/internal/metrics.rs @@ -2476,14 +2479,6 @@ metrics: - shard source: src/persist-client/src/internal/metrics.rs visibility: internal -- name: mz_persist_shard_batch_part_version_bytes - help: total bytes in batch parts by shard and version - labels: - - name - - shard - - version - source: src/persist-client/src/internal/metrics.rs - visibility: internal - name: mz_persist_shard_batch_part_version_count help: count of batch parts by shard and version labels: @@ -2573,20 +2568,6 @@ metrics: - shard source: src/persist-client/src/internal/metrics.rs visibility: internal -- name: mz_persist_shard_inline_backpressure_count - help: count of CaA attempts retried because of inline backpressure - labels: - - name - - shard - source: src/persist-client/src/internal/metrics.rs - visibility: internal -- name: mz_persist_shard_inline_part_bytes - help: total size of parts inline in shard metadata - labels: - - name - - shard - source: src/persist-client/src/internal/metrics.rs - visibility: internal - name: mz_persist_shard_inline_part_count help: count of parts inline in shard metadata labels: @@ -2601,13 +2582,6 @@ metrics: - shard source: src/persist-client/src/internal/metrics.rs visibility: internal -- name: mz_persist_shard_live_writers - help: number of writers that have recently appended updates to this shard - labels: - - name - - shard - source: src/persist-client/src/internal/metrics.rs - visibility: internal - name: mz_persist_shard_noncompact_batches help: number of batches in the shard that aren't compact and have no ongoing compaction labels: @@ -2615,34 +2589,6 @@ metrics: - shard source: src/persist-client/src/internal/metrics.rs visibility: internal -- name: mz_persist_shard_pubsub_diff_applied - help: number of diffs received via pubsub that applied - labels: - - name - - shard - source: src/persist-client/src/internal/metrics.rs - visibility: internal -- name: mz_persist_shard_pubsub_diff_not_applied_out_of_order - help: number of diffs received via pubsub that did not apply due to out-of-order delivery - labels: - - name - - shard - source: src/persist-client/src/internal/metrics.rs - visibility: internal -- name: mz_persist_shard_pubsub_diff_not_applied_stale - help: number of diffs received via pubsub that did not apply due to staleness - labels: - - name - - shard - source: src/persist-client/src/internal/metrics.rs - visibility: internal -- name: mz_persist_shard_rewrite_part_count - help: count of batch parts with rewrites by shard - labels: - - name - - shard - source: src/persist-client/src/internal/metrics.rs - visibility: internal - name: mz_persist_shard_rollup_count help: count of rollups by shard labels: @@ -2657,13 +2603,6 @@ metrics: - shard source: src/persist-client/src/internal/metrics.rs visibility: internal -- name: mz_persist_shard_schema_registry_version_count - help: count of versions in the schema registry - labels: - - name - - shard - source: src/persist-client/src/internal/metrics.rs - visibility: internal - name: mz_persist_shard_seqnos_held help: maximum count of gc-ineligible states by shard labels: @@ -2678,13 +2617,6 @@ metrics: - shard source: src/persist-client/src/internal/metrics.rs visibility: internal -- name: mz_persist_shard_since - help: since by shard - labels: - - name - - shard - source: src/persist-client/src/internal/metrics.rs - visibility: internal - name: mz_persist_shard_spine_batch_count help: count of spine batches by shard labels: @@ -2692,13 +2624,6 @@ metrics: - shard source: src/persist-client/src/internal/metrics.rs visibility: internal -- name: mz_persist_shard_stale_version - help: indicates whether the current version of the shard is less than the current version of the code - labels: - - name - - shard - source: src/persist-client/src/internal/metrics.rs - visibility: internal - name: mz_persist_shard_unconsolidated_snapshot help: in snapshot_and_read, the number of times consolidating the raw data wasn't enough to produce consolidated output labels: @@ -2713,13 +2638,6 @@ metrics: - shard source: src/persist-client/src/internal/metrics.rs visibility: internal -- name: mz_persist_shard_upper - help: upper by shard - labels: - - name - - shard - source: src/persist-client/src/internal/metrics.rs - visibility: internal - name: mz_persist_shard_usage_current_state_batches_bytes help: data in batches/parts referenced by current version of state labels: @@ -2783,6 +2701,10 @@ metrics: - worker_id source: src/persist-client/src/internal/metrics.rs visibility: internal +- name: mz_persist_stale_shard_count + help: count of shards on this process whose persisted state version is behind this process's build version; per-process, so summing across processes counts (shard, process) pairs, not distinct shards + source: src/persist-client/src/internal/metrics.rs + visibility: internal - name: mz_persist_state_apply_spine_fast_path help: count of spine diff applications that hit the fast path source: src/persist-client/src/internal/metrics.rs diff --git a/src/persist-client/src/cache.rs b/src/persist-client/src/cache.rs index f34baeccfd200..60588e7587ca6 100644 --- a/src/persist-client/src/cache.rs +++ b/src/persist-client/src/cache.rs @@ -415,17 +415,23 @@ where "applied pushed diff {}. seqno {} -> {}.", state.shard_id, seqno_before, state.seqno ); - self.shard_metrics.pubsub_push_diff_applied.inc(); + self.metrics.pubsub_client.receiver.diff_applied.inc(); } else { debug!( "failed to apply pushed diff {}. seqno {} vs diff {}", state.shard_id, seqno_before, diff.seqno ); if diff.seqno <= seqno_before { - self.shard_metrics.pubsub_push_diff_not_applied_stale.inc(); + self.metrics + .pubsub_client + .receiver + .diff_not_applied_stale + .inc(); } else { - self.shard_metrics - .pubsub_push_diff_not_applied_out_of_order + self.metrics + .pubsub_client + .receiver + .diff_not_applied_out_of_order .inc(); } } @@ -627,6 +633,9 @@ pub(crate) struct LockingTypedState { notifier: StateWatchNotifier, cfg: Arc, metrics: Arc, + // Retained only to keep this shard's per-shard series registered for as long + // as the state is cached; nothing reads it through this handle anymore. Don't + // drop it as "unused" without moving that lifetime guarantee elsewhere. shard_metrics: Arc, update_semaphore: AwaitableState>, /// A [SchemaCacheMaps], but stored as an Any so the `: Codec` bounds diff --git a/src/persist-client/src/internal/metrics.rs b/src/persist-client/src/internal/metrics.rs index e21908f9e5223..ef5f5181712e2 100644 --- a/src/persist-client/src/internal/metrics.rs +++ b/src/persist-client/src/internal/metrics.rs @@ -12,6 +12,7 @@ use async_stream::stream; use mz_persist_types::stats::PartStatsMetrics; use std::collections::BTreeMap; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, Weak}; use std::time::{Duration, Instant}; use tokio::sync::{OnceCell, OwnedSemaphorePermit, Semaphore}; @@ -24,9 +25,8 @@ use mz_ore::cast::{CastFrom, CastLossy}; use mz_ore::instrument; use mz_ore::metric; use mz_ore::metrics::{ - ComputedGauge, ComputedIntGauge, ComputedUIntGauge, Counter, DeleteOnDropCounter, - DeleteOnDropGauge, IntCounter, MakeCollector, MetricVecExt, MetricsRegistry, UIntGauge, - UIntGaugeVec, raw, + ComputedGauge, ComputedUIntGauge, Counter, DeleteOnDropCounter, DeleteOnDropGauge, IntCounter, + MakeCollector, MetricsRegistry, UIntGauge, UIntGaugeVec, raw, }; use mz_ore::stats::histogram_seconds_buckets; use mz_persist::location::{ @@ -107,6 +107,8 @@ pub struct Metrics { /// Metrics for the persist sink. pub sink: SinkMetrics, + /// Metrics for the persist_source backpressure operator. + pub backpressure: BackpressureMetrics, /// Metrics for S3-backed blob implementation pub s3_blob: S3BlobMetrics, @@ -169,6 +171,7 @@ impl Metrics { inline: InlineMetrics::new(registry), semaphore: SemaphoreMetrics::new(cfg.clone(), registry.clone()), sink: SinkMetrics::new(registry), + backpressure: BackpressureMetrics::new(registry), s3_blob, blob_hedge: BlobHedgeMetrics::new(registry), postgres_consensus: PostgresClientMetrics::new(registry, "mz_persist"), @@ -1258,16 +1261,17 @@ pub struct ShardsMetrics { // Unlike all the other metrics in here, ShardsMetrics intentionally uses // the DeleteOnDrop wrappers. A process might stop using a shard (drop all // handles to it) but e.g. the set of commands never changes. - _count: ComputedIntGauge, - since: mz_ore::metrics::IntGaugeVec, - upper: mz_ore::metrics::IntGaugeVec, + // + // The process-level shard aggregates (`mz_persist_shard_count`, + // `mz_persist_stale_shard_count`) are not fields here: they live in a + // `ShardsAggregateMetrics` collector that the registry owns, sharing this + // struct's `shards` map so one scrape walk feeds both. encoded_rollup_size: mz_ore::metrics::UIntGaugeVec, encoded_diff_size: mz_ore::metrics::IntCounterVec, hollow_batch_count: mz_ore::metrics::UIntGaugeVec, spine_batch_count: mz_ore::metrics::UIntGaugeVec, batch_part_count: mz_ore::metrics::UIntGaugeVec, batch_part_version_count: mz_ore::metrics::UIntGaugeVec, - batch_part_version_bytes: mz_ore::metrics::UIntGaugeVec, update_count: mz_ore::metrics::UIntGaugeVec, rollup_count: mz_ore::metrics::UIntGaugeVec, largest_batch_size: mz_ore::metrics::UIntGaugeVec, @@ -1283,25 +1287,13 @@ pub struct ShardsMetrics { usage_referenced_not_current_state_bytes: mz_ore::metrics::UIntGaugeVec, usage_not_leaked_not_referenced_bytes: mz_ore::metrics::UIntGaugeVec, usage_leaked_bytes: mz_ore::metrics::UIntGaugeVec, - pubsub_push_diff_applied: mz_ore::metrics::IntCounterVec, - pubsub_push_diff_not_applied_stale: mz_ore::metrics::IntCounterVec, - pubsub_push_diff_not_applied_out_of_order: mz_ore::metrics::IntCounterVec, - stale_version: mz_ore::metrics::UIntGaugeVec, blob_gets: mz_ore::metrics::IntCounterVec, blob_sets: mz_ore::metrics::IntCounterVec, - live_writers: mz_ore::metrics::UIntGaugeVec, unconsolidated_snapshot: mz_ore::metrics::IntCounterVec, - backpressure_emitted_bytes: IntCounterVec, - backpressure_last_backpressured_bytes: UIntGaugeVec, - backpressure_retired_bytes: IntCounterVec, - rewrite_part_count: UIntGaugeVec, inline_part_count: UIntGaugeVec, - inline_part_bytes: UIntGaugeVec, compact_batches: UIntGaugeVec, compacting_batches: UIntGaugeVec, noncompact_batches: UIntGaugeVec, - schema_registry_version_count: UIntGaugeVec, - inline_backpressure_count: IntCounterVec, // We hand out `Arc` to read and write handles, but store it // here as `Weak`. This allows us to discover if it's no longer in use and // so we can remove it from the map. @@ -1311,29 +1303,8 @@ pub struct ShardsMetrics { impl ShardsMetrics { fn new(registry: &MetricsRegistry) -> Self { let shards = Arc::new(Mutex::new(BTreeMap::new())); - let shards_count = Arc::clone(&shards); + registry.register_collector(ShardsAggregateMetrics::new(Arc::clone(&shards))); ShardsMetrics { - _count: registry.register_computed_gauge( - metric!( - name: "mz_persist_shard_count", - help: "count of all active shards on this process", - ), - move || { - let mut ret = 0; - Self::compute(&shards_count, |_m| ret += 1); - ret - }, - ), - since: registry.register(metric!( - name: "mz_persist_shard_since", - help: "since by shard", - var_labels: ["shard", "name"], - )), - upper: registry.register(metric!( - name: "mz_persist_shard_upper", - help: "upper by shard", - var_labels: ["shard", "name"], - )), encoded_rollup_size: registry.register(metric!( name: "mz_persist_shard_rollup_size_bytes", help: "total encoded rollup size by shard", @@ -1364,11 +1335,6 @@ impl ShardsMetrics { help: "count of batch parts by shard and version", var_labels: ["shard", "name", "version"], )), - batch_part_version_bytes: registry.register(metric!( - name: "mz_persist_shard_batch_part_version_bytes", - help: "total bytes in batch parts by shard and version", - var_labels: ["shard", "name", "version"], - )), update_count: registry.register(metric!( name: "mz_persist_shard_update_count", help: "count of updates by shard", @@ -1444,26 +1410,6 @@ impl ShardsMetrics { help: "data reclaimable by a leaked blob detector", var_labels: ["shard", "name"], )), - pubsub_push_diff_applied: registry.register(metric!( - name: "mz_persist_shard_pubsub_diff_applied", - help: "number of diffs received via pubsub that applied", - var_labels: ["shard", "name"], - )), - pubsub_push_diff_not_applied_stale: registry.register(metric!( - name: "mz_persist_shard_pubsub_diff_not_applied_stale", - help: "number of diffs received via pubsub that did not apply due to staleness", - var_labels: ["shard", "name"], - )), - pubsub_push_diff_not_applied_out_of_order: registry.register(metric!( - name: "mz_persist_shard_pubsub_diff_not_applied_out_of_order", - help: "number of diffs received via pubsub that did not apply due to out-of-order delivery", - var_labels: ["shard", "name"], - )), - stale_version: registry.register(metric!( - name: "mz_persist_shard_stale_version", - help: "indicates whether the current version of the shard is less than the current version of the code", - var_labels: ["shard", "name"], - )), blob_gets: registry.register(metric!( name: "mz_persist_shard_blob_gets", help: "number of Blob::get calls for this shard", @@ -1474,48 +1420,16 @@ impl ShardsMetrics { help: "number of Blob::set calls for this shard", var_labels: ["shard", "name"], )), - live_writers: registry.register(metric!( - name: "mz_persist_shard_live_writers", - help: "number of writers that have recently appended updates to this shard", - var_labels: ["shard", "name"], - )), unconsolidated_snapshot: registry.register(metric!( name: "mz_persist_shard_unconsolidated_snapshot", help: "in snapshot_and_read, the number of times consolidating the raw data wasn't enough to produce consolidated output", var_labels: ["shard", "name"], )), - backpressure_emitted_bytes: registry.register(metric!( - name: "mz_persist_backpressure_emitted_bytes", - help: "A counter with the number of emitted bytes.", - var_labels: ["shard", "name"], - )), - backpressure_last_backpressured_bytes: registry.register(metric!( - name: "mz_persist_backpressure_last_backpressured_bytes", - help: "The last count of bytes we are waiting to be retired in \ - the operator. This cannot be directly compared to \ - `retired_bytes`, but CAN indicate that backpressure is happening.", - var_labels: ["shard", "name"], - )), - backpressure_retired_bytes: registry.register(metric!( - name: "mz_persist_backpressure_retired_bytes", - help:"A counter with the number of bytes retired by downstream processing.", - var_labels: ["shard", "name"], - )), - rewrite_part_count: registry.register(metric!( - name: "mz_persist_shard_rewrite_part_count", - help: "count of batch parts with rewrites by shard", - var_labels: ["shard", "name"], - )), inline_part_count: registry.register(metric!( name: "mz_persist_shard_inline_part_count", help: "count of parts inline in shard metadata", var_labels: ["shard", "name"], )), - inline_part_bytes: registry.register(metric!( - name: "mz_persist_shard_inline_part_bytes", - help: "total size of parts inline in shard metadata", - var_labels: ["shard", "name"], - )), compact_batches: registry.register(metric!( name: "mz_persist_shard_compact_batches", help: "number of fully compact batches in the shard", @@ -1531,16 +1445,6 @@ impl ShardsMetrics { help: "number of batches in the shard that aren't compact and have no ongoing compaction", var_labels: ["shard", "name"], )), - schema_registry_version_count: registry.register(metric!( - name: "mz_persist_shard_schema_registry_version_count", - help: "count of versions in the schema registry", - var_labels: ["shard", "name"], - )), - inline_backpressure_count: registry.register(metric!( - name: "mz_persist_shard_inline_backpressure_count", - help: "count of CaA attempts retried because of inline backpressure", - var_labels: ["shard", "name"], - )), shards, } } @@ -1582,12 +1486,65 @@ impl ShardsMetrics { } } +/// Process-level gauges derived from the shards map, collected in a single walk. +/// +/// Each is the sum of a per-shard quantity, so a `register_computed_gauge` +/// closure apiece would lock and walk the map once per gauge per scrape. This +/// collector shares [`ShardsMetrics`]'s `shards` map and folds all of them in +/// one pass instead. +#[derive(Debug)] +struct ShardsAggregateMetrics { + shards: Arc>>>, + count: GenericGauge, + stale_count: GenericGauge, +} + +impl ShardsAggregateMetrics { + fn new(shards: Arc>>>) -> Self { + ShardsAggregateMetrics { + shards, + count: MakeCollector::make_collector(metric!( + name: "mz_persist_shard_count", + help: "count of all active shards on this process", + )), + stale_count: MakeCollector::make_collector(metric!( + name: "mz_persist_stale_shard_count", + help: "count of shards on this process whose persisted state version \ + is behind this process's build version; per-process, so summing \ + across processes counts (shard, process) pairs, not distinct shards", + )), + } + } +} + +impl Collector for ShardsAggregateMetrics { + fn desc(&self) -> Vec<&Desc> { + let mut descs = self.count.desc(); + descs.extend(self.stale_count.desc()); + descs + } + + fn collect(&self) -> Vec { + let mut count = 0; + let mut stale_count = 0; + ShardsMetrics::compute(&self.shards, |m| { + count += 1; + if m.stale.load(Ordering::Relaxed) { + stale_count += 1; + } + }); + self.count.set(count); + self.stale_count.set(stale_count); + let mut families = self.count.collect(); + families.extend(self.stale_count.collect()); + families + } +} + #[derive(Debug)] pub struct ShardMetrics { pub shard_id: ShardId, pub name: String, - pub since: DeleteOnDropGauge>, - pub upper: DeleteOnDropGauge>, pub largest_batch_size: DeleteOnDropGauge>, pub latest_rollup_size: DeleteOnDropGauge>, pub encoded_diff_size: DeleteOnDropCounter>, @@ -1595,7 +1552,6 @@ pub struct ShardMetrics { pub spine_batch_count: DeleteOnDropGauge>, pub batch_part_count: DeleteOnDropGauge>, batch_part_version_count: mz_ore::metrics::UIntGaugeVec, - batch_part_version_bytes: mz_ore::metrics::UIntGaugeVec, batch_part_version_map: Mutex>, pub update_count: DeleteOnDropGauge>, pub rollup_count: DeleteOnDropGauge>, @@ -1611,25 +1567,17 @@ pub struct ShardMetrics { pub gc_finished: DeleteOnDropCounter>, pub compaction_applied: DeleteOnDropCounter>, pub cmd_succeeded: DeleteOnDropCounter>, - pub pubsub_push_diff_applied: DeleteOnDropCounter>, - pub pubsub_push_diff_not_applied_stale: DeleteOnDropCounter>, - pub pubsub_push_diff_not_applied_out_of_order: DeleteOnDropCounter>, - pub stale_version: DeleteOnDropGauge>, pub blob_gets: DeleteOnDropCounter>, pub blob_sets: DeleteOnDropCounter>, - pub live_writers: DeleteOnDropGauge>, pub unconsolidated_snapshot: DeleteOnDropCounter>, - pub backpressure_emitted_bytes: Arc>>, - pub backpressure_last_backpressured_bytes: Arc>>, - pub backpressure_retired_bytes: Arc>>, - pub rewrite_part_count: DeleteOnDropGauge>, pub inline_part_count: DeleteOnDropGauge>, - pub inline_part_bytes: DeleteOnDropGauge>, pub compact_batches: DeleteOnDropGauge>, pub compacting_batches: DeleteOnDropGauge>, pub noncompact_batches: DeleteOnDropGauge>, - pub schema_registry_version_count: DeleteOnDropGauge>, - pub inline_backpressure_count: DeleteOnDropCounter>, + // Not a registered per-shard series: this feeds the process-level + // `mz_persist_stale_shard_count` computed gauge, which sums it across shards + // instead of exporting one series per shard. + pub stale: AtomicBool, } impl ShardMetrics { @@ -1638,12 +1586,6 @@ impl ShardMetrics { ShardMetrics { shard_id: *shard_id, name: name.to_string(), - since: shards_metrics - .since - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), - upper: shards_metrics - .upper - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), latest_rollup_size: shards_metrics .encoded_rollup_size .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), @@ -1660,7 +1602,6 @@ impl ShardMetrics { .batch_part_count .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), batch_part_version_count: shards_metrics.batch_part_version_count.clone(), - batch_part_version_bytes: shards_metrics.batch_part_version_bytes.clone(), batch_part_version_map: Mutex::new(BTreeMap::new()), update_count: shards_metrics .update_count @@ -1707,54 +1648,18 @@ impl ShardMetrics { usage_leaked_bytes: shards_metrics .usage_leaked_bytes .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), - pubsub_push_diff_applied: shards_metrics - .pubsub_push_diff_applied - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), - pubsub_push_diff_not_applied_stale: shards_metrics - .pubsub_push_diff_not_applied_stale - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), - pubsub_push_diff_not_applied_out_of_order: shards_metrics - .pubsub_push_diff_not_applied_out_of_order - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), - stale_version: shards_metrics - .stale_version - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), blob_gets: shards_metrics .blob_gets .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), blob_sets: shards_metrics .blob_sets .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), - live_writers: shards_metrics - .live_writers - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), unconsolidated_snapshot: shards_metrics .unconsolidated_snapshot .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), - backpressure_emitted_bytes: Arc::new( - shards_metrics - .backpressure_emitted_bytes - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), - ), - backpressure_last_backpressured_bytes: Arc::new( - shards_metrics - .backpressure_last_backpressured_bytes - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), - ), - backpressure_retired_bytes: Arc::new( - shards_metrics - .backpressure_retired_bytes - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), - ), - rewrite_part_count: shards_metrics - .rewrite_part_count - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), inline_part_count: shards_metrics .inline_part_count .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), - inline_part_bytes: shards_metrics - .inline_part_bytes - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), compact_batches: shards_metrics .compact_batches .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), @@ -1763,27 +1668,14 @@ impl ShardMetrics { .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), noncompact_batches: shards_metrics .noncompact_batches - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), - schema_registry_version_count: shards_metrics - .schema_registry_version_count - .get_delete_on_drop_metric(vec![shard.clone(), name.to_string()]), - inline_backpressure_count: shards_metrics - .inline_backpressure_count .get_delete_on_drop_metric(vec![shard, name.to_string()]), + stale: AtomicBool::new(false), } } - pub fn set_since(&self, since: &Antichain) { - self.since.set(encode_ts_metric(since)) - } - - pub fn set_upper(&self, upper: &Antichain) { - self.upper.set(encode_ts_metric(upper)) - } - pub(crate) fn set_batch_part_versions<'a>( &self, - batch_parts_by_version: impl Iterator, + batch_parts_by_version: impl Iterator, ) { let mut map = self .batch_part_version_map @@ -1797,12 +1689,11 @@ impl ShardMetrics { // map). First reset everything. for x in map.values() { x.batch_part_version_count.set(0); - x.batch_part_version_bytes.set(0); } // Then go through the iterator, creating new entries as necessary and // adding. - for (key, bytes) in batch_parts_by_version { + for key in batch_parts_by_version { if !map.contains_key(key) { map.insert( key.to_owned(), @@ -1814,19 +1705,11 @@ impl ShardMetrics { self.name.clone(), key.to_owned(), ]), - batch_part_version_bytes: self - .batch_part_version_bytes - .get_delete_on_drop_metric(vec![ - self.shard_id.to_string(), - self.name.clone(), - key.to_owned(), - ]), }, ); } let value = map.get(key).expect("inserted above"); value.batch_part_version_count.inc(); - value.batch_part_version_bytes.add(u64::cast_from(bytes)); } } } @@ -1834,7 +1717,6 @@ impl ShardMetrics { #[derive(Debug)] pub struct BatchPartVersionMetrics { pub batch_part_version_count: DeleteOnDropGauge>, - pub batch_part_version_bytes: DeleteOnDropGauge>, } /// Metrics recorded by audits of persist usage @@ -2047,6 +1929,43 @@ impl SinkWorkerMetrics { } } +/// Metrics for the `persist_source` backpressure operator, summed over every +/// instance of the operator in the process. Like [SinkMetrics], these belong +/// to a dataflow operator rather than the client, but the client owns the only +/// registry the operator can reach. +#[derive(Debug)] +pub struct BackpressureMetrics { + /// Bytes emitted by backpressure operators. + pub emitted_bytes: IntCounter, + /// Sum over live operator instances of the inflight bytes each one most + /// recently stalled on. Instances contribute deltas, so the sum stays exact + /// as operators start and stop. + pub last_backpressured_bytes: UIntGauge, + /// Bytes retired by processing downstream of backpressure operators. + pub retired_bytes: IntCounter, +} + +impl BackpressureMetrics { + fn new(registry: &MetricsRegistry) -> Self { + BackpressureMetrics { + emitted_bytes: registry.register(metric!( + name: "mz_persist_backpressure_emitted_bytes", + help: "bytes emitted by backpressure operators", + )), + last_backpressured_bytes: registry.register(metric!( + name: "mz_persist_backpressure_last_backpressured_bytes", + help: "sum over backpressure operators of the inflight bytes each last \ + stalled on; not comparable to retired bytes, but nonzero growth \ + indicates backpressure is happening", + )), + retired_bytes: registry.register(metric!( + name: "mz_persist_backpressure_retired_bytes", + help: "bytes retired by processing downstream of backpressure operators", + )), + } + } +} + /// A minimal set of metrics imported into honeycomb for alerting. #[derive(Debug)] pub struct AlertsMetrics { @@ -2195,6 +2114,10 @@ pub struct PubSubClientReceiverMetrics { pub(crate) state_pushed_diff_fast_path: IntCounter, pub(crate) state_pushed_diff_slow_path_succeeded: IntCounter, pub(crate) state_pushed_diff_slow_path_failed: IntCounter, + + pub(crate) diff_applied: IntCounter, + pub(crate) diff_not_applied_stale: IntCounter, + pub(crate) diff_not_applied_out_of_order: IntCounter, } impl PubSubClientReceiverMetrics { @@ -2226,6 +2149,18 @@ impl PubSubClientReceiverMetrics { name: "mz_persist_pubsub_client_receiver_state_push_diff_slow_path_failed", help: "count of unsuccessful slow-path state push_diff calls", )), + diff_applied: registry.register(metric!( + name: "mz_persist_pubsub_client_receiver_diff_applied", + help: "number of diffs received via pubsub that applied", + )), + diff_not_applied_stale: registry.register(metric!( + name: "mz_persist_pubsub_client_receiver_diff_not_applied_stale", + help: "number of diffs received via pubsub that did not apply due to staleness", + )), + diff_not_applied_out_of_order: registry.register(metric!( + name: "mz_persist_pubsub_client_receiver_diff_not_applied_out_of_order", + help: "number of diffs received via pubsub that did not apply due to out-of-order delivery", + )), } } } @@ -3255,3 +3190,39 @@ pub fn encode_ts_metric(ts: &Antichain) -> i64 { None => i64::MAX, } } + +#[cfg(test)] +mod tests { + use super::*; + + #[mz_ore::test] + fn shards_aggregate_metrics_one_pass() { + let registry = MetricsRegistry::new(); + let shards = ShardsMetrics::new(®istry); + let agg = ShardsAggregateMetrics::new(Arc::clone(&shards.shards)); + let a = shards.shard(&ShardId::new(), "a"); + let b = shards.shard(&ShardId::new(), "b"); + + // collect() recomputes both gauges from a single walk of the map. + agg.collect(); + assert_eq!(agg.count.get(), 2); + assert_eq!(agg.stale_count.get(), 0); + + a.stale.store(true, Ordering::Relaxed); + agg.collect(); + assert_eq!(agg.count.get(), 2); + assert_eq!(agg.stale_count.get(), 1); + + b.stale.store(true, Ordering::Relaxed); + agg.collect(); + assert_eq!(agg.stale_count.get(), 2); + + // Flipping back and dropping a handle both lower the counts: the gauge + // is recomputed, not ticked. + a.stale.store(false, Ordering::Relaxed); + drop(b); + agg.collect(); + assert_eq!(agg.count.get(), 1); + assert_eq!(agg.stale_count.get(), 0); + } +} diff --git a/src/persist-client/src/internal/state.rs b/src/persist-client/src/internal/state.rs index db3fb41b54f6e..5ae1d36a210ee 100644 --- a/src/persist-client/src/internal/state.rs +++ b/src/persist-client/src/internal/state.rs @@ -2486,12 +2486,8 @@ where let batch_size = x.encoded_size_bytes(); for x in x.parts.iter() { - if x.ts_rewrite().is_some() { - ret.rewrite_part_count += 1; - } if x.is_inline() { ret.inline_part_count += 1; - ret.inline_part_bytes += x.inline_bytes(); } } ret.largest_batch_bytes = std::cmp::max(ret.largest_batch_bytes, batch_size); @@ -2857,14 +2853,12 @@ impl Serialize for State { pub struct StateSizeMetrics { pub hollow_batch_count: usize, pub batch_part_count: usize, - pub rewrite_part_count: usize, pub num_updates: usize, pub largest_batch_bytes: usize, pub state_batches_bytes: usize, pub state_rollups_bytes: usize, pub state_rollup_count: usize, pub inline_part_count: usize, - pub inline_part_bytes: usize, } #[derive(Default)] diff --git a/src/persist-client/src/internal/state_versions.rs b/src/persist-client/src/internal/state_versions.rs index 242e832882638..0322d302b8277 100644 --- a/src/persist-client/src/internal/state_versions.rs +++ b/src/persist-client/src/internal/state_versions.rs @@ -14,6 +14,7 @@ use std::collections::BTreeSet; use std::fmt::Debug; use std::ops::ControlFlow::{Break, Continue}; use std::sync::Arc; +use std::sync::atomic::Ordering; use std::time::SystemTime; use bytes::Bytes; @@ -274,8 +275,6 @@ impl StateVersions { new_state ); - shard_metrics.set_since(new_state.since()); - shard_metrics.set_upper(new_state.upper()); shard_metrics.seqnos_since_last_rollup.set( new_state .seqno @@ -286,18 +285,12 @@ impl StateVersions { .spine_batch_count .set(u64::cast_from(new_state.spine_batch_count())); let size_metrics = new_state.size_metrics(); - shard_metrics - .schema_registry_version_count - .set(u64::cast_from(new_state.collections.schemas.len())); shard_metrics .hollow_batch_count .set(u64::cast_from(size_metrics.hollow_batch_count)); shard_metrics .batch_part_count .set(u64::cast_from(size_metrics.batch_part_count)); - shard_metrics - .rewrite_part_count - .set(u64::cast_from(size_metrics.rewrite_part_count)); shard_metrics .update_count .set(u64::cast_from(size_metrics.num_updates)); @@ -319,30 +312,17 @@ impl StateVersions { shard_metrics .encoded_diff_size .inc_by(u64::cast_from(payload_len)); - shard_metrics - .live_writers - .set(u64::cast_from(new_state.collections.writers.len())); - shard_metrics - .rewrite_part_count - .set(u64::cast_from(size_metrics.rewrite_part_count)); shard_metrics .inline_part_count .set(u64::cast_from(size_metrics.inline_part_count)); - shard_metrics - .inline_part_bytes - .set(u64::cast_from(size_metrics.inline_part_bytes)); - shard_metrics.stale_version.set( - if new_state + shard_metrics.stale.store( + new_state .state .collections .version .cmp_precedence(&self.cfg.build_version) - .is_lt() - { - 1 - } else { - 0 - }, + .is_lt(), + Ordering::Relaxed, ); let spine_metrics = new_state.collections.trace.spine_metrics(); @@ -371,8 +351,8 @@ impl StateVersions { // Carefully avoid any String allocs by splitting. let (writer_key, _) = key.0.split_once('/')?; match &writer_key[..1] { - "w" => Some(("old", part.encoded_size_bytes())), - "n" => Some((&writer_key[1..], part.encoded_size_bytes())), + "w" => Some("old"), + "n" => Some(&writer_key[1..]), _ => None, } }); diff --git a/src/persist-client/src/lib.rs b/src/persist-client/src/lib.rs index 752b4dfbb6ecb..8a19c91b792cb 100644 --- a/src/persist-client/src/lib.rs +++ b/src/persist-client/src/lib.rs @@ -78,7 +78,7 @@ pub mod iter; pub mod metrics { //! Utilities related to metrics. pub use crate::internal::metrics::{ - Metrics, SinkMetrics, SinkWorkerMetrics, UpdateDelta, encode_ts_metric, + BackpressureMetrics, Metrics, SinkMetrics, SinkWorkerMetrics, UpdateDelta, encode_ts_metric, }; } pub mod operators { diff --git a/src/storage-operators/src/metrics.rs b/src/storage-operators/src/metrics.rs index d0dcb5716c955..0fa1f5a424444 100644 --- a/src/storage-operators/src/metrics.rs +++ b/src/storage-operators/src/metrics.rs @@ -9,19 +9,97 @@ //! Metrics for storage dataflow operators -use std::sync::Arc; - -use mz_ore::metrics::{DeleteOnDropCounter, DeleteOnDropGauge}; -use prometheus::core::AtomicU64; - -/// Metrics used by the `backpressure` operator. -#[derive(Debug, Clone)] -pub struct BackpressureMetrics { - /// A counter with the number of emitted bytes. - pub emitted_bytes: Arc>>, - /// The last count of bytes we are waiting to be retired in the operator. This cannot - /// be directly compared to `retired_bytes`, but CAN indicate that backpressure is happening. - pub last_backpressured_bytes: Arc>>, - /// A counter with the number of bytes retired by downstream processing. - pub retired_bytes: Arc>>, +use std::sync::atomic::{AtomicU64, Ordering}; + +use mz_ore::metrics::{IntCounter, UIntGauge}; + +/// Metric handles for one instance of the `backpressure` operator. +/// +/// The series behind the handles belong to whoever constructed them and stay +/// registered for as long as that owner keeps them: the persist client's +/// process-level backpressure series for `persist_source`, or a source's own +/// per-worker series for upsert. Several operator instances may share one +/// gauge, which is why the gauge handle is a [GaugeContribution]. +#[derive(Debug)] +pub struct BackpressureOperatorMetrics { + /// Bytes this operator emitted. + pub emitted_bytes: IntCounter, + /// This operator's share of the gauge: the inflight bytes it most recently + /// stalled on. + pub last_backpressured_bytes: GaugeContribution, + /// Bytes retired by processing downstream of this operator. + pub retired_bytes: IntCounter, +} + +impl BackpressureOperatorMetrics { + pub fn new( + emitted_bytes: IntCounter, + last_backpressured_bytes: UIntGauge, + retired_bytes: IntCounter, + ) -> Self { + BackpressureOperatorMetrics { + emitted_bytes, + last_backpressured_bytes: GaugeContribution::new(last_backpressured_bytes), + retired_bytes, + } + } +} + +/// One contributor's share of a gauge that reads as the sum over all +/// contributors. Dropping the contribution withdraws it. +#[derive(Debug)] +pub struct GaugeContribution { + gauge: UIntGauge, + contributed: AtomicU64, +} + +impl GaugeContribution { + pub fn new(gauge: UIntGauge) -> Self { + GaugeContribution { + gauge, + contributed: AtomicU64::new(0), + } + } + + /// Replaces this contributor's share with `value`. + pub fn set(&self, value: u64) { + let previous = self.contributed.swap(value, Ordering::AcqRel); + // Add before subtracting so the shared total never dips below the sum + // of the other contributions. + self.gauge.add(value); + self.gauge.sub(previous); + } +} + +impl Drop for GaugeContribution { + fn drop(&mut self) { + self.gauge.sub(self.contributed.load(Ordering::Acquire)); + } +} + +#[cfg(test)] +mod tests { + use mz_ore::metrics::UIntGauge; + + use super::GaugeContribution; + + #[mz_ore::test] + fn gauge_contribution_sums_live_shares_and_withdraws_on_drop() { + let gauge = UIntGauge::new("gauge", "help").expect("valid metric"); + let a = GaugeContribution::new(gauge.clone()); + let b = GaugeContribution::new(gauge.clone()); + + a.set(5); + b.set(3); + assert_eq!(gauge.get(), 8); + + a.set(2); + assert_eq!(gauge.get(), 5); + + drop(a); + assert_eq!(gauge.get(), 3); + + b.set(0); + assert_eq!(gauge.get(), 0); + } } diff --git a/src/storage-operators/src/persist_source.rs b/src/storage-operators/src/persist_source.rs index ab10d44ce5e2b..12c97ec06304c 100644 --- a/src/storage-operators/src/persist_source.rs +++ b/src/storage-operators/src/persist_source.rs @@ -64,7 +64,7 @@ use timely::scheduling::Activator; use tokio::sync::mpsc::UnboundedSender; use tracing::{error, trace}; -use crate::metrics::BackpressureMetrics; +use crate::metrics::BackpressureOperatorMetrics; /// This opaque token represents progress within a timestamp, allowing finer-grained frontier /// progress than would otherwise be possible. @@ -188,21 +188,18 @@ pub fn persist_source<'scope, E>( where E: timely::ExchangeData + Ord + Clone + Debug + From + From, { - let shard_metrics = persist_clients.shard_metrics(&metadata.data_shard, &source_id.to_string()); - let mut tokens = vec![]; let outer = scope.clone(); let stream = scope.scoped(&format!("granular_backpressure({})", source_id), |scope| { let (flow_control, flow_control_probe) = match max_inflight_bytes { Some(max_inflight_bytes) => { - let backpressure_metrics = BackpressureMetrics { - emitted_bytes: Arc::clone(&shard_metrics.backpressure_emitted_bytes), - last_backpressured_bytes: Arc::clone( - &shard_metrics.backpressure_last_backpressured_bytes, - ), - retired_bytes: Arc::clone(&shard_metrics.backpressure_retired_bytes), - }; + let series = &persist_clients.metrics().backpressure; + let backpressure_metrics = BackpressureOperatorMetrics::new( + series.emitted_bytes.clone(), + series.last_backpressured_bytes.clone(), + series.retired_bytes.clone(), + ); let probe = mz_timely_util::probe::Handle::default(); let progress_stream = mz_timely_util::probe::source( @@ -773,7 +770,7 @@ pub struct FlowControl<'scope, T: timely::progress::Timestamp> { pub summary: T::Summary, /// Optional metrics for the `backpressure` operator to keep up-to-date. - pub metrics: Option, + pub metrics: Option, } /// Apply flow control to the `data` input, based on the given `FlowControl`. diff --git a/src/storage/src/metrics.rs b/src/storage/src/metrics.rs index 285aefc85e943..54f9c9ba4b9e2 100644 --- a/src/storage/src/metrics.rs +++ b/src/storage/src/metrics.rs @@ -28,13 +28,10 @@ //! needing to interact with metrics _definitions_ into the code that actually bumps those //! metrics. -use std::sync::Arc; - use mz_ore::metrics::MetricsRegistry; use mz_repr::GlobalId; use crate::statistics::{SinkStatisticsMetricDefs, SourceStatisticsMetricDefs}; -use mz_storage_operators::metrics::BackpressureMetrics; pub mod decode; pub mod sink; @@ -77,37 +74,22 @@ impl StorageMetrics { } } - /// Get a `BackpressureMetrics` for the given id and worker id. + /// Get the backpressure series for the given id and worker id. pub(crate) fn get_backpressure_metrics( &self, id: GlobalId, index: usize, - ) -> BackpressureMetrics { - BackpressureMetrics { - emitted_bytes: Arc::new( - self.upsert_backpressure_defs - .emitted_bytes - .get_delete_on_drop_metric(vec![id.to_string(), index.to_string()]), - ), - last_backpressured_bytes: Arc::new( - self.upsert_backpressure_defs - .last_backpressured_bytes - .get_delete_on_drop_metric(vec![id.to_string(), index.to_string()]), - ), - retired_bytes: Arc::new( - self.upsert_backpressure_defs - .retired_bytes - .get_delete_on_drop_metric(vec![id.to_string(), index.to_string()]), - ), - } + ) -> upsert::UpsertBackpressureMetrics { + upsert::UpsertBackpressureMetrics::new(&self.upsert_backpressure_defs, id, index) } - /// Get an `UpsertMetrics` for the given id and worker id (and optional `BackpressureMetrics`). + /// Get an `UpsertMetrics` for the given id and worker id (and optional + /// `UpsertBackpressureMetrics`). pub(crate) fn get_upsert_metrics( &self, id: GlobalId, worker_id: usize, - backpressure_metrics: Option, + backpressure_metrics: Option, ) -> upsert::UpsertMetrics { upsert::UpsertMetrics::new(&self.upsert_defs, id, worker_id, backpressure_metrics) } diff --git a/src/storage/src/metrics/upsert.rs b/src/storage/src/metrics/upsert.rs index dc99c5b547d45..34f8ea43cf0b3 100644 --- a/src/storage/src/metrics/upsert.rs +++ b/src/storage/src/metrics/upsert.rs @@ -20,7 +20,7 @@ use mz_ore::metrics::{ use mz_ore::stats::histogram_seconds_buckets; use mz_repr::GlobalId; use mz_rocksdb::RocksDBInstanceMetrics; -use mz_storage_operators::metrics::BackpressureMetrics; +use mz_storage_operators::metrics::BackpressureOperatorMetrics; use prometheus::core::{AtomicF64, AtomicU64}; /// Metric definitions for the `upsert` operator. @@ -374,9 +374,42 @@ pub struct UpsertMetrics { pub(crate) shared: Arc, pub(crate) rocksdb_shared: Arc, pub(crate) rocksdb_instance_metrics: Arc, - // `UpsertMetrics` keeps a reference (through `Arc`'s) to backpressure metrics, so that - // they are not dropped when the `persist_source` operator is dropped. - _backpressure_metrics: Option, + // Owning the backpressure series here keeps them registered for the life + // of the upsert operator, past the `persist_source` operator that reports + // into them. + _backpressure_metrics: Option, +} + +/// The per-source, per-worker backpressure series for one upsert operator. +/// +/// Owning this keeps the series registered. [Self::operator_metrics] hands the +/// `backpressure` operator plain handles to them. +#[derive(Debug)] +pub(crate) struct UpsertBackpressureMetrics { + emitted_bytes: DeleteOnDropCounter>, + last_backpressured_bytes: DeleteOnDropGauge>, + retired_bytes: DeleteOnDropCounter>, +} + +impl UpsertBackpressureMetrics { + pub(crate) fn new(defs: &UpsertBackpressureMetricDefs, id: GlobalId, worker_id: usize) -> Self { + let labels = vec![id.to_string(), worker_id.to_string()]; + UpsertBackpressureMetrics { + emitted_bytes: defs.emitted_bytes.get_delete_on_drop_metric(labels.clone()), + last_backpressured_bytes: defs + .last_backpressured_bytes + .get_delete_on_drop_metric(labels.clone()), + retired_bytes: defs.retired_bytes.get_delete_on_drop_metric(labels), + } + } + + pub(crate) fn operator_metrics(&self) -> BackpressureOperatorMetrics { + BackpressureOperatorMetrics::new( + (*self.emitted_bytes).clone(), + (*self.last_backpressured_bytes).clone(), + (*self.retired_bytes).clone(), + ) + } } impl UpsertMetrics { @@ -385,7 +418,7 @@ impl UpsertMetrics { defs: &UpsertMetricDefs, source_id: GlobalId, worker_id: usize, - backpressure_metrics: Option, + backpressure_metrics: Option, ) -> Self { let source_id_s = source_id.to_string(); let worker_id = worker_id.to_string(); diff --git a/src/storage/src/render/sources.rs b/src/storage/src/render/sources.rs index e64b4a2fa5b6a..a4175a3b1f6b9 100644 --- a/src/storage/src/render/sources.rs +++ b/src/storage/src/render/sources.rs @@ -282,7 +282,9 @@ where progress_stream: feedback_data, max_inflight_bytes: storage_dataflow_max_inflight_bytes, summary: (Default::default(), Subtime::least_summary()), - metrics: backpressure_metrics.clone(), + metrics: backpressure_metrics + .as_ref() + .map(|m| m.operator_metrics()), }), backpressure_metrics, ) diff --git a/src/storage/src/upsert.rs b/src/storage/src/upsert.rs index 6bdb4bcd6acd4..a4b99ebd70ae8 100644 --- a/src/storage/src/upsert.rs +++ b/src/storage/src/upsert.rs @@ -25,7 +25,6 @@ use mz_ore::error::ErrorExt; use mz_repr::{Datum, DatumVec, Diff, GlobalId, Row}; use mz_rocksdb::ValueIterator; use mz_sql_server_util::cdc::Lsn; -use mz_storage_operators::metrics::BackpressureMetrics; use mz_storage_types::configuration::StorageConfiguration; use mz_storage_types::dyncfgs; use mz_storage_types::errors::{DataflowError, EnvelopeError, UpsertError}; @@ -47,7 +46,7 @@ use timely::progress::timestamp::Refines; use timely::progress::{Antichain, Timestamp}; use crate::healthcheck::HealthStatusUpdate; -use crate::metrics::upsert::UpsertMetrics; +use crate::metrics::upsert::{UpsertBackpressureMetrics, UpsertMetrics}; use crate::storage_state::StorageInstanceContext; use crate::{upsert_continual_feedback, upsert_continual_feedback_v2}; use types::{ @@ -508,7 +507,7 @@ pub(crate) fn upsert<'scope, T, FromTime>( instance_context: &StorageInstanceContext, storage_configuration: &StorageConfiguration, dataflow_paramters: &crate::internal_control::DataflowParameters, - backpressure_metrics: Option, + backpressure_metrics: Option, ) -> ( VecCollection<'scope, T, Result, Diff>, StreamVec<'scope, T, (Option, HealthStatusUpdate)>, @@ -641,7 +640,7 @@ pub(crate) fn upsert_v2<'scope, T, FromTime>( previous: VecCollection<'scope, T, Result, Diff>, previous_token: Option>, source_config: crate::source::SourceExportCreationConfig, - backpressure_metrics: Option, + backpressure_metrics: Option, stash_flavor: upsert_continual_feedback_v2::UpsertStashFlavor, ) -> ( VecCollection<'scope, T, Result, Diff>,