compute: expose replica resource usage observations

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 = <cluster_name>;
+SET cluster_replica = <replica_name>;
+
+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.

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.
 
 <!-- RELATION_SPEC mz_internal.mz_cluster_replica_metrics -->
 | 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.
 
 <!-- RELATION_SPEC mz_internal.mz_cluster_replica_metrics_history -->
 | Field            | Type      | Meaning

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/).
+
+<!-- RELATION_SPEC mz_introspection.mz_cluster_replica_resource_usage NO_COMMENTS -->
+| 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.

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:

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",

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",

src/adapter/src/catalog/open/builtin_schema_migration.rs

@@ -424,6 +424,23 @@ static MIGRATIONS: LazyLock<Vec<MigrationStep>> = 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",
+        ),
     ]
 });
 

src/catalog/src/builtin.rs

@@ -1108,6 +1108,7 @@ pub static BUILTINS_STATIC: LazyLock<Vec<Builtin<NameReference>>> = 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),

src/catalog/src/builtin/mz_introspection.rs

@@ -248,6 +248,21 @@ pub static MZ_CLUSTER_PROMETHEUS_METRICS: LazyLock<BuiltinLog> = LazyLock::new(|
     }),
 });
 
+pub static MZ_CLUSTER_REPLICA_RESOURCE_USAGE: LazyLock<BuiltinLog> = 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<BuiltinLog> = LazyLock::new(|| BuiltinLog {
     name: "mz_compute_frontiers_per_worker",
     schema: MZ_INTROSPECTION_SCHEMA,

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;

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);