diff --git a/Cargo.lock b/Cargo.lock index ce9398cd42d20..a7a67317f2e3b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7487,10 +7487,12 @@ dependencies = [ name = "mz-metrics" version = "0.0.0" dependencies = [ + "anyhow", "lgalloc", "libc", "mz-dyncfg", "mz-ore", + "nix 0.31.3", "paste", "prometheus", "thiserror 2.0.18", diff --git a/doc/user/content/manage/monitor/replica-resource-usage.md b/doc/user/content/manage/monitor/replica-resource-usage.md new file mode 100644 index 0000000000000..328fcd8a626ea --- /dev/null +++ b/doc/user/content/manage/monitor/replica-resource-usage.md @@ -0,0 +1,129 @@ +--- +title: "Replica resource usage" +description: "The per-process memory, swap and disk observations a cluster replica reports about itself, and how to interpret them." +menu: + main: + parent: "monitor" + identifier: "monitor-replica-resource-usage" + weight: 18 +--- + +Every process of a cluster replica reports its own resource usage through +[`mz_introspection.mz_cluster_replica_resource_usage`](/reference/system-catalog/mz_introspection/#mz_cluster_replica_resource_usage). +Each row is one measurement, taken from one source, reported as that source +gave it. Sources measure overlapping but distinct quantities, and the +differences between them are informative, so no row is a combination of two +others. Deciding which number is "the" memory usage of a replica, or how close +it is to its limit, is left to queries over the relation. + +A metric whose name ends in `peak` is a high-water mark since the process +started, and the rest are instantaneous. Peaks the operating system maintains +itself are exact, and are unaffected by how often the replica reads them. Peaks +folded from samples are marked as such below and can miss a spike shorter than +the sampling interval, which makes them lower bounds. An observation the +replica could not read is absent rather than zero, so which metrics appear +depends on the platform and the kernel version. + +## Sources + +| Source | Reads | Measures | +|---------------|-----------------------------------------|-------------------------------------------------------------------------------------------| +| `cgroup` | the process's cgroup v2 interface files | the whole container, and the accounting that limit enforcement and the OOM killer act on | +| `proc_status` | `/proc/self/status` | this process only, with resident memory broken down by backing | +| `rusage` | `getrusage(RUSAGE_SELF)` | this process only | +| `statvfs` | the replica's scratch filesystem | the filesystem as a whole, absent where disk is provided as swap | + +## Metrics + +| Source | Metric | Meaning | +|---------------|-------------------|--------------------------------------------------------------------------------------------------------------------------------| +| `cgroup` | `memory_current` | Memory charged to the cgroup: anonymous, page cache, kernel and socket memory. | +| `cgroup` | `memory_peak` | High-water mark of `memory_current`, maintained by the kernel. | +| `cgroup` | `memory_max` | The cgroup's memory limit. | +| `cgroup` | `swap_current` | Swap charged to the cgroup, including pages already read back whose swap slot is still allocated. | +| `cgroup` | `swap_peak` | High-water mark of `swap_current`, maintained by the kernel. | +| `cgroup` | `swap_max` | The cgroup's swap limit. | +| `cgroup` | `anon` | The part of `memory_current` backed by no file. | +| `cgroup` | `file` | Page cache charged to the cgroup. | +| `cgroup` | `shmem` | Shared memory and tmpfs pages. | +| `cgroup` | `swapcached` | Pages resident in memory whose swap slot is still allocated. Counted in both `anon` and `swap_current`. | +| `cgroup` | `kernel` | Kernel memory charged to the cgroup. | +| `cgroup` | `slab` | Kernel slab allocations, part of `kernel`. | +| `cgroup` | `sock` | Socket buffer memory. | +| `cgroup` | `events_max` | Times an allocation hit `memory_max`. See the caveat below before using this as a limit-hit signal. | +| `cgroup` | `events_oom_kill` | Processes in the cgroup killed by the OOM killer. | +| `proc_status` | `vm_rss` | Resident set size of this process, the sum of `rss_anon`, `rss_file` and `rss_shmem`. | +| `proc_status` | `rss_anon` | Resident memory backed by no file. The replica's own memory. | +| `proc_status` | `rss_file` | Resident file-backed memory, largely this binary's text. Shared between replicas and charged to whichever cgroup first faulted it in. | +| `proc_status` | `rss_shmem` | Resident shared memory. | +| `proc_status` | `vm_swap` | This process's pages currently in swap. Excludes swap-cached pages, so it reads below `cgroup` `swap_current`. | +| `proc_status` | `vm_swap_peak` | Maximum `vm_swap` over samples, so a lower bound on the true peak. | +| `proc_status` | `heap` | `vm_rss` plus `vm_swap`, the quantity a replica is limited on. | +| `proc_status` | `heap_peak` | Maximum `heap` over samples, so a lower bound on the true peak. | +| `rusage` | `max_rss` | Peak resident set size. Maintained by the kernel, but refreshed only at internal checkpoints, so it can read below a concurrent `vm_rss`. | +| `statvfs` | `fs_used` | Used bytes of the filesystem, which on a shared filesystem counts writes this replica never made. | +| `statvfs` | `fs_used_peak` | Maximum `fs_used` over samples, so a lower bound on the true peak. | + +## Interpreting + +Values from different sources are not interchangeable, and adding them together +generally produces a number that means nothing. In particular: + +* **How close is this replica to its memory limit?** Compare `cgroup` + `memory_current` against `memory_max`. **Did it ever reach it?** Compare + `memory_peak` against `memory_max`. A `memory_peak` at the limit means the + replica ran out of RAM and spilled to swap, even if the current reading is + comfortable. +* **How close is it to its heap limit?** Compare `proc_status` `heap` against + [`mz_internal.mz_cluster_replica_metrics`](/reference/system-catalog/mz_internal/#mz_cluster_replica_metrics)'s + `heap_limit`. `heap_peak` bounds the high-water mark from below, and no source + bounds it from above: the `cgroup` peaks describe a smaller quantity, and + `max_rss` lags. +* **Do not use `events_max` as a limit-hit signal.** Where swap is configured it + stays at zero even for a replica pinned at its ceiling, because reclaim + succeeds by swapping instead of failing. `events_oom_kill` does report kills. +* **How much memory does this replica itself account for?** Use `proc_status` + `rss_anon`. Do not use `vm_rss`: it includes `rss_file`, which is charged to + another cgroup and so runs a roughly constant amount above the replica's own + charge. +* **Do not add `memory_current` and `swap_current`.** A page read back from swap + is counted in both, and `swapcached` reports how much is in that state. +* **Where disk is provided as swap**, disk usage appears as `swap_current` and + there are no `statvfs` rows at all. +* **These readings live and die with the replica process.** A restart resets + every peak. For history that survives restarts, see + [`mz_internal.mz_cluster_replica_metrics_history`](/reference/system-catalog/mz_internal/#mz_cluster_replica_metrics_history). + +## Example + +Introspection relations are replica-local: a query reads the replica that +serves it, so pin both the cluster and the replica. This reports how close each +process came to a memory-limiter kill, comparing the quantity the limiter +enforces against the limit it enforces: + +```mzsql +SET cluster = ; +SET cluster_replica = ; + +SELECT + u.process_id, + round((max(u.value) FILTER (WHERE u.metric = 'heap'))::numeric / 1073741824, 2) AS heap_gib, + round((max(u.value) FILTER (WHERE u.metric = 'heap_peak'))::numeric / 1073741824, 2) AS heap_peak_gib, + round(m.heap_limit::numeric / 1073741824, 2) AS limit_gib, + round(100 * (max(u.value) FILTER (WHERE u.metric = 'heap_peak'))::numeric / m.heap_limit, 1) AS peak_pct +FROM mz_introspection.mz_cluster_replica_resource_usage u +JOIN mz_cluster_replicas r + ON r.name = current_setting('cluster_replica') + AND r.cluster_id = (SELECT id FROM mz_clusters WHERE name = current_setting('cluster')) +JOIN mz_internal.mz_cluster_replica_metrics m + ON m.replica_id = r.id AND m.process_id = u.process_id +WHERE u.source = 'proc_status' +GROUP BY u.process_id, m.heap_limit +ORDER BY u.process_id; +``` + +`peak_pct` is a lower bound, because `heap_peak` is: a spike shorter than the +sampling interval can slip through it. No source provides a matching upper +bound. The query stays within `proc_status` deliberately. The `cgroup` metrics +measure a different quantity, and `memory_current` plus `swap_current` +double-counts every swap-cached page. diff --git a/doc/user/content/reference/system-catalog/mz_internal.md b/doc/user/content/reference/system-catalog/mz_internal.md index 8d6f3347be034..39fbf312341a4 100644 --- a/doc/user/content/reference/system-catalog/mz_internal.md +++ b/doc/user/content/reference/system-catalog/mz_internal.md @@ -204,6 +204,13 @@ The `mz_cluster_replica_metrics` view gives the last known CPU and RAM utilizati for all processes of all extant cluster replicas. At this time, we do not make any guarantees about the exactness or freshness of these numbers. +They are sampled roughly once a minute, so a spike shorter than the sampling interval is not +visible here at all. For a view of a single replica sampled every few seconds, including high-water +marks that survive a spike the sampling missed, see [Replica resource +usage](/manage/monitor/replica-resource-usage/). + +Where a replica's disk is provided as swap rather than as a filesystem, `disk_bytes` reports swap +usage. | Field | Type | Meaning @@ -223,6 +230,11 @@ The `mz_cluster_replica_metrics_history` table records resource utilization metr for all processes of all extant cluster replicas. At this time, we do not make any guarantees about the exactness or freshness of these numbers. +They are sampled roughly once a minute, so a spike shorter than the sampling interval leaves no +trace. Unlike +[`mz_introspection.mz_cluster_replica_resource_usage`](/reference/system-catalog/mz_introspection/#mz_cluster_replica_resource_usage), +which is sampled every few seconds but is replica-local and resets when a replica restarts, this +history is retained across restarts. | Field | Type | Meaning diff --git a/doc/user/content/reference/system-catalog/mz_introspection.md b/doc/user/content/reference/system-catalog/mz_introspection.md index f238ce5e8240b..bf048dddc15b2 100644 --- a/doc/user/content/reference/system-catalog/mz_introspection.md +++ b/doc/user/content/reference/system-catalog/mz_introspection.md @@ -170,6 +170,27 @@ Summaries are flattened into separate quantile, sum, and count rows. | `value` | [`double precision`] | The numeric value of the metric. | | `help` | [`text`] | The help string describing the metric. | +## `mz_cluster_replica_resource_usage` + +The `mz_cluster_replica_resource_usage` source reports the resource usage of each process of a +cluster replica, as one row per measurement source and metric. Each row is what that source +reported, without interpretation: sources measure overlapping but distinct quantities, so combining +them into a single figure for memory usage, or for how close a replica is to its limit, is left to +queries over this relation. + +The replica samples its sources every few seconds, and reports a high-water mark alongside the +instantaneous value where one is available. For the sources and metrics that appear here, how to +interpret them, and an example query, see [Replica resource +usage](/manage/monitor/replica-resource-usage/). + + +| Field | Type | Meaning | +|--------------|-----------|----------------------------------------------------------------------| +| `process_id` | [`uint8`] | The ID of the process within the replica. | +| `source` | [`text`] | The measurement source, for example `cgroup` or `rusage`. | +| `metric` | [`text`] | What the source measured, for example `memory_current`. | +| `value` | [`uint8`] | The reported value, in bytes for a size and as a count otherwise. | + ## `mz_dataflows` The `mz_dataflows` view describes the [dataflows][dataflow] in the system. diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index 034f404364004..f8af94bdea893 100644 --- a/doc/user/data/metrics.yml +++ b/doc/user/data/metrics.yml @@ -1402,6 +1402,13 @@ metrics: help: user CPU time used source: src/metrics/src/rusage.rs visibility: internal +- name: mz_metrics_resource_usage + help: Resource usage observations, by source and metric. + labels: + - metric + - source + source: src/metrics/src/usage.rs + visibility: internal - name: mz_metrics_update_duration_bucket help: The time it took to update lgalloc stats labels: diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 0244c6f41f21d..90c20566ff0b3 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -800,6 +800,7 @@ def get_default_system_parameters( "mz_metrics_lgalloc_map_refresh_interval", "mz_metrics_lgalloc_refresh_interval", "mz_metrics_rusage_refresh_interval", + "mz_metrics_usage_refresh_interval", "compute_peek_response_stash_batch_max_runs", "compute_peek_response_stash_read_batch_size_bytes", "compute_peek_response_stash_read_memory_budget_bytes", diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 405cce9e78092..54d28749dc190 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -3388,6 +3388,7 @@ def __init__( "mz_metrics_lgalloc_map_refresh_interval", "mz_metrics_lgalloc_refresh_interval", "mz_metrics_rusage_refresh_interval", + "mz_metrics_usage_refresh_interval", "compute_peek_stash_num_batches", "compute_peek_stash_batch_size", "compute_peek_response_stash_batch_max_runs", diff --git a/src/adapter/src/catalog/open/builtin_schema_migration.rs b/src/adapter/src/catalog/open/builtin_schema_migration.rs index 394a479d54202..a96caf5281975 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration.rs @@ -424,6 +424,23 @@ static MIGRATIONS: LazyLock> = LazyLock::new(|| { MZ_CATALOG_SCHEMA, "mz_views", ), + // Required because we added the `mz_cluster_replica_resource_usage` builtin log. + // make_mz_indexes and make_mz_sources inline the builtin-log set as + // VALUES, so adding one changes both MVs' SQL fingerprints. See the NOTE + // above: this version must stay at the workspace's current dev version + // until the change ships. + MigrationStep::replacement( + "26.40.0-dev.0", + CatalogItemType::MaterializedView, + MZ_CATALOG_SCHEMA, + "mz_indexes", + ), + MigrationStep::replacement( + "26.40.0-dev.0", + CatalogItemType::MaterializedView, + MZ_CATALOG_SCHEMA, + "mz_sources", + ), ] }); diff --git a/src/catalog/src/builtin.rs b/src/catalog/src/builtin.rs index 0a4c6d914c209..a3dbda548f476 100644 --- a/src/catalog/src/builtin.rs +++ b/src/catalog/src/builtin.rs @@ -1108,6 +1108,7 @@ pub static BUILTINS_STATIC: LazyLock>> = LazyLock::ne Builtin::Log(&MZ_COMPUTE_EXPORTS_PER_WORKER), Builtin::Log(&MZ_COMPUTE_DATAFLOW_GLOBAL_IDS_PER_WORKER), Builtin::Log(&MZ_CLUSTER_PROMETHEUS_METRICS), + Builtin::Log(&MZ_CLUSTER_REPLICA_RESOURCE_USAGE), Builtin::Log(&MZ_MESSAGE_COUNTS_RECEIVED_RAW), Builtin::Log(&MZ_MESSAGE_COUNTS_SENT_RAW), Builtin::Log(&MZ_MESSAGE_BATCH_COUNTS_RECEIVED_RAW), diff --git a/src/catalog/src/builtin/mz_introspection.rs b/src/catalog/src/builtin/mz_introspection.rs index d5ad93b478809..2e3bdf9764d02 100644 --- a/src/catalog/src/builtin/mz_introspection.rs +++ b/src/catalog/src/builtin/mz_introspection.rs @@ -248,6 +248,21 @@ pub static MZ_CLUSTER_PROMETHEUS_METRICS: LazyLock = LazyLock::new(| }), }); +pub static MZ_CLUSTER_REPLICA_RESOURCE_USAGE: LazyLock = LazyLock::new(|| BuiltinLog { + name: "mz_cluster_replica_resource_usage", + schema: MZ_INTROSPECTION_SCHEMA, + oid: oid::LOG_MZ_CLUSTER_REPLICA_RESOURCE_USAGE_OID, + variant: LogVariant::Compute(ComputeLog::ResourceUsage), + access: vec![PUBLIC_SELECT], + ontology: Some(Ontology { + entity_name: "cluster_replica_resource_usage", + description: "Resource usage of each process of the cluster replica, as reported by each \ + measurement source.", + links: &const { [] }, + column_semantic_types: &[], + }), +}); + pub static MZ_COMPUTE_FRONTIERS_PER_WORKER: LazyLock = LazyLock::new(|| BuiltinLog { name: "mz_compute_frontiers_per_worker", schema: MZ_INTROSPECTION_SCHEMA, diff --git a/src/catalog/src/durable/transaction.rs b/src/catalog/src/durable/transaction.rs index c27c43fadcc93..a83d5dd6def40 100644 --- a/src/catalog/src/durable/transaction.rs +++ b/src/catalog/src/durable/transaction.rs @@ -1021,6 +1021,7 @@ impl<'a> Transaction<'a> { LogVariant::Compute(ComputeLog::DataflowGlobal) => 31, LogVariant::Compute(ComputeLog::OperatorHydrationStatus) => 32, LogVariant::Compute(ComputeLog::PrometheusMetrics) => 33, + LogVariant::Compute(ComputeLog::ResourceUsage) => 34, }; let mut id: u64 = u64::from(cluster_variant) << 56; diff --git a/src/clusterd/src/lib.rs b/src/clusterd/src/lib.rs index 4d202a6fcaabd..87e9d932c0b29 100644 --- a/src/clusterd/src/lib.rs +++ b/src/clusterd/src/lib.rs @@ -268,7 +268,12 @@ async fn run(args: Args) -> Result<(), anyhow::Error> { emit_boot_diagnostics!(&BUILD_INFO); mz_alloc::register_metrics_into(&metrics_registry).await; - mz_metrics::register_metrics_into(&metrics_registry, mz_dyncfgs::all_dyncfgs()).await; + mz_metrics::register_metrics_into( + &metrics_registry, + mz_dyncfgs::all_dyncfgs(), + args.scratch_directory.clone(), + ) + .await; if let Some(heap_limit) = args.heap_limit { mz_compute::memory_limiter::start_limiter(heap_limit, &metrics_registry); diff --git a/src/clusterd/src/usage_metrics.rs b/src/clusterd/src/usage_metrics.rs index fbd648c59d49c..08f407e5613d4 100644 --- a/src/clusterd/src/usage_metrics.rs +++ b/src/clusterd/src/usage_metrics.rs @@ -15,7 +15,7 @@ use std::path::PathBuf; use serde::Serialize; -use tracing::{debug, error}; +use tracing::error; /// A system usage metrics collector. pub(crate) struct Collector { @@ -38,26 +38,14 @@ impl Collector { } fn collect_disk_usage(&self) -> Option { - let Some(root) = &self.disk_root else { - return None; - }; - - let stat = match nix::sys::statvfs::statvfs(root) { - Ok(stat) => stat, + let root = self.disk_root.as_deref()?; + match mz_metrics::usage::disk_usage(root) { + Ok(bytes) => Some(bytes), Err(err) => { error!("statvfs error: {err}"); - return None; + None } - }; - - // `fsblkcnt_t` is a `u32` on macOS but a `u64` on Linux. - #[allow(clippy::useless_conversion)] - let used_blocks = u64::from(stat.blocks() - stat.blocks_available()); - let used_bytes = used_blocks * stat.fragment_size(); - - debug!("disk usage: {used_bytes}"); - - Some(used_bytes) + } } } @@ -77,6 +65,7 @@ mod linux { use anyhow::{anyhow, bail}; use mz_compute::memory_limiter; + use mz_metrics::usage::ProcStatus; use mz_ore::cast::CastInto; use tracing::{debug, error}; @@ -84,7 +73,7 @@ mod linux { pub fn collect_heap_usage() -> (Option, Option) { use mz_ore::cast::CastInto; - match memory_limiter::ProcStatus::from_proc() { + match ProcStatus::from_proc() { Ok(status) => { let memory_bytes = status.vm_rss.cast_into(); let swap_bytes = status.vm_swap.cast_into(); diff --git a/src/compute-client/src/logging.rs b/src/compute-client/src/logging.rs index d3b788af63efe..86df318074e85 100644 --- a/src/compute-client/src/logging.rs +++ b/src/compute-client/src/logging.rs @@ -184,6 +184,8 @@ pub enum ComputeLog { DataflowGlobal, /// Prometheus metrics gathered from the metrics registry. PrometheusMetrics, + /// Resource usage observations of each replica process. + ResourceUsage, } impl LogVariant { @@ -413,6 +415,14 @@ impl LogVariant { .with_column("help", SqlScalarType::String.nullable(false)) .with_key(vec![0, 1, 3]) .finish(), + + LogVariant::Compute(ComputeLog::ResourceUsage) => RelationDesc::builder() + .with_column("process_id", SqlScalarType::UInt64.nullable(false)) + .with_column("source", SqlScalarType::String.nullable(false)) + .with_column("metric", SqlScalarType::String.nullable(false)) + .with_column("value", SqlScalarType::UInt64.nullable(false)) + .with_key(vec![0, 1, 2]) + .finish(), } } } diff --git a/src/compute/src/logging.rs b/src/compute/src/logging.rs index b7414f8b2fbe6..e8352b134ca9a 100644 --- a/src/compute/src/logging.rs +++ b/src/compute/src/logging.rs @@ -14,20 +14,21 @@ mod differential; pub(super) mod initialize; mod prometheus; mod reachability; +mod resource_usage; mod timely; use std::any::Any; use std::collections::BTreeMap; use std::marker::PhantomData; use std::rc::Rc; -use std::time::Duration; +use std::time::{Duration, Instant}; use ::timely::container::{CapacityContainerBuilder, PushInto}; use ::timely::dataflow::Stream; use ::timely::dataflow::channels::pact::Pipeline; use ::timely::dataflow::operators::capture::{Event, EventLink, EventPusher}; use ::timely::dataflow::operators::generic::Session; -use ::timely::dataflow::operators::{InputCapability, Operator}; +use ::timely::dataflow::operators::{Capability, CapabilityTrait, InputCapability, Operator}; use ::timely::progress::Timestamp as TimelyTimestamp; use ::timely::scheduling::Activator; use ::timely::{Container, ContainerBuilder}; @@ -211,6 +212,74 @@ impl PermutedRowPacker { } } +/// Downgrade `cap` to the next logging-interval boundary and schedule the operator's next +/// activation there. Returns the time the capability now holds. +/// +/// `now` and `start_offset` must be the ones the logging dataflow was constructed with, so that +/// every collection in it reports on the same boundaries. Scheduling off the boundary rather than +/// off a fixed delay keeps the output frontier progressing at the logging rate without drifting +/// from wall-clock elapsed time. +/// +/// NOTE: downgrading the capability asserts the collection is complete up to the new time, so an +/// operator that samples less often than the logging interval publishes a stale value rather than +/// withholding it. +pub(super) fn downgrade_to_interval_boundary( + cap: &mut Capability, + activator: &Activator, + now: Instant, + start_offset: Duration, + interval_ms: u128, +) -> Timestamp { + let elapsed = now.elapsed().as_millis(); + let time_ms: u128 = ((elapsed + start_offset.as_millis()) / interval_ms + 1) * interval_ms; + let ts: Timestamp = time_ms.try_into().expect("must fit"); + cap.downgrade(&ts); + + let next_boundary_ms = time_ms - start_offset.as_millis(); + let next_activation = + now + Duration::from_millis(next_boundary_ms.try_into().expect("must fit")); + activator.activate_after(next_activation.saturating_duration_since(Instant::now())); + + ts +} + +/// Emit the difference between two snapshots of a sampled source as updates at `ts`. +/// +/// A sampled source reports its whole state on every read, so a changed value has to be expressed +/// as a retraction of the previous one paired with an insertion of the new one. A key absent from +/// `current` is retracted without a replacement, so a source that stops being readable drops out of +/// the collection rather than lingering at its last value. +/// +/// `pack` takes the packer as an argument instead of closing over it, because it hands back rows +/// borrowed from it. +pub(super) fn emit_snapshot_diff( + session: &mut Session<'_, '_, Timestamp, CB, P>, + packer: &mut PermutedRowPacker, + prev: &BTreeMap, + current: &BTreeMap, + ts: Timestamp, + pack: F, +) where + K: Ord, + V: PartialEq, + CB: ContainerBuilder + for<'a> PushInto<((&'a RowRef, &'a RowRef), Timestamp, Diff)>, + P: CapabilityTrait, + F: for<'a> Fn(&'a mut PermutedRowPacker, &K, &V) -> (&'a RowRef, &'a RowRef), +{ + for (key, value) in prev { + if current.get(key) != Some(value) { + let row = pack(packer, key, value); + session.give((row, ts, Diff::MINUS_ONE)); + } + } + for (key, value) in current { + if prev.get(key) != Some(value) { + let row = pack(packer, key, value); + session.give((row, ts, Diff::ONE)); + } + } +} + /// Information about a collection exported from a logging dataflow. struct LogCollection { /// Trace handle providing access to the logged records. diff --git a/src/compute/src/logging/initialize.rs b/src/compute/src/logging/initialize.rs index 860f303a5b864..245cef517d309 100644 --- a/src/compute/src/logging/initialize.rs +++ b/src/compute/src/logging/initialize.rs @@ -182,6 +182,17 @@ impl LoggingContext<'_> { ); collections.extend(prometheus_collections); + let super::resource_usage::Return { + collections: resource_usage_collections, + } = super::resource_usage::construct( + scope, + self.config, + self.now, + self.start_offset, + self.workers_per_process, + ); + collections.extend(resource_usage_collections); + let errs = scope.scoped("logging errors", |scope| { let collection: KeyCollection<_, DataflowErrorSer, Diff> = VecCollection::empty(scope).into(); diff --git a/src/compute/src/logging/prometheus.rs b/src/compute/src/logging/prometheus.rs index 431107ad292be..9c71433fdcac4 100644 --- a/src/compute/src/logging/prometheus.rs +++ b/src/compute/src/logging/prometheus.rs @@ -19,7 +19,7 @@ use mz_ore::cast::{CastFrom, CastLossy}; use mz_ore::collections::CollectionExt; use mz_ore::metrics::MetricsRegistry; use mz_ore::soft_panic_or_log; -use mz_repr::{Datum, Diff, Timestamp}; +use mz_repr::{Datum, Timestamp}; use mz_timely_util::columnar::batcher; use mz_timely_util::columnar::builder::ColumnBuilder; use mz_timely_util::columnar::{Col2ValBatcher, columnar_exchange}; @@ -30,7 +30,10 @@ use timely::dataflow::operators::generic::OutputBuilder; use timely::dataflow::operators::generic::builder_rc::OperatorBuilder; use crate::extensions::arrange::MzArrangeCore; -use crate::logging::{ComputeLog, LogCollection, LogVariant, PermutedRowPacker}; +use crate::logging::{ + ComputeLog, LogCollection, LogVariant, PermutedRowPacker, downgrade_to_interval_boundary, + emit_snapshot_diff, +}; use crate::typedefs::RowRowSpine; use mz_row_spine::RowRowBuilder; @@ -88,24 +91,8 @@ pub(super) fn construct( move |_frontiers| { let Some(cap) = &mut cap else { return }; - // Advance the capability to the next logging interval boundary. - // This keeps the output frontier progressing at the logging - // rate, even when scrapes happen less frequently. Note that - // advancing the frontier implies the data is up-to-date, but - // the metrics snapshot may be stale by up to the scrape - // interval when it exceeds the logging interval. - let elapsed = now.elapsed().as_millis(); - let time_ms: u128 = - ((elapsed + start_offset.as_millis()) / interval_ms + 1) * interval_ms; - let ts: Timestamp = time_ms.try_into().expect("must fit"); - cap.downgrade(&ts); - - // Schedule the next activation at the interval boundary - // to avoid drift from wall-clock elapsed time. - let next_boundary_ms = time_ms - start_offset.as_millis(); - let next_activation = - now + Duration::from_millis(next_boundary_ms.try_into().expect("must fit")); - activator.activate_after(next_activation.saturating_duration_since(Instant::now())); + let ts = + downgrade_to_interval_boundary(cap, &activator, now, start_offset, interval_ms); // Only scrape when the scrape interval has elapsed. // The operator wakes every logging interval to advance the @@ -131,44 +118,18 @@ pub(super) fn construct( // Diff against previous snapshot and emit packed Row pairs. let mut output = output.activate(); let mut session = output.session_with_builder(&cap); - - // Retract entries that were removed or changed. - for (key, old_val) in &prev_snapshot { - match new_snapshot.get(key) { - Some(new_val) if new_val == old_val => {} - _ => { - let (row_key, row_val) = pack_row( - &mut packer, - &key.0, - old_val.1, - &key.1, - old_val.0, - &old_val.2, - process_id, - ); - session.give(((row_key, row_val), ts, Diff::MINUS_ONE)); - } - } - } - - // Insert entries that are new or changed. - for (key, new_val) in &new_snapshot { - match prev_snapshot.get(key) { - Some(old_val) if old_val == new_val => {} - _ => { - let (row_key, row_val) = pack_row( - &mut packer, - &key.0, - new_val.1, - &key.1, - new_val.0, - &new_val.2, - process_id, - ); - session.give(((row_key, row_val), ts, Diff::ONE)); - } - } - } + emit_snapshot_diff( + &mut session, + &mut packer, + &prev_snapshot, + &new_snapshot, + ts, + |packer, key, value| { + pack_row( + packer, &key.0, value.1, &key.1, value.0, &value.2, process_id, + ) + }, + ); prev_snapshot = new_snapshot; } diff --git a/src/compute/src/logging/resource_usage.rs b/src/compute/src/logging/resource_usage.rs new file mode 100644 index 0000000000000..6d1cf773afbdd --- /dev/null +++ b/src/compute/src/logging/resource_usage.rs @@ -0,0 +1,143 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +//! Logging dataflow for the resource usage of the replica's processes. +//! +//! The observations are read by `mz_metrics::usage`, on a task that keeps sampling while the +//! timely workers are busy. This dataflow only reports them, so a worker that stalls delays the +//! report without losing an observation: the values it reads are either kernel-maintained +//! high-water marks or the sampler's own folded peaks, neither of which a late read can miss. +//! +//! One row per `(process_id, source, metric)`, so a metric that moves every sample does not drag +//! the stable ones through a retraction with it. + +use std::collections::BTreeMap; +use std::rc::Rc; +use std::time::{Duration, Instant}; + +use mz_metrics::usage::{MetricKey, observations}; +use mz_ore::cast::CastFrom; +use mz_ore::collections::CollectionExt; +use mz_repr::{Datum, Timestamp}; +use mz_row_spine::RowRowBuilder; +use mz_timely_util::columnar::builder::ColumnBuilder; +use mz_timely_util::columnar::{Col2ValBatcher, batcher, columnar_exchange}; +use timely::dataflow::Scope; +use timely::dataflow::channels::pact::ExchangeCore; +use timely::dataflow::operators::generic::OutputBuilder; +use timely::dataflow::operators::generic::builder_rc::OperatorBuilder; + +use crate::extensions::arrange::MzArrangeCore; +use crate::logging::{ + ComputeLog, LogCollection, LogVariant, PermutedRowPacker, downgrade_to_interval_boundary, + emit_snapshot_diff, +}; +use crate::typedefs::RowRowSpine; + +/// The return type of [`construct`]. +pub(super) struct Return { + /// Collections to export. + pub collections: BTreeMap, +} + +/// Constructs the logging dataflow fragment for process resource usage. +pub(super) fn construct( + scope: Scope<'_, Timestamp>, + config: &mz_compute_client::logging::LoggingConfig, + now: Instant, + start_offset: Duration, + workers_per_process: usize, +) -> Return { + let variant = LogVariant::Compute(ComputeLog::ResourceUsage); + let mut collections = BTreeMap::new(); + let interval_ms = std::cmp::max(1, config.interval.as_millis()); + + if !config.index_logs.contains_key(&variant) { + return Return { collections }; + } + + let process_id = scope.index() / workers_per_process; + let enable = scope.index() % workers_per_process == 0; + + let mut builder = OperatorBuilder::new("ResourceUsage".to_string(), scope.clone()); + let (output, stream) = builder.new_output(); + let mut output = OutputBuilder::<_, ColumnBuilder<_>>::from(output); + + let operator_info = builder.operator_info(); + builder.build(move |capabilities| { + // Usage is per-process, so only one worker per process reports it. Drop the capability for + // disabled workers so the frontier can advance without this operator holding it back. + let mut cap = enable.then_some(capabilities.into_element()); + let activator = scope.activator_for(operator_info.address); + + let mut prev: BTreeMap = BTreeMap::new(); + let mut packer = PermutedRowPacker::new(ComputeLog::ResourceUsage); + + move |_frontiers| { + let Some(cap) = &mut cap else { return }; + + // The capability is downgraded on this operator's own timer rather than on the + // sampler's, so a sampler that stops ticking cannot freeze this collection's frontier. + let ts = + downgrade_to_interval_boundary(cap, &activator, now, start_offset, interval_ms); + + let current = observations().unwrap_or_default(); + if prev == current { + return; + } + + let mut output = output.activate(); + let mut session = output.session_with_builder(&cap); + emit_snapshot_diff( + &mut session, + &mut packer, + &prev, + ¤t, + ts, + |packer, key, value| pack_row(packer, process_id, *key, *value), + ); + + prev = current; + } + }); + + let exchange = ExchangeCore::, _>::new_core( + columnar_exchange::, + ); + let trace = stream + .mz_arrange_core::< + _, + batcher::Chunker<_>, + Col2ValBatcher<_, _, _, _>, + RowRowBuilder<_, _>, + RowRowSpine<_, _>, + >(exchange, "Arrange ResourceUsage") + .trace; + let token: Rc = Rc::new(()); + let collection = LogCollection { trace, token }; + collections.insert(variant, collection); + + Return { collections } +} + +/// Pack one observation into key/value row pairs. +fn pack_row( + packer: &mut PermutedRowPacker, + process_id: usize, + (source, metric): MetricKey, + value: u64, +) -> (&mz_repr::RowRef, &mz_repr::RowRef) { + packer.pack_by_index(|row_packer, index| match index { + 0 => row_packer.push(Datum::UInt64(u64::cast_from(process_id))), + 1 => row_packer.push(Datum::String(source)), + 2 => row_packer.push(Datum::String(metric)), + 3 => row_packer.push(Datum::UInt64(value)), + _ => unreachable!("unexpected column index {index}"), + }) +} diff --git a/src/compute/src/memory_limiter.rs b/src/compute/src/memory_limiter.rs index 947f0cb9ec867..cf7e07d903c54 100644 --- a/src/compute/src/memory_limiter.rs +++ b/src/compute/src/memory_limiter.rs @@ -15,11 +15,11 @@ use std::sync::Mutex; use std::time::{Duration, Instant}; -use anyhow::Context; use mz_compute_types::dyncfgs::{ MEMORY_LIMITER_BURST_FACTOR, MEMORY_LIMITER_INTERVAL, MEMORY_LIMITER_USAGE_BIAS, }; use mz_dyncfg::ConfigSet; +use mz_metrics::usage::ProcStatus; use mz_ore::cast::{CastFrom, CastLossy}; use mz_ore::metric; use mz_ore::metrics::{MetricsRegistry, UIntGauge}; @@ -193,10 +193,7 @@ impl LimiterTask { Err(err) } #[cfg(not(target_os = "linux"))] - Err(_err) => Ok(ProcStatus { - vm_rss: 0, - vm_swap: 0, - }), + Err(_err) => Ok(ProcStatus::default()), } } @@ -204,7 +201,9 @@ impl LimiterTask { fn check(&mut self) -> Result<(), anyhow::Error> { debug!("checking memory limits"); - let ProcStatus { vm_rss, vm_swap } = Self::current_utilization()?; + let ProcStatus { + vm_rss, vm_swap, .. + } = Self::current_utilization()?; let memory_limit = self.config.memory_limit; let burst_budget_remaining = self.burst_budget_remaining; @@ -315,14 +314,6 @@ impl LimiterMetrics { } } -/// Helper for reading and parsing `/proc/self/status` on Linux. -pub struct ProcStatus { - /// Resident Set Size (RSS) in bytes. - pub vm_rss: usize, - /// Swap memory in bytes. - pub vm_swap: usize, -} - #[cfg(test)] mod tests { use super::*; @@ -383,34 +374,3 @@ mod tests { assert_eq!(task.last_check, stale); } } - -impl ProcStatus { - /// Populate a new `ProcStatus` with information in /proc/self/status. - pub fn from_proc() -> anyhow::Result { - let contents = std::fs::read_to_string("/proc/self/status")?; - let mut vm_rss = 0; - let mut vm_swap = 0; - - for line in contents.lines() { - if line.starts_with("VmRSS:") { - vm_rss = line - .split_whitespace() - .nth(1) - .ok_or_else(|| anyhow::anyhow!("failed to parse VmRSS"))? - .parse::() - .context("failed to parse VmRSS")? - * 1024 - } else if line.starts_with("VmSwap:") { - vm_swap = line - .split_whitespace() - .nth(1) - .ok_or_else(|| anyhow::anyhow!("failed to parse VmSwap"))? - .parse::() - .context("failed to parse VmSwap")? - * 1024; - } - } - - Ok(Self { vm_rss, vm_swap }) - } -} diff --git a/src/environmentd/src/environmentd/main.rs b/src/environmentd/src/environmentd/main.rs index 60a744a57d0d2..73d1ea5350307 100644 --- a/src/environmentd/src/environmentd/main.rs +++ b/src/environmentd/src/environmentd/main.rs @@ -755,6 +755,8 @@ fn run(mut args: Args) -> Result<(), anyhow::Error> { runtime.block_on(mz_metrics::register_metrics_into( &metrics_registry, mz_dyncfgs::all_dyncfgs(), + // environmentd has no scratch directory, so it tracks no disk usage. + None, )); // Initialize fail crate for failpoint support diff --git a/src/metrics/Cargo.toml b/src/metrics/Cargo.toml index af03146bef6c3..8f40864dd6f1c 100644 --- a/src/metrics/Cargo.toml +++ b/src/metrics/Cargo.toml @@ -10,15 +10,20 @@ publish = false workspace = true [dependencies] +anyhow.workspace = true lgalloc.workspace = true libc.workspace = true mz-dyncfg = { path = "../dyncfg" } mz-ore = { path = "../ore", features = ["metrics"] } +nix.workspace = true paste.workspace = true prometheus.workspace = true thiserror.workspace = true tokio.workspace = true tracing.workspace = true +[dev-dependencies] +mz-ore = { path = "../ore", features = ["metrics", "test"] } + [features] default = [] diff --git a/src/metrics/src/dyncfgs.rs b/src/metrics/src/dyncfgs.rs index f75c44a314ee6..22f4f5ed2b1b2 100644 --- a/src/metrics/src/dyncfgs.rs +++ b/src/metrics/src/dyncfgs.rs @@ -37,10 +37,24 @@ pub(crate) const MZ_METRICS_RUSAGE_REFRESH_INTERVAL: Config = Config:: ParameterScope::Replica, ); +/// How frequently to sample process resource usage. +/// +/// This interval bounds how short a spike can be and still be observed, but only for the sources +/// with no kernel-side high-water mark. A kernel-maintained peak such as `cgroup memory.peak` is +/// exact no matter how rarely it is read. Deliberately separate from `memory_limiter_interval`, +/// which governs OOM-kill behavior and must not be retuned for introspection's sake. +pub(crate) const MZ_METRICS_USAGE_REFRESH_INTERVAL: Config = Config::new( + "mz_metrics_usage_refresh_interval", + Duration::from_secs(5), + "How frequently to sample process resource usage. A zero duration disables sampling.", + ParameterScope::Replica, +); + /// Adds the full set of all storage `Config`s. pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { configs .add(&MZ_METRICS_LGALLOC_MAP_REFRESH_INTERVAL) .add(&MZ_METRICS_LGALLOC_REFRESH_INTERVAL) .add(&MZ_METRICS_RUSAGE_REFRESH_INTERVAL) + .add(&MZ_METRICS_USAGE_REFRESH_INTERVAL) } diff --git a/src/metrics/src/lib.rs b/src/metrics/src/lib.rs index d024044763a31..ff0df44a62f44 100644 --- a/src/metrics/src/lib.rs +++ b/src/metrics/src/lib.rs @@ -17,6 +17,7 @@ #![warn(missing_docs, missing_debug_implementations)] +use std::path::PathBuf; use std::time::Duration; use mz_dyncfg::{ConfigSet, ConfigUpdates}; @@ -28,6 +29,7 @@ pub use dyncfgs::all_dyncfgs; mod dyncfgs; pub mod lgalloc; pub mod rusage; +pub mod usage; /// Handle to metrics defined in this crate. #[derive(Debug)] @@ -36,6 +38,7 @@ pub struct Metrics { lgalloc: MetricsTask, lgalloc_map: MetricsTask, rusage: MetricsTask, + usage: MetricsTask, } static METRICS: std::sync::Mutex> = std::sync::Mutex::new(None); @@ -47,8 +50,15 @@ static METRICS: std::sync::Mutex> = std::sync::Mutex::new(None); /// remove the shared static mutex and make this function return a handle to the metrics. /// /// This function is async, because it needs to be called from a tokio runtime context. +/// +/// `disk_root` is a directory whose filesystem usage should be tracked, or `None` for processes +/// that do not use disk. #[allow(clippy::unused_async)] -pub async fn register_metrics_into(metrics_registry: &MetricsRegistry, config_set: ConfigSet) { +pub async fn register_metrics_into( + metrics_registry: &MetricsRegistry, + config_set: ConfigSet, + disk_root: Option, +) { let update_duration_metric = metrics_registry.register(mz_ore::metric!( name: "mz_metrics_update_duration", help: "The time it took to update lgalloc stats", @@ -75,10 +85,18 @@ pub async fn register_metrics_into(metrics_registry: &MetricsRegistry, config_se &update_duration_metric, ); + let usage = Metrics::new_metrics_task( + metrics_registry, + |registry| usage::register_metrics_into(registry, disk_root), + dyncfgs::MZ_METRICS_USAGE_REFRESH_INTERVAL, + &update_duration_metric, + ); + *METRICS.lock().expect("lock poisoned") = Some(Metrics { lgalloc, lgalloc_map, rusage, + usage, config_set, }); } @@ -146,6 +164,7 @@ impl Metrics { self.lgalloc.update_dyncfg(&self.config_set); self.lgalloc_map.update_dyncfg(&self.config_set); self.rusage.update_dyncfg(&self.config_set); + self.usage.update_dyncfg(&self.config_set); } fn new_metrics_task( diff --git a/src/metrics/src/rusage.rs b/src/metrics/src/rusage.rs index f1d97fded526a..c5b59cab95b5c 100644 --- a/src/metrics/src/rusage.rs +++ b/src/metrics/src/rusage.rs @@ -133,6 +133,22 @@ metrics! { (ru_nivcsw, "involuntary context switches", "_total", Unitless) } +/// Read this process's peak resident set size, in bytes. +/// +/// The kernel maintains this high-water mark itself, so it cannot miss a short-lived spike the +/// way a sampled maximum can. +pub(crate) fn max_rss_bytes() -> Result { + let rusage = unsafe { + let mut rusage = std::mem::zeroed(); + let ret = libc::getrusage(libc::RUSAGE_SELF, &mut rusage); + if ret < 0 { + return Err(std::io::Error::last_os_error()); + } + rusage + }; + Ok(::from(rusage.ru_maxrss)) +} + /// Register a task to read rusage stats. pub(crate) fn register_metrics_into(metrics_registry: &MetricsRegistry) -> RuMetrics { RuMetrics::new(metrics_registry) diff --git a/src/metrics/src/usage.rs b/src/metrics/src/usage.rs new file mode 100644 index 0000000000000..ddea17fd67f21 --- /dev/null +++ b/src/metrics/src/usage.rs @@ -0,0 +1,449 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +//! Resource usage observations for the current process. +//! +//! This module is a mechanism, not a policy. It reports what each source says, under that +//! source's own name, and never combines two sources into a third figure. The discrepancies +//! between sources carry information: cgroup memory far above `VmRSS` means page cache or kernel +//! memory is charged to the replica, which is exactly the case a single fused "memory" number +//! hides. Deciding which source answers "how much memory is this replica using" belongs to the +//! SQL views built on top. +//! +//! Peaks are observations too. `cgroup memory.peak` and `getrusage`'s `ru_maxrss` are high-water +//! marks the kernel maintains itself, so they carry no sampling error and outlive anything short +//! of the process exiting. Only a source with no kernel-side peak gets one folded here, reported +//! under a distinct metric name so a caller can tell an exact peak from a sampled one. + +use std::collections::BTreeMap; +use std::convert::Infallible; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; + +use anyhow::Context; +use mz_ore::cast::CastFrom; +use mz_ore::cgroup::CgroupV2; +use mz_ore::metric; +use mz_ore::metrics::{MetricsRegistry, raw}; +use tracing::{debug, info}; + +use crate::MetricsUpdate; + +/// Identifies one observation: which reader produced it, and what it measures. +/// +/// Static strings rather than an enum, so adding a metric touches only the sampler. +pub type MetricKey = (&'static str, &'static str); + +/// The most recent observations, keyed by [`MetricKey`]. +/// +/// Published as a whole map rather than as independent cells so a reader always sees one +/// self-consistent sample. Reading a set of independent atomics could mix two samples into a +/// combination that never existed, for instance a peak below the current value it bounds. +static OBSERVATIONS: Mutex>> = Mutex::new(None); + +/// Source names, as reported alongside each observation. +pub mod source { + /// This process's cgroup v2 interface files. + pub const CGROUP: &str = "cgroup"; + /// `getrusage(RUSAGE_SELF)`. + pub const RUSAGE: &str = "rusage"; + /// `/proc/self/status`. + pub const PROC_STATUS: &str = "proc_status"; + /// `statvfs` on the scratch directory's filesystem. + pub const STATVFS: &str = "statvfs"; +} + +/// Return the most recent resource usage observations of this process. +/// +/// `None` until the sampler has taken its first sample, and forever in a process that never +/// registered one. An observation the sampler could not read is absent from the map, never +/// present as zero. +pub fn observations() -> Option> { + OBSERVATIONS.lock().expect("poisoned").clone() +} + +/// Observations that get a peak folded in this process, because their source has no kernel-side +/// high-water mark. Each is a maximum over samples and therefore a lower bound on the true peak. +const DERIVED_PEAKS: &[(MetricKey, &str)] = &[ + ((source::STATVFS, "fs_used"), "fs_used_peak"), + // Load-bearing only below the kernel version that provides `memory.swap.peak`. + ((source::PROC_STATUS, "vm_swap"), "vm_swap_peak"), + // A lower bound on the peak of the quantity the memory limiter enforces. No upper bound is + // available: the cgroup peaks describe a smaller quantity, excluding resident file-backed + // pages, and `ru_maxrss` is refreshed at kernel checkpoints and has been seen reading below + // the concurrent `vm_rss`. + ((source::PROC_STATUS, "heap"), "heap_peak"), +]; + +/// Sampler of resource usage observations, driven by the metrics update task. +/// +/// Sampling happens here rather than in a compute logging operator because a saturated timely +/// worker stops scheduling its logging operators during exactly the episodes we most want +/// sampled. Folding the derived peaks here also keeps them alive across a teardown and rebuild of +/// the logging dataflow, which an operator-local fold would lose. +pub(crate) struct UsageMetrics { + /// This process's cgroup, if it has a v2 one with the memory controller enabled. + cgroup: Option, + /// Directory whose filesystem usage is tracked, if disk is in use. + disk_root: Option, + /// Peaks folded here, for the sources listed in [`DERIVED_PEAKS`]. + derived_peaks: BTreeMap, + gauges: raw::UIntGaugeVec, +} + +impl UsageMetrics { + fn new(registry: &MetricsRegistry, disk_root: Option) -> Self { + // Readings taken from the wrong cgroup look plausible rather than absent, and diagnosing + // that otherwise takes access to the container. Reported once, at registration. + let cgroup = CgroupV2::detect(); + match &cgroup { + Some(cgroup) => info!( + dir = %cgroup.path().display(), + "reading resource usage from cgroup v2", + ), + None => info!("no cgroup v2 with a memory controller; cgroup usage unavailable"), + } + + Self { + cgroup, + disk_root, + derived_peaks: BTreeMap::new(), + gauges: registry.register(metric!( + name: "mz_metrics_resource_usage", + help: "Resource usage observations, by source and metric.", + var_labels: ["source", "metric"], + )), + } + } + + /// Read every source, without interpreting any of them. + fn sample(&self) -> BTreeMap { + let mut out = BTreeMap::new(); + let mut put = |source: &'static str, metric: &'static str, value: Option| { + if let Some(value) = value { + out.insert((source, metric), value); + } + }; + + if let Some(cgroup) = &self.cgroup { + // `memory.peak` and `memory.swap.peak` are absent on kernels too old to provide them + // and read as `None` there. `memory.current` is the accounting that limit enforcement + // and the OOM killer act on, which is why it is worth reporting next to `vm_rss`. + let files: &[(&'static str, &str)] = &[ + ("memory_current", "memory.current"), + ("memory_peak", "memory.peak"), + ("memory_max", "memory.max"), + ("swap_current", "memory.swap.current"), + ("swap_peak", "memory.swap.peak"), + ("swap_max", "memory.swap.max"), + ]; + for &(metric, file) in files { + put(source::CGROUP, metric, cgroup.read_u64(file)); + } + + // `oom_kill` counts kills inside this cgroup and `max` counts times the limit was + // hit, which together answer whether a replica died of memory pressure. + let keyed: &[(&'static str, &str, &str)] = &[ + ("anon", "memory.stat", "anon"), + ("file", "memory.stat", "file"), + ("shmem", "memory.stat", "shmem"), + // Pages held in memory with their swap slot still allocated. They are charged + // twice, once as `anon` here and once in `memory.swap.current`, and they are the + // whole of the difference between that and `proc_status vm_swap`. + ("swapcached", "memory.stat", "swapcached"), + ("kernel", "memory.stat", "kernel"), + ("slab", "memory.stat", "slab"), + ("sock", "memory.stat", "sock"), + ("events_max", "memory.events", "max"), + ("events_oom_kill", "memory.events", "oom_kill"), + ]; + for &(metric, file, key) in keyed { + put(source::CGROUP, metric, cgroup.read_keyed_u64(file, key)); + } + } + + put(source::RUSAGE, "max_rss", max_rss_bytes()); + + match ProcStatus::from_proc() { + Ok(status) => { + put(source::PROC_STATUS, "vm_rss", Some(status.rss())); + put(source::PROC_STATUS, "vm_swap", Some(status.swap())); + // The quantity the memory limiter enforces against `--heap-limit`. Reported as + // its own observation because the kernel maintains no combined memory-plus-swap + // peak, so a peak of the sum has to be folded from samples of the sum. It is one + // source added to itself, not two sources fused. + put(source::PROC_STATUS, "heap", Some(status.heap())); + // Decomposes `vm_rss`. `rss_file` is the part charged to another cgroup, so it + // explains the gap between `vm_rss` and `cgroup memory_current`. + put(source::PROC_STATUS, "rss_anon", Some(status.rss_anon())); + put(source::PROC_STATUS, "rss_file", Some(status.rss_file())); + put(source::PROC_STATUS, "rss_shmem", Some(status.rss_shmem())); + } + Err(err) => debug!("failed to read /proc/self/status: {err}"), + } + + if let Some(root) = self.disk_root.as_deref() { + // NOTE: filesystem-wide used bytes, not this process's usage. Named for what it is, + // since on a shared filesystem it counts writes this replica never made. + match disk_usage(root) { + Ok(bytes) => put(source::STATVFS, "fs_used", Some(bytes)), + Err(err) => debug!("statvfs on {} failed: {err}", root.display()), + } + } + + out + } + + /// Fold the derived peaks over `sample`, adding them to it. + fn fold_derived_peaks(&mut self, sample: &mut BTreeMap) { + for ((source, metric), peak_metric) in DERIVED_PEAKS { + let Some(&value) = sample.get(&(*source, *metric)) else { + continue; + }; + let peak = self + .derived_peaks + .entry((source, peak_metric)) + .and_modify(|peak| *peak = (*peak).max(value)) + .or_insert(value); + sample.insert((source, peak_metric), *peak); + } + } +} + +impl MetricsUpdate for UsageMetrics { + type Error = Infallible; + const NAME: &'static str = "usage"; + + fn update(&mut self) -> Result<(), Self::Error> { + let mut sample = self.sample(); + self.fold_derived_peaks(&mut sample); + + for ((source, metric), value) in &sample { + self.gauges.with_label_values(&[source, metric]).set(*value); + } + + *OBSERVATIONS.lock().expect("poisoned") = Some(sample); + + Ok(()) + } +} + +/// Register the resource usage sampler. +/// +/// `disk_root` is a directory on the filesystem whose usage should be tracked, or `None` if this +/// process does not use disk. +pub(crate) fn register_metrics_into( + registry: &MetricsRegistry, + disk_root: Option, +) -> UsageMetrics { + UsageMetrics::new(registry, disk_root) +} + +/// Return the used bytes of the filesystem containing `root`. +/// +/// Callers decide how to report a failure. The sampler polls this on a short interval, so logging +/// an error here would repeat for as long as the directory is unavailable. +pub fn disk_usage(root: &Path) -> Result { + let stat = nix::sys::statvfs::statvfs(root)?; + + // `fsblkcnt_t` is a `u32` on macOS but a `u64` on Linux. + #[allow(clippy::useless_conversion)] + let used_blocks = u64::from(stat.blocks() - stat.blocks_available()); + let used_bytes = used_blocks * stat.fragment_size(); + + debug!("disk usage: {used_bytes}"); + + Ok(used_bytes) +} + +/// Return this process's peak resident set size, in bytes. +/// +/// This is the kernel's own high-water mark, so unlike a sampled maximum it cannot miss a +/// short-lived spike. +fn max_rss_bytes() -> Option { + match crate::rusage::max_rss_bytes() { + Ok(bytes) => u64::try_from(bytes).ok(), + Err(err) => { + debug!("getrusage failed: {err}"); + None + } + } +} + +/// Memory usage of the current process, read from `/proc/self/status`. +/// +/// The `rss_*` fields decompose `vm_rss`. The decomposition is load-bearing rather than +/// decorative: `rss_file` counts pages of file-backed mappings, most of it this binary's own text, +/// and those pages are charged to whichever cgroup first faulted them in. On a Kubernetes node +/// that is the runtime that unpacked the image, not the replica, so `vm_rss` runs a roughly +/// constant amount above the replica's own cgroup charge. +#[derive(Clone, Copy, Debug, Default)] +pub struct ProcStatus { + /// Resident Set Size (RSS) in bytes. + pub vm_rss: usize, + /// Swap memory in bytes. + pub vm_swap: usize, + /// Resident anonymous memory in bytes. + pub rss_anon: usize, + /// Resident file-backed memory in bytes. + pub rss_file: usize, + /// Resident shared memory in bytes. + pub rss_shmem: usize, +} + +impl ProcStatus { + /// Read a new `ProcStatus` from `/proc/self/status`. + /// + /// Fails on platforms without a Linux-style procfs. + pub fn from_proc() -> anyhow::Result { + let contents = std::fs::read_to_string("/proc/self/status")?; + let mut status = Self::default(); + + for line in contents.lines() { + let (field, target) = match line.split_once(':') { + Some(("VmRSS", rest)) => ("VmRSS", (&mut status.vm_rss, rest)), + Some(("VmSwap", rest)) => ("VmSwap", (&mut status.vm_swap, rest)), + Some(("RssAnon", rest)) => ("RssAnon", (&mut status.rss_anon, rest)), + Some(("RssFile", rest)) => ("RssFile", (&mut status.rss_file, rest)), + Some(("RssShmem", rest)) => ("RssShmem", (&mut status.rss_shmem, rest)), + _ => continue, + }; + let (slot, rest) = target; + *slot = parse_kib(rest).with_context(|| format!("failed to parse {field}"))?; + } + + Ok(status) + } + + /// Memory (RAM) usage, in bytes. + pub fn rss(&self) -> u64 { + u64::cast_from(self.vm_rss) + } + + /// Swap usage, in bytes. + pub fn swap(&self) -> u64 { + u64::cast_from(self.vm_swap) + } + + /// Heap (RAM + swap) usage, in bytes. + pub fn heap(&self) -> u64 { + self.rss().saturating_add(self.swap()) + } + + /// Resident anonymous memory, in bytes. + pub fn rss_anon(&self) -> u64 { + u64::cast_from(self.rss_anon) + } + + /// Resident file-backed memory, in bytes. + pub fn rss_file(&self) -> u64 { + u64::cast_from(self.rss_file) + } + + /// Resident shared memory, in bytes. + pub fn rss_shmem(&self) -> u64 { + u64::cast_from(self.rss_shmem) + } +} + +/// Parse the value part of a `/proc/self/status` line reporting a size in KiB, returning bytes. +fn parse_kib(rest: &str) -> anyhow::Result { + let kib: usize = rest + .split_whitespace() + .next() + .ok_or_else(|| anyhow::anyhow!("missing value: {rest}"))? + .parse()?; + Ok(kib * 1024) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn metrics_for_test() -> UsageMetrics { + UsageMetrics::new(&MetricsRegistry::new(), None) + } + + /// Something must be measurable on the platforms we test on, otherwise the other tests here + /// assert nothing. + #[mz_ore::test] + fn sample_is_not_empty() { + let metrics = metrics_for_test(); + assert!(!metrics.sample().is_empty()); + } + + /// A derived peak must rise with a higher observation and survive a lower one. + /// + /// Drives the fold over a synthetic sample rather than through `update`, so the assertion does + /// not depend on this machine's disk usage actually moving. + #[mz_ore::test] + fn derived_peaks_are_monotonic() { + let mut metrics = metrics_for_test(); + let key = (source::STATVFS, "fs_used"); + let peak_key = (source::STATVFS, "fs_used_peak"); + + let mut fold = |value| { + let mut sample = BTreeMap::from_iter([(key, value)]); + metrics.fold_derived_peaks(&mut sample); + sample[&peak_key] + }; + + assert_eq!(fold(100), 100, "first observation sets the peak"); + assert_eq!(fold(200), 200, "a higher observation raises the peak"); + assert_eq!(fold(50), 200, "a lower observation must not lower the peak"); + } + + /// `vm_rss` must decompose exactly into its three parts, since a caller comparing `rss_file` + /// against a cgroup charge relies on the decomposition being complete. + #[mz_ore::test] + #[cfg_attr(not(target_os = "linux"), ignore = "requires a Linux procfs")] + fn vm_rss_decomposes() { + let status = ProcStatus::from_proc().expect("procfs available"); + assert_eq!( + status.rss(), + status.rss_anon() + status.rss_file() + status.rss_shmem(), + "vm_rss {} != anon {} + file {} + shmem {}", + status.rss(), + status.rss_anon(), + status.rss_file(), + status.rss_shmem(), + ); + } + + /// `heap` must be exactly the sum the memory limiter compares against its limit, since a + /// caller reading `heap_peak` to ask how close a replica came to a kill relies on it. + #[mz_ore::test] + #[cfg_attr(not(target_os = "linux"), ignore = "requires a Linux procfs")] + fn heap_is_rss_plus_swap() { + let metrics = metrics_for_test(); + let sample = metrics.sample(); + + let get = |metric| sample[&(source::PROC_STATUS, metric)]; + assert_eq!(get("heap"), get("vm_rss") + get("vm_swap")); + } + + /// A source that cannot be read is absent, never zero. + #[mz_ore::test] + fn unmeasured_observation_is_absent() { + let metrics = metrics_for_test(); + let sample = metrics.sample(); + + // No `disk_root` was configured, so nothing from `statvfs` may appear. + assert!(sample.keys().all(|(source, _)| *source != source::STATVFS)); + } + + /// A derived peak is only published for a source that was actually read. + #[mz_ore::test] + fn derived_peak_needs_an_observation() { + let mut metrics = metrics_for_test(); + let mut sample = BTreeMap::new(); + metrics.fold_derived_peaks(&mut sample); + assert!(sample.is_empty()); + } +} diff --git a/src/ore/src/cgroup.rs b/src/ore/src/cgroup.rs index 318546655248a..d1d851409103f 100644 --- a/src/ore/src/cgroup.rs +++ b/src/ore/src/cgroup.rs @@ -14,6 +14,11 @@ // limitations under the License. //! Linux cgroup detection utilities. +//! +//! NOTE: this module must stay free of non-`std` dependencies. It is compiled unconditionally, +//! including into feature-reduced builds such as the wasm32 one, where `mz_ore`'s optional +//! dependencies are absent. Reaching for `tracing` here breaks that build. Callers that want the +//! resolved cgroup reported should log [`CgroupV2::path`] themselves. use std::fs::File; use std::io::{BufRead, BufReader}; use std::path::{Path, PathBuf}; @@ -172,6 +177,69 @@ pub fn detect_memory_limit() -> Option { read_v1_memory_limit(&cgroups, &v1_mounts) } +/// The directory holding this process's cgroup v2 interface files. +/// +/// Resolving the directory walks `/proc/self/mountinfo` and `/proc/self/cgroup`, so callers that +/// read repeatedly should [`CgroupV2::detect`] once and keep the handle. +#[derive(Clone, Debug)] +pub struct CgroupV2 { + dir: PathBuf, +} + +impl CgroupV2 { + /// Resolve this process's cgroup v2 directory, if it has one with the memory controller + /// enabled. + /// + /// Returns `None` on a v1-only hierarchy, a mixed hierarchy, and on non-Linux platforms. + pub fn detect() -> Option { + let (v2_mounts, _v1_mounts) = parse_proc_self_mountinfo()?; + let cgroups = parse_proc_self_cgroup()?; + + // cgroups v2 supports only a single cgroup per process. + let mount = v2_mounts.first()?; + if mount.root != cgroups.first()?.root { + // Mixed v1/v2 hierarchies are not supported. + return None; + } + + let dir = &mount.mount_point; + let controllers = std::fs::read_to_string(dir.join("cgroup.controllers")).ok()?; + if !controllers.trim().split(' ').any(|c| c == "memory") { + return None; + } + + Some(Self { dir: dir.clone() }) + } + + /// The resolved directory. + /// + /// Readings taken from the wrong cgroup look plausible rather than absent, so callers are + /// expected to report this once so that a misresolution is diagnosable without container + /// access. This crate does not log it itself, since `tracing` is an optional dependency here. + pub fn path(&self) -> &Path { + &self.dir + } + + /// Read an interface file holding a single integer, in bytes or as a count. + /// + /// Returns `None` when the file is absent, which is how a kernel too old to provide it + /// presents, and when it holds a non-integer. `memory.max` and friends read as `None` when + /// unlimited, since they then hold the literal `max`. + pub fn read_u64(&self, file: &str) -> Option { + let contents = std::fs::read_to_string(self.dir.join(file)).ok()?; + contents.trim().parse().ok() + } + + /// Read a `key value` interface file, returning the value for `key`. + pub fn read_keyed_u64(&self, file: &str, key: &str) -> Option { + let contents = std::fs::read_to_string(self.dir.join(file)).ok()?; + contents.lines().find_map(|line| { + let (name, value) = line.split_once(' ')?; + (name == key).then(|| value.trim().parse().ok())? + }) + } +} + #[cfg(test)] mod tests { use std::path::PathBuf; diff --git a/src/ore/src/lib.rs b/src/ore/src/lib.rs index fd9b1f40da1f0..e1ea84d82fca1 100644 --- a/src/ore/src/lib.rs +++ b/src/ore/src/lib.rs @@ -30,6 +30,7 @@ pub mod bits; #[cfg(feature = "bytes")] pub mod bytes; pub mod cast; +pub mod cgroup; #[cfg_attr(nightly_doc_features, doc(cfg(feature = "async")))] #[cfg(feature = "async")] pub mod channel; diff --git a/src/ore/src/memory.rs b/src/ore/src/memory.rs index 76a2d73f0f86a..72ec906635dde 100644 --- a/src/ore/src/memory.rs +++ b/src/ore/src/memory.rs @@ -15,14 +15,6 @@ //! Physical memory introspection. -// Only this probe uses the cgroup helpers, so the module lives here rather -// than at the crate root. The file carries more surface than the probe -// needs, hence the dead-code allowance. -#[cfg(target_os = "linux")] -#[path = "cgroup.rs"] -#[allow(dead_code)] -mod cgroup; - /// Returns the physical memory available to this process in bytes: the /// host's RAM, clamped by the cgroup memory limit when one is set. Both /// cgroup v1 and v2 are honored, resolved through `/proc/self/mountinfo` @@ -81,7 +73,7 @@ fn host_memory_bytes() -> Option { /// The RAM limit of the cgroup governing this process, if any. #[cfg(target_os = "linux")] fn cgroup_memory_max() -> Option { - cgroup::detect_memory_limit()?.max + crate::cgroup::detect_memory_limit()?.max } #[cfg(not(target_os = "linux"))] diff --git a/src/pgrepr-consts/src/oid.rs b/src/pgrepr-consts/src/oid.rs index c2f7d677821a6..aa820195e6f1a 100644 --- a/src/pgrepr-consts/src/oid.rs +++ b/src/pgrepr-consts/src/oid.rs @@ -829,3 +829,4 @@ pub const VIEW_MZ_BUILTIN_VIEWS_OID: u32 = 17119; pub const MV_MZ_METRIC_SINKS_OID: u32 = 17120; pub const INDEX_MZ_METRIC_SINKS_IND_OID: u32 = 17121; pub const TABLE_MZ_OBJECT_HYDRATION_HISTORY_OID: u32 = 17122; +pub const LOG_MZ_CLUSTER_REPLICA_RESOURCE_USAGE_OID: u32 = 17123; diff --git a/test/cluster/mzcompose.py b/test/cluster/mzcompose.py index e687825fb2b64..804ff0e6bd5b6 100644 --- a/test/cluster/mzcompose.py +++ b/test/cluster/mzcompose.py @@ -8188,6 +8188,86 @@ def workflow_test_prometheus_metrics(c: Composition) -> None: """)) +def workflow_test_resource_usage(c: Composition) -> None: + """Test that mz_cluster_replica_resource_usage reports observations per replica process.""" + + c.up("materialized") + + # Sample fast, so the test does not have to wait out the default interval. + c.sql( + "ALTER SYSTEM SET mz_metrics_usage_refresh_interval = '1s';", + port=6877, + user="mz_system", + ) + c.sql("CREATE CLUSTER cluster1 SIZE 'scale=2,workers=2';") + + def observations() -> dict[tuple[int, str, str], int]: + with c.sql_cursor() as cursor: + cursor.execute(b"SET cluster = cluster1") + # `process_id` and `value` are `uint8`, which psycopg has no adapter for and hands + # back as strings. Cast so the comparisons below are numeric, not lexicographic. + cursor.execute(b""" + SELECT process_id::int8, source, metric, value::int8 + FROM mz_introspection.mz_cluster_replica_resource_usage + """) + return {(r[0], r[1], r[2]): r[3] for r in cursor.fetchall()} + + # Both processes must report, and every process must report the same set of metrics: the two + # run the same binary in the same environment, so a metric readable on one and not the other + # means a reader failed rather than that the source is unavailable. + for _ in range(60): + before = observations() + processes = {process_id for process_id, _, _ in before} + per_process = { + process_id: {(s, m) for p, s, m in before if p == process_id} + for process_id in processes + } + if processes == {0, 1} and len(set(map(frozenset, per_process.values()))) == 1: + break + time.sleep(1) + else: + assert False, f"resource usage not reported for both processes: {before}" + + # `rusage` and `proc_status` are available on any Linux replica, so their absence is a bug + # rather than an unsupported configuration. cgroup metrics are deliberately not asserted: + # `memory.peak` and `memory.swap.peak` depend on the kernel version. + metrics = {(source, metric) for _, source, metric in before} + for required in [("rusage", "max_rss"), ("proc_status", "vm_rss")]: + assert required in metrics, f"{required} missing from {sorted(metrics)}" + + # Every reported value must be a plain number. Nothing is allowed to surface as a sentinel. + for key, value in before.items(): + assert value >= 0, f"{key} reported {value}" + + # Do some work to push usage up. Peaks must not go backwards, whether or not this particular + # workload moves them. + c.sql(""" + SET cluster = cluster1; + CREATE TABLE t (a int); + INSERT INTO t SELECT generate_series(1, 500000); + CREATE MATERIALIZED VIEW mv AS SELECT count(*) FROM t; + """) + with c.sql_cursor() as cursor: + cursor.execute(b"SET cluster = cluster1") + cursor.execute(b"SELECT * FROM mv") + cursor.fetchall() + + # Give the sampler a few intervals to observe the new usage. + time.sleep(5) + + after = observations() + + # Peaks are monotonic, whether the kernel maintains them or the sampler folds them. Current + # values are free to fall, so they are deliberately not checked here. + peaks = [key for key in before if key[2].endswith("peak") or key[2] == "max_rss"] + assert peaks, f"no peak metrics reported: {sorted(before)}" + for key in peaks: + assert key in after, f"{key} stopped being reported" + assert ( + after[key] >= before[key] + ), f"{key} peak fell from {before[key]} to {after[key]}" + + def workflow_test_metrics_null_label(c: Composition) -> None: """SQL-198: `/metrics/mz_usage` must not abort environmentd when a Prometheus label column is SQL NULL. An unorchestrated cluster replica has diff --git a/test/sqllogictest/autogenerated/mz_introspection.slt b/test/sqllogictest/autogenerated/mz_introspection.slt index a914ba4344620..b1f881892146d 100644 --- a/test/sqllogictest/autogenerated/mz_introspection.slt +++ b/test/sqllogictest/autogenerated/mz_introspection.slt @@ -101,6 +101,14 @@ labels map value double␠precision help text +query TT +SELECT name, type FROM objects WHERE schema = 'mz_introspection' AND object = 'mz_cluster_replica_resource_usage' ORDER BY position +---- +process_id uint8 +source text +metric text +value uint8 + query TTT SELECT name, type, comment FROM objects WHERE schema = 'mz_introspection' AND object = 'mz_dataflows' ORDER BY position ---- @@ -271,6 +279,7 @@ mz_arrangement_sharing_raw mz_arrangement_sizes mz_arrangement_sizes_per_worker mz_cluster_prometheus_metrics +mz_cluster_replica_resource_usage mz_compute_dataflow_global_ids_per_worker mz_compute_error_counts mz_compute_error_counts_per_worker diff --git a/test/sqllogictest/catalog_server_explain.slt b/test/sqllogictest/catalog_server_explain.slt index 769ed420cb0ab..b06437fb06918 100644 --- a/test/sqllogictest/catalog_server_explain.slt +++ b/test/sqllogictest/catalog_server_explain.slt @@ -4986,7 +4986,7 @@ mz_catalog.mz_indexes: Filter: ("2" = (#1 ->> "object_type")) AND ("mz_introspection" = (#1 ->> "schema_name")) AND ((#1 ->> "object_name")) IS NOT NULL AND ("GidMapping" = #2) →Read mz_internal.mz_catalog_raw →Arrange (#0{log_name}) - →Constant (32 rows) + →Constant (33 rows) Source mz_internal.mz_catalog_raw project=(#0..=#2) @@ -5392,7 +5392,7 @@ mz_catalog.mz_sources: Project: #4, #0, #5, #1, #2, #6..=#12, #3, #13, #14 Map: null, null, null, null, null, null, "s1", null, null →Arrange (#1{schema_name}, #2{name}) - →Constant (56 rows) + →Constant (57 rows) →Arrange (#0{schema_name}) (#0{schema_name}, #1{name}) →Fused with Child Map/Filter/Project Project: #4, #3, #5 @@ -8236,7 +8236,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_builtin_sources"; ---- Explained Query (fast path): - →Constant (56 rows) + →Constant (57 rows) Target cluster: mz_catalog_server @@ -10238,7 +10238,7 @@ query T multiline EXPLAIN SELECT * FROM "mz_internal"."mz_ontology_entity_types"; ---- Explained Query (fast path): - →Constant (136 rows) + →Constant (137 rows) Target cluster: mz_catalog_server @@ -10262,7 +10262,7 @@ Explained Query: cte l0 = →Differential Join %1:mz_schemas[#0{id}] » %2:mz_objects[#2{schema_id}] » %0[#0{schema_name}, #1{table_name}] » %3:mz_columns[#0{id}] →Arrange (#0{schema_name}, #1{table_name}) - →Constant (136 rows) + →Constant (137 rows) →Arrange (#0{id}) →Fused with Child Map/Filter/Project Project: #1, #3 diff --git a/test/sqllogictest/cluster.slt b/test/sqllogictest/cluster.slt index 7d38047d85087..b793028eeabfd 100644 --- a/test/sqllogictest/cluster.slt +++ b/test/sqllogictest/cluster.slt @@ -201,6 +201,9 @@ bar mz_arrangement_sharing_raw mz_arrangement_sharing_raw_u7_primary_idx 2 w bar mz_cluster_prometheus_metrics mz_cluster_prometheus_metrics_u7_primary_idx 1 process_id NULL false bar mz_cluster_prometheus_metrics mz_cluster_prometheus_metrics_u7_primary_idx 2 metric_name NULL false bar mz_cluster_prometheus_metrics mz_cluster_prometheus_metrics_u7_primary_idx 3 labels NULL false +bar mz_cluster_replica_resource_usage mz_cluster_replica_resource_usage_u7_primary_idx 1 process_id NULL false +bar mz_cluster_replica_resource_usage mz_cluster_replica_resource_usage_u7_primary_idx 2 source NULL false +bar mz_cluster_replica_resource_usage mz_cluster_replica_resource_usage_u7_primary_idx 3 metric NULL false bar mz_compute_dataflow_global_ids_per_worker mz_compute_dataflow_global_ids_per_worker_u7_primary_idx 1 id NULL false bar mz_compute_dataflow_global_ids_per_worker mz_compute_dataflow_global_ids_per_worker_u7_primary_idx 2 worker_id NULL false bar mz_compute_dataflow_global_ids_per_worker mz_compute_dataflow_global_ids_per_worker_u7_primary_idx 3 global_id NULL false @@ -407,7 +410,7 @@ DROP CLUSTER foo, foo2, foo3, foo4 CASCADE query I SELECT COUNT(name) FROM mz_indexes WHERE cluster_id = 'u1'; ---- -32 +33 query I SELECT COUNT(name) FROM mz_indexes WHERE cluster_id <> 'u1' AND cluster_id NOT LIKE 's%'; @@ -420,7 +423,7 @@ CREATE CLUSTER test REPLICAS (foo (SIZE 'scale=1,workers=1')); query I SELECT COUNT(name) FROM mz_indexes; ---- -306 +313 statement ok DROP CLUSTER test CASCADE @@ -428,7 +431,7 @@ DROP CLUSTER test CASCADE query T SELECT COUNT(name) FROM mz_indexes; ---- -274 +280 simple conn=mz_system,user=mz_system ALTER CLUSTER quickstart OWNER TO materialize diff --git a/test/sqllogictest/cockroach/srfs.slt b/test/sqllogictest/cockroach/srfs.slt index 629a18d555394..e8e5aa17e4007 100644 --- a/test/sqllogictest/cockroach/srfs.slt +++ b/test/sqllogictest/cockroach/srfs.slt @@ -1164,6 +1164,24 @@ mz_cluster_replica_history 7 mz_cluster_replica_metrics 1 mz_cluster_replica_metrics_history 1 mz_cluster_replica_name_history 2 +mz_cluster_replica_resource_usage 1 +mz_cluster_replica_resource_usage 1 +mz_cluster_replica_resource_usage 1 +mz_cluster_replica_resource_usage 1 +mz_cluster_replica_resource_usage 1 +mz_cluster_replica_resource_usage 1 +mz_cluster_replica_resource_usage 2 +mz_cluster_replica_resource_usage 2 +mz_cluster_replica_resource_usage 2 +mz_cluster_replica_resource_usage 2 +mz_cluster_replica_resource_usage 2 +mz_cluster_replica_resource_usage 2 +mz_cluster_replica_resource_usage 3 +mz_cluster_replica_resource_usage 3 +mz_cluster_replica_resource_usage 3 +mz_cluster_replica_resource_usage 3 +mz_cluster_replica_resource_usage 3 +mz_cluster_replica_resource_usage 3 mz_cluster_replica_size_internal 1 mz_cluster_replica_sizes 1 mz_cluster_replica_status_history 1 diff --git a/test/sqllogictest/distinct_arrangements.slt b/test/sqllogictest/distinct_arrangements.slt index 05d66f9d9a694..e3d7d108b4dcd 100644 --- a/test/sqllogictest/distinct_arrangements.slt +++ b/test/sqllogictest/distinct_arrangements.slt @@ -1124,6 +1124,7 @@ Arrange Differential(BatcherRecords) Arrange Differential(BatcherSize) Arrange Differential(Sharing) Arrange PrometheusMetrics +Arrange ResourceUsage Arrange Timely(Addresses) Arrange Timely(BatchesReceived) Arrange Timely(BatchesSent) diff --git a/test/sqllogictest/information_schema_tables.slt b/test/sqllogictest/information_schema_tables.slt index 2081606984765..4d0d5d7ac2a25 100644 --- a/test/sqllogictest/information_schema_tables.slt +++ b/test/sqllogictest/information_schema_tables.slt @@ -945,6 +945,10 @@ mz_cluster_prometheus_metrics SOURCE materialize mz_introspection +mz_cluster_replica_resource_usage +SOURCE +materialize +mz_introspection mz_compute_dataflow_global_ids_per_worker SOURCE materialize diff --git a/test/sqllogictest/introspection/relations.slt b/test/sqllogictest/introspection/relations.slt index b8acec4390bb0..5a36a7cee42a8 100644 --- a/test/sqllogictest/introspection/relations.slt +++ b/test/sqllogictest/introspection/relations.slt @@ -136,6 +136,7 @@ Arrange␠Differential(BatcherRecords) ArrangementSize alloc::vec::Vec)>>>> Arrange␠Differential(Sharing) ArrangementSize alloc::vec::Vec)>>>> Arrange␠PrometheusMetrics ArrangementSize alloc::vec::Vec)>>>> +Arrange␠ResourceUsage ArrangementSize alloc::vec::Vec)>>>> Arrange␠Timely(Addresses) ArrangementSize alloc::vec::Vec)>>>> Arrange␠Timely(BatchesReceived) ArrangementSize alloc::vec::Vec)>>>> Arrange␠Timely(BatchesSent) ArrangementSize alloc::vec::Vec)>>>> @@ -192,6 +193,7 @@ Replay␠differential␠logs Differential␠Logging␠Demux alloc::vec::Vec<(c Replay␠reachability␠logs FlatMapReachability mz_timely_util::columnar::Column<(core::time::Duration,␠(usize,␠alloc::vec::Vec<(usize,␠usize,␠bool,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)>))> Replay␠storage␠timely␠logs Concatenate alloc::vec::Vec<(core::time::Duration,␠timely::logging::TimelyEvent)> Replay␠timely␠logs Concatenate alloc::vec::Vec<(core::time::Duration,␠timely::logging::TimelyEvent)> +ResourceUsage Arrange␠ResourceUsage mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> Timely␠Logging␠Demux Consolidate␠Timely(Addresses) alloc::vec::Vec<((usize,␠alloc::vec::Vec),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> Timely␠Logging␠Demux Consolidate␠Timely(BatchesReceived) alloc::vec::Vec<((mz_compute::logging::timely::MessageDatum,␠()),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> Timely␠Logging␠Demux Consolidate␠Timely(BatchesSent) alloc::vec::Vec<((mz_compute::logging::timely::MessageDatum,␠()),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> @@ -245,8 +247,8 @@ GROUP BY type; 1 mz_timely_util::columnar::Column<(core::time::Duration,␠(usize,␠alloc::vec::Vec<(usize,␠usize,␠bool,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)>))> 1 mz_timely_util::columnar::Column<(core::time::Duration,␠mz_compute::logging::compute::ComputeEvent)> 3 alloc::vec::Vec<(core::time::Duration,␠timely::logging::TimelyEvent)> -32 alloc::vec::Vec)>>>> -32 mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> +33 alloc::vec::Vec)>>>> +33 mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> 4 alloc::vec::Vec<((mz_compute::logging::timely::MessageDatum,␠()),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> 4 alloc::vec::Vec)>>> 8 alloc::vec::Vec<((usize,␠()),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> diff --git a/test/sqllogictest/mz_catalog_server_index_accounting.slt b/test/sqllogictest/mz_catalog_server_index_accounting.slt index 1cc1103fdf6bc..e8c638c152d60 100644 --- a/test/sqllogictest/mz_catalog_server_index_accounting.slt +++ b/test/sqllogictest/mz_catalog_server_index_accounting.slt @@ -37,109 +37,110 @@ mz_arrangement_heap_capacity_raw_s2_primary_idx CREATE␠INDEX␠"mz_arrangemen mz_arrangement_heap_size_raw_s2_primary_idx CREATE␠INDEX␠"mz_arrangement_heap_size_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_arrangement_heap_size_raw"␠("operator_id",␠"worker_id") mz_arrangement_records_raw_s2_primary_idx CREATE␠INDEX␠"mz_arrangement_records_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_arrangement_records_raw"␠("operator_id",␠"worker_id") mz_arrangement_sharing_raw_s2_primary_idx CREATE␠INDEX␠"mz_arrangement_sharing_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_arrangement_sharing_raw"␠("operator_id",␠"worker_id") -mz_cluster_auto_scaling_strategies_ind CREATE␠INDEX␠"mz_cluster_auto_scaling_strategies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s520␠AS␠"mz_internal"."mz_cluster_auto_scaling_strategies"]␠("cluster_id") -mz_cluster_deployment_lineage_ind CREATE␠INDEX␠"mz_cluster_deployment_lineage_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s769␠AS␠"mz_internal"."mz_cluster_deployment_lineage"]␠("cluster_id") +mz_cluster_auto_scaling_strategies_ind CREATE␠INDEX␠"mz_cluster_auto_scaling_strategies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s521␠AS␠"mz_internal"."mz_cluster_auto_scaling_strategies"]␠("cluster_id") +mz_cluster_deployment_lineage_ind CREATE␠INDEX␠"mz_cluster_deployment_lineage_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s770␠AS␠"mz_internal"."mz_cluster_deployment_lineage"]␠("cluster_id") mz_cluster_prometheus_metrics_s2_primary_idx CREATE␠INDEX␠"mz_cluster_prometheus_metrics_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_cluster_prometheus_metrics"␠("process_id",␠"metric_name",␠"labels") -mz_cluster_reconfigurations_ind CREATE␠INDEX␠"mz_cluster_reconfigurations_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s519␠AS␠"mz_internal"."mz_cluster_reconfigurations"]␠("cluster_id") -mz_cluster_replica_frontiers_ind CREATE␠INDEX␠"mz_cluster_replica_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s763␠AS␠"mz_catalog"."mz_cluster_replica_frontiers"]␠("object_id") -mz_cluster_replica_history_ind CREATE␠INDEX␠"mz_cluster_replica_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s618␠AS␠"mz_internal"."mz_cluster_replica_history"]␠("dropped_at") -mz_cluster_replica_metrics_history_ind CREATE␠INDEX␠"mz_cluster_replica_metrics_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s525␠AS␠"mz_internal"."mz_cluster_replica_metrics_history"]␠("replica_id") -mz_cluster_replica_metrics_ind CREATE␠INDEX␠"mz_cluster_replica_metrics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s526␠AS␠"mz_internal"."mz_cluster_replica_metrics"]␠("replica_id") -mz_cluster_replica_name_history_ind CREATE␠INDEX␠"mz_cluster_replica_name_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s619␠AS␠"mz_internal"."mz_cluster_replica_name_history"]␠("id") -mz_cluster_replica_size_internal_ind CREATE␠INDEX␠"mz_cluster_replica_size_internal_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s515␠AS␠"mz_internal"."mz_cluster_replica_size_internal"]␠("size") -mz_cluster_replica_sizes_ind CREATE␠INDEX␠"mz_cluster_replica_sizes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s514␠AS␠"mz_catalog"."mz_cluster_replica_sizes"]␠("size") -mz_cluster_replica_status_history_ind CREATE␠INDEX␠"mz_cluster_replica_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s527␠AS␠"mz_internal"."mz_cluster_replica_status_history"]␠("replica_id") -mz_cluster_replica_statuses_ind CREATE␠INDEX␠"mz_cluster_replica_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s528␠AS␠"mz_internal"."mz_cluster_replica_statuses"]␠("replica_id") -mz_cluster_replicas_ind CREATE␠INDEX␠"mz_cluster_replicas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s524␠AS␠"mz_catalog"."mz_cluster_replicas"]␠("id") -mz_clusters_ind CREATE␠INDEX␠"mz_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s516␠AS␠"mz_catalog"."mz_clusters"]␠("id") -mz_columns_ind CREATE␠INDEX␠"mz_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s487␠AS␠"mz_catalog"."mz_columns"]␠("name") -mz_comments_ind CREATE␠INDEX␠"mz_comments_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s543␠AS␠"mz_internal"."mz_comments"]␠("id") +mz_cluster_reconfigurations_ind CREATE␠INDEX␠"mz_cluster_reconfigurations_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s520␠AS␠"mz_internal"."mz_cluster_reconfigurations"]␠("cluster_id") +mz_cluster_replica_frontiers_ind CREATE␠INDEX␠"mz_cluster_replica_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s764␠AS␠"mz_catalog"."mz_cluster_replica_frontiers"]␠("object_id") +mz_cluster_replica_history_ind CREATE␠INDEX␠"mz_cluster_replica_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s619␠AS␠"mz_internal"."mz_cluster_replica_history"]␠("dropped_at") +mz_cluster_replica_metrics_history_ind CREATE␠INDEX␠"mz_cluster_replica_metrics_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s526␠AS␠"mz_internal"."mz_cluster_replica_metrics_history"]␠("replica_id") +mz_cluster_replica_metrics_ind CREATE␠INDEX␠"mz_cluster_replica_metrics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s527␠AS␠"mz_internal"."mz_cluster_replica_metrics"]␠("replica_id") +mz_cluster_replica_name_history_ind CREATE␠INDEX␠"mz_cluster_replica_name_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s620␠AS␠"mz_internal"."mz_cluster_replica_name_history"]␠("id") +mz_cluster_replica_resource_usage_s2_primary_idx CREATE␠INDEX␠"mz_cluster_replica_resource_usage_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_cluster_replica_resource_usage"␠("process_id",␠"source",␠"metric") +mz_cluster_replica_size_internal_ind CREATE␠INDEX␠"mz_cluster_replica_size_internal_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s516␠AS␠"mz_internal"."mz_cluster_replica_size_internal"]␠("size") +mz_cluster_replica_sizes_ind CREATE␠INDEX␠"mz_cluster_replica_sizes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s515␠AS␠"mz_catalog"."mz_cluster_replica_sizes"]␠("size") +mz_cluster_replica_status_history_ind CREATE␠INDEX␠"mz_cluster_replica_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s528␠AS␠"mz_internal"."mz_cluster_replica_status_history"]␠("replica_id") +mz_cluster_replica_statuses_ind CREATE␠INDEX␠"mz_cluster_replica_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s529␠AS␠"mz_internal"."mz_cluster_replica_statuses"]␠("replica_id") +mz_cluster_replicas_ind CREATE␠INDEX␠"mz_cluster_replicas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s525␠AS␠"mz_catalog"."mz_cluster_replicas"]␠("id") +mz_clusters_ind CREATE␠INDEX␠"mz_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s517␠AS␠"mz_catalog"."mz_clusters"]␠("id") +mz_columns_ind CREATE␠INDEX␠"mz_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s488␠AS␠"mz_catalog"."mz_columns"]␠("name") +mz_comments_ind CREATE␠INDEX␠"mz_comments_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s544␠AS␠"mz_internal"."mz_comments"]␠("id") mz_compute_dataflow_global_ids_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_dataflow_global_ids_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_dataflow_global_ids_per_worker"␠("id",␠"worker_id",␠"global_id") -mz_compute_dependencies_ind CREATE␠INDEX␠"mz_compute_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s744␠AS␠"mz_internal"."mz_compute_dependencies"]␠("dependency_id") +mz_compute_dependencies_ind CREATE␠INDEX␠"mz_compute_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s745␠AS␠"mz_internal"."mz_compute_dependencies"]␠("dependency_id") mz_compute_error_counts_raw_s2_primary_idx CREATE␠INDEX␠"mz_compute_error_counts_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_error_counts_raw"␠("export_id",␠"worker_id") mz_compute_exports_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_exports_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_exports_per_worker"␠("export_id",␠"worker_id") mz_compute_frontiers_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_frontiers_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_frontiers_per_worker"␠("export_id",␠"worker_id") -mz_compute_hydration_times_ind CREATE␠INDEX␠"mz_compute_hydration_times_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s754␠AS␠"mz_internal"."mz_compute_hydration_times"]␠("replica_id") +mz_compute_hydration_times_ind CREATE␠INDEX␠"mz_compute_hydration_times_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s755␠AS␠"mz_internal"."mz_compute_hydration_times"]␠("replica_id") mz_compute_hydration_times_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_hydration_times_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_hydration_times_per_worker"␠("export_id",␠"worker_id") mz_compute_import_frontiers_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_import_frontiers_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_import_frontiers_per_worker"␠("export_id",␠"import_id",␠"worker_id") mz_compute_lir_mapping_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_lir_mapping_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_lir_mapping_per_worker"␠("global_id",␠"lir_id",␠"worker_id") mz_compute_operator_durations_histogram_raw_s2_primary_idx CREATE␠INDEX␠"mz_compute_operator_durations_histogram_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_operator_durations_histogram_raw"␠("id",␠"worker_id",␠"duration_ns") mz_compute_operator_hydration_statuses_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_compute_operator_hydration_statuses_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_compute_operator_hydration_statuses_per_worker"␠("export_id",␠"lir_id",␠"worker_id") -mz_connections_ind CREATE␠INDEX␠"mz_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s522␠AS␠"mz_catalog"."mz_connections"]␠("schema_id") -mz_console_cluster_utilization_overview_24h_ind CREATE␠INDEX␠"mz_console_cluster_utilization_overview_24h_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s750␠AS␠"mz_internal"."mz_console_cluster_utilization_overview_24h"]␠("cluster_id") -mz_console_cluster_utilization_overview_3h_ind CREATE␠INDEX␠"mz_console_cluster_utilization_overview_3h_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s749␠AS␠"mz_internal"."mz_console_cluster_utilization_overview_3h"]␠("cluster_id") -mz_console_cluster_utilization_overview_ind CREATE␠INDEX␠"mz_console_cluster_utilization_overview_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s748␠AS␠"mz_internal"."mz_console_cluster_utilization_overview"]␠("cluster_id") -mz_databases_ind CREATE␠INDEX␠"mz_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s485␠AS␠"mz_catalog"."mz_databases"]␠("name") +mz_connections_ind CREATE␠INDEX␠"mz_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s523␠AS␠"mz_catalog"."mz_connections"]␠("schema_id") +mz_console_cluster_utilization_overview_24h_ind CREATE␠INDEX␠"mz_console_cluster_utilization_overview_24h_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s751␠AS␠"mz_internal"."mz_console_cluster_utilization_overview_24h"]␠("cluster_id") +mz_console_cluster_utilization_overview_3h_ind CREATE␠INDEX␠"mz_console_cluster_utilization_overview_3h_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s750␠AS␠"mz_internal"."mz_console_cluster_utilization_overview_3h"]␠("cluster_id") +mz_console_cluster_utilization_overview_ind CREATE␠INDEX␠"mz_console_cluster_utilization_overview_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s749␠AS␠"mz_internal"."mz_console_cluster_utilization_overview"]␠("cluster_id") +mz_databases_ind CREATE␠INDEX␠"mz_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s486␠AS␠"mz_catalog"."mz_databases"]␠("name") mz_dataflow_addresses_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_dataflow_addresses_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_dataflow_addresses_per_worker"␠("id",␠"worker_id") mz_dataflow_channels_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_dataflow_channels_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_dataflow_channels_per_worker"␠("id",␠"worker_id") mz_dataflow_operator_reachability_raw_s2_primary_idx CREATE␠INDEX␠"mz_dataflow_operator_reachability_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_dataflow_operator_reachability_raw"␠("id",␠"worker_id",␠"source",␠"port",␠"update_type",␠"time") mz_dataflow_operators_per_worker_s2_primary_idx CREATE␠INDEX␠"mz_dataflow_operators_per_worker_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_dataflow_operators_per_worker"␠("id",␠"worker_id") -mz_frontiers_ind CREATE␠INDEX␠"mz_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s735␠AS␠"mz_internal"."mz_frontiers"]␠("object_id") -mz_hydration_statuses_ind CREATE␠INDEX␠"mz_hydration_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s765␠AS␠"mz_internal"."mz_hydration_statuses"]␠("object_id",␠"replica_id") -mz_indexes_ind CREATE␠INDEX␠"mz_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s488␠AS␠"mz_catalog"."mz_indexes"]␠("id") -mz_kafka_sources_ind CREATE␠INDEX␠"mz_kafka_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s482␠AS␠"mz_catalog"."mz_kafka_sources"]␠("id") -mz_materialized_views_ind CREATE␠INDEX␠"mz_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s547␠AS␠"mz_catalog"."mz_materialized_views"]␠("id") +mz_frontiers_ind CREATE␠INDEX␠"mz_frontiers_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s736␠AS␠"mz_internal"."mz_frontiers"]␠("object_id") +mz_hydration_statuses_ind CREATE␠INDEX␠"mz_hydration_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s766␠AS␠"mz_internal"."mz_hydration_statuses"]␠("object_id",␠"replica_id") +mz_indexes_ind CREATE␠INDEX␠"mz_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s489␠AS␠"mz_catalog"."mz_indexes"]␠("id") +mz_kafka_sources_ind CREATE␠INDEX␠"mz_kafka_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s483␠AS␠"mz_catalog"."mz_kafka_sources"]␠("id") +mz_materialized_views_ind CREATE␠INDEX␠"mz_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s548␠AS␠"mz_catalog"."mz_materialized_views"]␠("id") mz_message_batch_counts_received_raw_s2_primary_idx CREATE␠INDEX␠"mz_message_batch_counts_received_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_batch_counts_received_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id") mz_message_batch_counts_sent_raw_s2_primary_idx CREATE␠INDEX␠"mz_message_batch_counts_sent_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_batch_counts_sent_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id") mz_message_counts_received_raw_s2_primary_idx CREATE␠INDEX␠"mz_message_counts_received_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_counts_received_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id") mz_message_counts_sent_raw_s2_primary_idx CREATE␠INDEX␠"mz_message_counts_sent_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_message_counts_sent_raw"␠("channel_id",␠"from_worker_id",␠"to_worker_id") -mz_metric_sinks_ind CREATE␠INDEX␠"mz_metric_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s545␠AS␠"mz_internal"."mz_metric_sinks"]␠("id") -mz_notices_ind CREATE␠INDEX␠"mz_notices_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s842␠AS␠"mz_internal"."mz_notices"]␠("id") -mz_object_arrangement_size_history_object_ind CREATE␠INDEX␠"mz_object_arrangement_size_history_object_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s757␠AS␠"mz_internal"."mz_object_arrangement_size_history"]␠("object_id") -mz_object_arrangement_size_history_ts_ind CREATE␠INDEX␠"mz_object_arrangement_size_history_ts_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s757␠AS␠"mz_internal"."mz_object_arrangement_size_history"]␠("collection_timestamp") -mz_object_arrangement_sizes_ind CREATE␠INDEX␠"mz_object_arrangement_sizes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s755␠AS␠"mz_internal"."mz_object_arrangement_sizes"]␠("replica_id") -mz_object_dependencies_ind CREATE␠INDEX␠"mz_object_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s483␠AS␠"mz_internal"."mz_object_dependencies"]␠("object_id") -mz_object_graph_edges_ind CREATE␠INDEX␠"mz_object_graph_edges_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s746␠AS␠"mz_internal"."mz_object_graph_edges"]␠("object_id") -mz_object_history_ind CREATE␠INDEX␠"mz_object_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s558␠AS␠"mz_internal"."mz_object_history"]␠("id") -mz_object_lifetimes_ind CREATE␠INDEX␠"mz_object_lifetimes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s559␠AS␠"mz_internal"."mz_object_lifetimes"]␠("id") -mz_object_transitive_dependencies_ind CREATE␠INDEX␠"mz_object_transitive_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s575␠AS␠"mz_internal"."mz_object_transitive_dependencies"]␠("object_id") -mz_objects_ind CREATE␠INDEX␠"mz_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s555␠AS␠"mz_catalog"."mz_objects"]␠("schema_id") +mz_metric_sinks_ind CREATE␠INDEX␠"mz_metric_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s546␠AS␠"mz_internal"."mz_metric_sinks"]␠("id") +mz_notices_ind CREATE␠INDEX␠"mz_notices_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s843␠AS␠"mz_internal"."mz_notices"]␠("id") +mz_object_arrangement_size_history_object_ind CREATE␠INDEX␠"mz_object_arrangement_size_history_object_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s758␠AS␠"mz_internal"."mz_object_arrangement_size_history"]␠("object_id") +mz_object_arrangement_size_history_ts_ind CREATE␠INDEX␠"mz_object_arrangement_size_history_ts_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s758␠AS␠"mz_internal"."mz_object_arrangement_size_history"]␠("collection_timestamp") +mz_object_arrangement_sizes_ind CREATE␠INDEX␠"mz_object_arrangement_sizes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s756␠AS␠"mz_internal"."mz_object_arrangement_sizes"]␠("replica_id") +mz_object_dependencies_ind CREATE␠INDEX␠"mz_object_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s484␠AS␠"mz_internal"."mz_object_dependencies"]␠("object_id") +mz_object_graph_edges_ind CREATE␠INDEX␠"mz_object_graph_edges_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s747␠AS␠"mz_internal"."mz_object_graph_edges"]␠("object_id") +mz_object_history_ind CREATE␠INDEX␠"mz_object_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s559␠AS␠"mz_internal"."mz_object_history"]␠("id") +mz_object_lifetimes_ind CREATE␠INDEX␠"mz_object_lifetimes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s560␠AS␠"mz_internal"."mz_object_lifetimes"]␠("id") +mz_object_transitive_dependencies_ind CREATE␠INDEX␠"mz_object_transitive_dependencies_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s576␠AS␠"mz_internal"."mz_object_transitive_dependencies"]␠("object_id") +mz_objects_ind CREATE␠INDEX␠"mz_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s556␠AS␠"mz_catalog"."mz_objects"]␠("schema_id") mz_peek_durations_histogram_raw_s2_primary_idx CREATE␠INDEX␠"mz_peek_durations_histogram_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_peek_durations_histogram_raw"␠("worker_id",␠"type",␠"duration_ns") -mz_recent_activity_log_thinned_ind CREATE␠INDEX␠"mz_recent_activity_log_thinned_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s719␠AS␠"mz_internal"."mz_recent_activity_log_thinned"]␠("sql_hash") -mz_recent_sql_text_ind CREATE␠INDEX␠"mz_recent_sql_text_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s715␠AS␠"mz_internal"."mz_recent_sql_text"]␠("sql_hash") -mz_recent_storage_usage_ind CREATE␠INDEX␠"mz_recent_storage_usage_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s834␠AS␠"mz_catalog"."mz_recent_storage_usage"]␠("object_id") -mz_roles_ind CREATE␠INDEX␠"mz_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s506␠AS␠"mz_catalog"."mz_roles"]␠("id") +mz_recent_activity_log_thinned_ind CREATE␠INDEX␠"mz_recent_activity_log_thinned_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s720␠AS␠"mz_internal"."mz_recent_activity_log_thinned"]␠("sql_hash") +mz_recent_sql_text_ind CREATE␠INDEX␠"mz_recent_sql_text_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s716␠AS␠"mz_internal"."mz_recent_sql_text"]␠("sql_hash") +mz_recent_storage_usage_ind CREATE␠INDEX␠"mz_recent_storage_usage_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s835␠AS␠"mz_catalog"."mz_recent_storage_usage"]␠("object_id") +mz_roles_ind CREATE␠INDEX␠"mz_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s507␠AS␠"mz_catalog"."mz_roles"]␠("id") mz_scheduling_elapsed_raw_s2_primary_idx CREATE␠INDEX␠"mz_scheduling_elapsed_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_scheduling_elapsed_raw"␠("id",␠"worker_id") mz_scheduling_parks_histogram_raw_s2_primary_idx CREATE␠INDEX␠"mz_scheduling_parks_histogram_raw_s2_primary_idx"␠IN␠CLUSTER␠[s2]␠ON␠"mz_introspection"."mz_scheduling_parks_histogram_raw"␠("worker_id",␠"slept_for_ns",␠"requested_ns") -mz_schemas_ind CREATE␠INDEX␠"mz_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s486␠AS␠"mz_catalog"."mz_schemas"]␠("database_id") -mz_secrets_ind CREATE␠INDEX␠"mz_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s521␠AS␠"mz_catalog"."mz_secrets"]␠("name") -mz_show_all_objects_ind CREATE␠INDEX␠"mz_show_all_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s603␠AS␠"mz_internal"."mz_show_all_objects"]␠("schema_id") -mz_show_cluster_replicas_ind CREATE␠INDEX␠"mz_show_cluster_replicas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s767␠AS␠"mz_internal"."mz_show_cluster_replicas"]␠("cluster") -mz_show_clusters_ind CREATE␠INDEX␠"mz_show_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s605␠AS␠"mz_internal"."mz_show_clusters"]␠("name") -mz_show_columns_ind CREATE␠INDEX␠"mz_show_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s604␠AS␠"mz_internal"."mz_show_columns"]␠("id") -mz_show_connections_ind CREATE␠INDEX␠"mz_show_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s613␠AS␠"mz_internal"."mz_show_connections"]␠("schema_id") -mz_show_databases_ind CREATE␠INDEX␠"mz_show_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s607␠AS␠"mz_internal"."mz_show_databases"]␠("name") -mz_show_indexes_ind CREATE␠INDEX␠"mz_show_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s617␠AS␠"mz_internal"."mz_show_indexes"]␠("schema_id") -mz_show_materialized_views_ind CREATE␠INDEX␠"mz_show_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s616␠AS␠"mz_internal"."mz_show_materialized_views"]␠("schema_id") -mz_show_roles_ind CREATE␠INDEX␠"mz_show_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s612␠AS␠"mz_internal"."mz_show_roles"]␠("name") -mz_show_schemas_ind CREATE␠INDEX␠"mz_show_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s608␠AS␠"mz_internal"."mz_show_schemas"]␠("database_id") -mz_show_secrets_ind CREATE␠INDEX␠"mz_show_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s606␠AS␠"mz_internal"."mz_show_secrets"]␠("schema_id") -mz_show_sinks_ind CREATE␠INDEX␠"mz_show_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s615␠AS␠"mz_internal"."mz_show_sinks"]␠("schema_id") -mz_show_sources_ind CREATE␠INDEX␠"mz_show_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s614␠AS␠"mz_internal"."mz_show_sources"]␠("schema_id") -mz_show_tables_ind CREATE␠INDEX␠"mz_show_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s609␠AS␠"mz_internal"."mz_show_tables"]␠("schema_id") -mz_show_types_ind CREATE␠INDEX␠"mz_show_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s611␠AS␠"mz_internal"."mz_show_types"]␠("schema_id") -mz_show_views_ind CREATE␠INDEX␠"mz_show_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s610␠AS␠"mz_internal"."mz_show_views"]␠("schema_id") -mz_sink_statistics_ind CREATE␠INDEX␠"mz_sink_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s732␠AS␠"mz_internal"."mz_sink_statistics"]␠("id",␠"replica_id") -mz_sink_status_history_ind CREATE␠INDEX␠"mz_sink_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s704␠AS␠"mz_internal"."mz_sink_status_history"]␠("sink_id") -mz_sink_statuses_ind CREATE␠INDEX␠"mz_sink_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s705␠AS␠"mz_internal"."mz_sink_statuses"]␠("id") -mz_sinks_ind CREATE␠INDEX␠"mz_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s498␠AS␠"mz_catalog"."mz_sinks"]␠("id") -mz_source_statistics_ind CREATE␠INDEX␠"mz_source_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s730␠AS␠"mz_internal"."mz_source_statistics"]␠("id",␠"replica_id") -mz_source_statistics_with_history_ind CREATE␠INDEX␠"mz_source_statistics_with_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s728␠AS␠"mz_internal"."mz_source_statistics_with_history"]␠("id",␠"replica_id") -mz_source_status_history_ind CREATE␠INDEX␠"mz_source_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s706␠AS␠"mz_internal"."mz_source_status_history"]␠("source_id") -mz_source_statuses_ind CREATE␠INDEX␠"mz_source_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s723␠AS␠"mz_internal"."mz_source_statuses"]␠("id") -mz_sources_ind CREATE␠INDEX␠"mz_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s491␠AS␠"mz_catalog"."mz_sources"]␠("id") -mz_tables_ind CREATE␠INDEX␠"mz_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s490␠AS␠"mz_catalog"."mz_tables"]␠("schema_id") -mz_types_ind CREATE␠INDEX␠"mz_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s500␠AS␠"mz_catalog"."mz_types"]␠("schema_id") -mz_views_ind CREATE␠INDEX␠"mz_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s499␠AS␠"mz_catalog"."mz_views"]␠("schema_id") -mz_wallclock_global_lag_recent_history_ind CREATE␠INDEX␠"mz_wallclock_global_lag_recent_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s739␠AS␠"mz_internal"."mz_wallclock_global_lag_recent_history"]␠("object_id") -mz_webhook_sources_ind CREATE␠INDEX␠"mz_webhook_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s544␠AS␠"mz_internal"."mz_webhook_sources"]␠("id") -pg_attrdef_all_databases_ind CREATE␠INDEX␠"pg_attrdef_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s644␠AS␠"mz_internal"."pg_attrdef_all_databases"]␠("oid",␠"adrelid",␠"adnum",␠"adbin",␠"adsrc") -pg_attribute_all_databases_ind CREATE␠INDEX␠"pg_attribute_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s637␠AS␠"mz_internal"."pg_attribute_all_databases"]␠("attrelid",␠"attname",␠"atttypid",␠"attlen",␠"attnum",␠"atttypmod",␠"attnotnull",␠"atthasdef",␠"attidentity",␠"attgenerated",␠"attisdropped",␠"attcollation",␠"database_name",␠"pg_type_database_name") -pg_authid_core_ind CREATE␠INDEX␠"pg_authid_core_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s654␠AS␠"mz_internal"."pg_authid_core"]␠("rolname") -pg_class_all_databases_ind CREATE␠INDEX␠"pg_class_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s625␠AS␠"mz_internal"."pg_class_all_databases"]␠("relname") -pg_description_all_databases_ind CREATE␠INDEX␠"pg_description_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s634␠AS␠"mz_internal"."pg_description_all_databases"]␠("objoid",␠"classoid",␠"objsubid",␠"description",␠"oid_database_name",␠"class_database_name") -pg_namespace_all_databases_ind CREATE␠INDEX␠"pg_namespace_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s622␠AS␠"mz_internal"."pg_namespace_all_databases"]␠("nspname") -pg_type_all_databases_ind CREATE␠INDEX␠"pg_type_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s631␠AS␠"mz_internal"."pg_type_all_databases"]␠("oid") +mz_schemas_ind CREATE␠INDEX␠"mz_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s487␠AS␠"mz_catalog"."mz_schemas"]␠("database_id") +mz_secrets_ind CREATE␠INDEX␠"mz_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s522␠AS␠"mz_catalog"."mz_secrets"]␠("name") +mz_show_all_objects_ind CREATE␠INDEX␠"mz_show_all_objects_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s604␠AS␠"mz_internal"."mz_show_all_objects"]␠("schema_id") +mz_show_cluster_replicas_ind CREATE␠INDEX␠"mz_show_cluster_replicas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s768␠AS␠"mz_internal"."mz_show_cluster_replicas"]␠("cluster") +mz_show_clusters_ind CREATE␠INDEX␠"mz_show_clusters_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s606␠AS␠"mz_internal"."mz_show_clusters"]␠("name") +mz_show_columns_ind CREATE␠INDEX␠"mz_show_columns_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s605␠AS␠"mz_internal"."mz_show_columns"]␠("id") +mz_show_connections_ind CREATE␠INDEX␠"mz_show_connections_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s614␠AS␠"mz_internal"."mz_show_connections"]␠("schema_id") +mz_show_databases_ind CREATE␠INDEX␠"mz_show_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s608␠AS␠"mz_internal"."mz_show_databases"]␠("name") +mz_show_indexes_ind CREATE␠INDEX␠"mz_show_indexes_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s618␠AS␠"mz_internal"."mz_show_indexes"]␠("schema_id") +mz_show_materialized_views_ind CREATE␠INDEX␠"mz_show_materialized_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s617␠AS␠"mz_internal"."mz_show_materialized_views"]␠("schema_id") +mz_show_roles_ind CREATE␠INDEX␠"mz_show_roles_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s613␠AS␠"mz_internal"."mz_show_roles"]␠("name") +mz_show_schemas_ind CREATE␠INDEX␠"mz_show_schemas_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s609␠AS␠"mz_internal"."mz_show_schemas"]␠("database_id") +mz_show_secrets_ind CREATE␠INDEX␠"mz_show_secrets_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s607␠AS␠"mz_internal"."mz_show_secrets"]␠("schema_id") +mz_show_sinks_ind CREATE␠INDEX␠"mz_show_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s616␠AS␠"mz_internal"."mz_show_sinks"]␠("schema_id") +mz_show_sources_ind CREATE␠INDEX␠"mz_show_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s615␠AS␠"mz_internal"."mz_show_sources"]␠("schema_id") +mz_show_tables_ind CREATE␠INDEX␠"mz_show_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s610␠AS␠"mz_internal"."mz_show_tables"]␠("schema_id") +mz_show_types_ind CREATE␠INDEX␠"mz_show_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s612␠AS␠"mz_internal"."mz_show_types"]␠("schema_id") +mz_show_views_ind CREATE␠INDEX␠"mz_show_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s611␠AS␠"mz_internal"."mz_show_views"]␠("schema_id") +mz_sink_statistics_ind CREATE␠INDEX␠"mz_sink_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s733␠AS␠"mz_internal"."mz_sink_statistics"]␠("id",␠"replica_id") +mz_sink_status_history_ind CREATE␠INDEX␠"mz_sink_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s705␠AS␠"mz_internal"."mz_sink_status_history"]␠("sink_id") +mz_sink_statuses_ind CREATE␠INDEX␠"mz_sink_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s706␠AS␠"mz_internal"."mz_sink_statuses"]␠("id") +mz_sinks_ind CREATE␠INDEX␠"mz_sinks_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s499␠AS␠"mz_catalog"."mz_sinks"]␠("id") +mz_source_statistics_ind CREATE␠INDEX␠"mz_source_statistics_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s731␠AS␠"mz_internal"."mz_source_statistics"]␠("id",␠"replica_id") +mz_source_statistics_with_history_ind CREATE␠INDEX␠"mz_source_statistics_with_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s729␠AS␠"mz_internal"."mz_source_statistics_with_history"]␠("id",␠"replica_id") +mz_source_status_history_ind CREATE␠INDEX␠"mz_source_status_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s707␠AS␠"mz_internal"."mz_source_status_history"]␠("source_id") +mz_source_statuses_ind CREATE␠INDEX␠"mz_source_statuses_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s724␠AS␠"mz_internal"."mz_source_statuses"]␠("id") +mz_sources_ind CREATE␠INDEX␠"mz_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s492␠AS␠"mz_catalog"."mz_sources"]␠("id") +mz_tables_ind CREATE␠INDEX␠"mz_tables_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s491␠AS␠"mz_catalog"."mz_tables"]␠("schema_id") +mz_types_ind CREATE␠INDEX␠"mz_types_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s501␠AS␠"mz_catalog"."mz_types"]␠("schema_id") +mz_views_ind CREATE␠INDEX␠"mz_views_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s500␠AS␠"mz_catalog"."mz_views"]␠("schema_id") +mz_wallclock_global_lag_recent_history_ind CREATE␠INDEX␠"mz_wallclock_global_lag_recent_history_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s740␠AS␠"mz_internal"."mz_wallclock_global_lag_recent_history"]␠("object_id") +mz_webhook_sources_ind CREATE␠INDEX␠"mz_webhook_sources_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s545␠AS␠"mz_internal"."mz_webhook_sources"]␠("id") +pg_attrdef_all_databases_ind CREATE␠INDEX␠"pg_attrdef_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s645␠AS␠"mz_internal"."pg_attrdef_all_databases"]␠("oid",␠"adrelid",␠"adnum",␠"adbin",␠"adsrc") +pg_attribute_all_databases_ind CREATE␠INDEX␠"pg_attribute_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s638␠AS␠"mz_internal"."pg_attribute_all_databases"]␠("attrelid",␠"attname",␠"atttypid",␠"attlen",␠"attnum",␠"atttypmod",␠"attnotnull",␠"atthasdef",␠"attidentity",␠"attgenerated",␠"attisdropped",␠"attcollation",␠"database_name",␠"pg_type_database_name") +pg_authid_core_ind CREATE␠INDEX␠"pg_authid_core_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s655␠AS␠"mz_internal"."pg_authid_core"]␠("rolname") +pg_class_all_databases_ind CREATE␠INDEX␠"pg_class_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s626␠AS␠"mz_internal"."pg_class_all_databases"]␠("relname") +pg_description_all_databases_ind CREATE␠INDEX␠"pg_description_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s635␠AS␠"mz_internal"."pg_description_all_databases"]␠("objoid",␠"classoid",␠"objsubid",␠"description",␠"oid_database_name",␠"class_database_name") +pg_namespace_all_databases_ind CREATE␠INDEX␠"pg_namespace_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s623␠AS␠"mz_internal"."pg_namespace_all_databases"]␠("nspname") +pg_type_all_databases_ind CREATE␠INDEX␠"pg_type_all_databases_ind"␠IN␠CLUSTER␠[s2]␠ON␠[s632␠AS␠"mz_internal"."pg_type_all_databases"]␠("oid") # Record all transitive dependencies (tables, sources, views, mvs) of indexes on # the mz_catalog_server cluster. @@ -285,6 +286,10 @@ mz_cluster_replica_name_history id mz_cluster_replica_name_history new_name mz_cluster_replica_name_history occurred_at mz_cluster_replica_name_history previous_name +mz_cluster_replica_resource_usage metric +mz_cluster_replica_resource_usage process_id +mz_cluster_replica_resource_usage source +mz_cluster_replica_resource_usage value mz_cluster_replica_size_internal disk_bytes mz_cluster_replica_size_internal size mz_cluster_replica_size_internal swap_enabled diff --git a/test/sqllogictest/oid.slt b/test/sqllogictest/oid.slt index f7a72291710a0..ba84f1540219a 100644 --- a/test/sqllogictest/oid.slt +++ b/test/sqllogictest/oid.slt @@ -1252,3 +1252,4 @@ SELECT oid, name FROM mz_objects WHERE id LIKE 's%' AND oid < 20000 ORDER BY oid 17120 mz_metric_sinks 17121 mz_metric_sinks_ind 17122 mz_object_hydration_history +17123 mz_cluster_replica_resource_usage diff --git a/test/sqllogictest/pg_catalog_user.slt b/test/sqllogictest/pg_catalog_user.slt index d892d2d48689a..b09c3b5f32729 100644 --- a/test/sqllogictest/pg_catalog_user.slt +++ b/test/sqllogictest/pg_catalog_user.slt @@ -27,7 +27,7 @@ CREATE ROLE "materialize@foocorp.io" WITH LOGIN query TIBBBBTTT rowsort SELECT usename, usesysid, usecreatedb, usesuper, userepl, usebypassrls, passwd, valuntil, useconfig FROM pg_user; ---- -materialize@foocorp.io 20196 false NULL false false ******** NULL NULL +materialize@foocorp.io 20202 false NULL false false ******** NULL NULL mz_support 16662 false true false false ******** NULL NULL mz_system 16661 true true false false ******** NULL NULL diff --git a/test/sqllogictest/regclass.slt b/test/sqllogictest/regclass.slt index d3816146ae8ea..7257fab6c0108 100644 --- a/test/sqllogictest/regclass.slt +++ b/test/sqllogictest/regclass.slt @@ -35,12 +35,12 @@ CREATE MATERIALIZED VIEW s.m AS SELECT * FROM s.t; query T SELECT 't'::regclass::oid::int ---- -20196 +20202 query T SELECT 's.t'::regclass::oid::int ---- -20197 +20203 query T SELECT 't'::regclass = 's.t'::regclass @@ -73,7 +73,7 @@ t query T SELECT 't'::regclass::oid::int ---- -20197 +20203 query T SELECT 'public.t'::regclass::text; @@ -101,12 +101,12 @@ d.public.t query T SELECT 'm'::regclass::oid::int ---- -20201 +20207 query T SELECT 's.m'::regclass::oid::int ---- -20202 +20208 query T SELECT 'm'::regclass = 's.m'::regclass @@ -296,7 +296,7 @@ true query T SELECT 'materialize.public.t'::regclass::oid::int ---- -20196 +20202 query error relation "t" does not exist SELECT 't'::regclass::oid::int diff --git a/test/sqllogictest/regtype.slt b/test/sqllogictest/regtype.slt index 8ba6b5b88e8f3..98aa4a37e3671 100644 --- a/test/sqllogictest/regtype.slt +++ b/test/sqllogictest/regtype.slt @@ -142,12 +142,12 @@ CREATE TYPE d.public.t AS LIST (ELEMENT TYPE = int4); query T SELECT 't'::regtype::oid::int ---- -20197 +20203 query T SELECT 's.t'::regtype::oid::int ---- -20198 +20204 query T SELECT 't'::regtype = 's.t'::regtype diff --git a/test/testdrive/catalog.td b/test/testdrive/catalog.td index 3e4ad1606a6a3..d783af6de9cce 100644 --- a/test/testdrive/catalog.td +++ b/test/testdrive/catalog.td @@ -763,6 +763,7 @@ mz_compute_import_frontiers_per_worker log "" mz_compute_lir_mapping_per_worker log "" mz_compute_operator_durations_histogram_raw log "" mz_compute_operator_hydration_statuses_per_worker log "" +mz_cluster_replica_resource_usage log "" mz_cluster_prometheus_metrics log "" mz_dataflow_addresses_per_worker log "" mz_dataflow_channels_per_worker log "" @@ -839,7 +840,7 @@ test_table "" # There is one entry in mz_indexes for each field_number/expression of the index. > SELECT COUNT(id) FROM mz_indexes WHERE id LIKE 's%' -274 +280 # Create a second schema with the same table name as above > CREATE SCHEMA tester2 diff --git a/test/testdrive/indexes.td b/test/testdrive/indexes.td index 6368143730dbf..ab3beb5266589 100644 --- a/test/testdrive/indexes.td +++ b/test/testdrive/indexes.td @@ -322,6 +322,7 @@ mz_compute_import_frontiers_per_worker_s2_primary_idx mz_compute_import_fr mz_compute_lir_mapping_per_worker_s2_primary_idx mz_compute_lir_mapping_per_worker mz_catalog_server {global_id,lir_id,worker_id} "" mz_compute_operator_durations_histogram_raw_s2_primary_idx mz_compute_operator_durations_histogram_raw mz_catalog_server {id,worker_id,duration_ns} "" mz_compute_operator_hydration_statuses_per_worker_s2_primary_idx mz_compute_operator_hydration_statuses_per_worker mz_catalog_server {export_id,lir_id,worker_id} "" +mz_cluster_replica_resource_usage_s2_primary_idx mz_cluster_replica_resource_usage mz_catalog_server {process_id,source,metric} "" mz_cluster_prometheus_metrics_s2_primary_idx mz_cluster_prometheus_metrics mz_catalog_server {process_id,metric_name,labels} "" mz_connections_ind mz_connections mz_catalog_server {schema_id} "" mz_console_cluster_utilization_overview_24h_ind mz_console_cluster_utilization_overview_24h mz_catalog_server {cluster_id} "" diff --git a/test/workload-replay/system_catalog_identifiers.txt b/test/workload-replay/system_catalog_identifiers.txt index 32959e9a05343..354f1df0e5662 100644 --- a/test/workload-replay/system_catalog_identifiers.txt +++ b/test/workload-replay/system_catalog_identifiers.txt @@ -665,6 +665,13 @@ mz_builtin_sources mz_catalog_raw mz_cluster_deployment_lineage mz_cluster_deployment_lineage_ind +mz_cluster_replica_resource_usage +mz_cluster_replica_resource_usage_s1_primary_idx +mz_cluster_replica_resource_usage_s2_primary_idx +mz_cluster_replica_resource_usage_s3_primary_idx +mz_cluster_replica_resource_usage_s4_primary_idx +mz_cluster_replica_resource_usage_s5_primary_idx +mz_cluster_replica_resource_usage_u1_primary_idx mz_cluster_prometheus_metrics mz_cluster_prometheus_metrics_s1_primary_idx mz_cluster_prometheus_metrics_s2_primary_idx