compute: add ComputeRuntimeRole and role-labeled metrics

Introduce `ComputeRuntimeRole` (Solo, Maintenance, Interactive) and thread a
role through `serve` into `ComputeMetrics::register_with`. Each named role
stamps a distinct `role` const label on its metrics so that two compute
runtimes sharing one process registry register distinct series rather than
colliding, and gates the whole-registry `workload_class` postprocessor on the
globals-owning role so it is installed exactly once.

`Solo` is the single-runtime default and is behaviorally identical to compute
before a second runtime existed: it emits no `role` label, so exact-match
dashboards and alerts are byte-unchanged, and it owns the process globals as
the sole runtime always has. Every call site passes `Solo` here. The
maintenance and interactive runtimes that use the other roles arrive with the
two-runtime work that builds on this.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CY7GHdrTBfAG4tgvSgPGJ9

Author: antiguru

src/clusterd/src/lib.rs

@@ -21,7 +21,7 @@ use hyper_util::rt::TokioIo;
 use mz_build_info::{BuildInfo, build_info};
 use mz_cloud_resources::AwsExternalIdPrefix;
 use mz_cluster_client::client::TimelyConfig;
-use mz_compute::server::ComputeInstanceContext;
+use mz_compute::server::{ComputeInstanceContext, ComputeRuntimeRole};
 use mz_http_util::DynamicFilterTarget;
 use mz_orchestrator_tracing::{StaticTracingConfig, TracingCliArgs};
 use mz_ore::cli::{self, CliConfig};
@@ -470,6 +470,7 @@ async fn run(args: Args) -> Result<(), anyhow::Error> {
     // Start compute server.
     let compute_client_builder = mz_compute::server::serve(
         compute_timely_config,
+        ComputeRuntimeRole::Solo,
         &metrics_registry,
         persist_clients,
         txns_ctx,

src/compute/src/metrics.rs

@@ -12,7 +12,9 @@ use std::sync::{Arc, Mutex};
 use mz_compute_client::metrics::{CommandMetrics, HistoryMetrics};
 use mz_ore::cast::CastFrom;
 use mz_ore::metric;
-use mz_ore::metrics::{MetricTag, MetricVisibility, MetricsRegistry, UIntGauge, raw};
+use mz_ore::metrics::{
+    MakeCollectorOpts, MetricTag, MetricVisibility, MetricsRegistry, UIntGauge, raw,
+};
 use mz_repr::{GlobalId, SharedRow};
 use prometheus::core::{AtomicF64, GenericCounter};
 use prometheus::proto::LabelPair;
@@ -83,160 +85,190 @@ pub struct ComputeMetrics {
     subscribe_snapshots_skipped_total: IntCounter,
 }
 
+/// Applies the per-role const label to `opts`, unless `role` is `Solo`.
+///
+/// The two named roles (maintenance, interactive) each get a distinct `role` label so a second
+/// compute runtime in the same process registers a distinct series rather than colliding with the
+/// first. `Solo` omits the label so a single-runtime deployment registers exactly as it did before
+/// a second runtime existed.
+fn with_role(
+    mut opts: MakeCollectorOpts,
+    role: crate::server::ComputeRuntimeRole,
+) -> MakeCollectorOpts {
+    if let Some(label) = role.label() {
+        opts.opts = opts.opts.const_label("role", label);
+    }
+    opts
+}
+
 impl ComputeMetrics {
-    pub fn register_with(registry: &MetricsRegistry) -> Self {
+    /// Registers the compute metrics for `role` into `registry`.
+    ///
+    /// The two named roles carry a `role` const label so that a second compute runtime in the same
+    /// process registers a distinct series rather than colliding with the first. `Solo` carries no
+    /// such label.
+    pub fn register_with(
+        registry: &MetricsRegistry,
+        role: crate::server::ComputeRuntimeRole,
+    ) -> Self {
         let workload_class = Arc::new(Mutex::new(None));
 
         // Apply a `workload_class` label to all metrics in the registry when we
         // have a known workload class.
-        registry.register_postprocessor({
-            let workload_class = Arc::clone(&workload_class);
-            move |metrics| {
-                let workload_class: Option<String> =
-                    workload_class.lock().expect("lock poisoned").clone();
-                let Some(workload_class) = workload_class else {
-                    return;
-                };
-                for metric in metrics {
-                    for metric in metric.mut_metric() {
-                        let mut label = LabelPair::default();
-                        label.set_name("workload_class".into());
-                        label.set_value(workload_class.clone());
-
-                        let mut labels = metric.take_label();
-                        labels.push(label);
-                        metric.set_label(labels);
+        //
+        // The postprocessor rewrites every metric in the whole registry, so only the maintenance
+        // runtime registers it. A second registration from the interactive runtime would push the
+        // label twice onto each metric and produce a duplicate-label scrape error.
+        if role.owns_process_globals() {
+            registry.register_postprocessor({
+                let workload_class = Arc::clone(&workload_class);
+                move |metrics| {
+                    let workload_class: Option<String> =
+                        workload_class.lock().expect("lock poisoned").clone();
+                    let Some(workload_class) = workload_class else {
+                        return;
+                    };
+                    for metric in metrics {
+                        for metric in metric.mut_metric() {
+                            let mut label = LabelPair::default();
+                            label.set_name("workload_class".into());
+                            label.set_value(workload_class.clone());
+
+                            let mut labels = metric.take_label();
+                            labels.push(label);
+                            metric.set_label(labels);
+                        }
                     }
                 }
-            }
-        });
+            });
+        }
 
         Self {
             workload_class,
-            history_command_count: registry.register(metric!(
+            history_command_count: registry.register(with_role(metric!(
                 name: "mz_compute_replica_history_command_count",
                 help: "The number of commands in the replica's command history.",
                 var_labels: ["worker_id", "command_type"],
-            )),
-            history_dataflow_count: registry.register(metric!(
+            ), role)),
+            history_dataflow_count: registry.register(with_role(metric!(
                 name: "mz_compute_replica_history_dataflow_count",
                 help: "The number of dataflows in the replica's command history.",
                 var_labels: ["worker_id"],
                 visibility: MetricVisibility::Public,
                 tags: [MetricTag::Compute],
-            )),
-            reconciliation_reused_dataflows_count_total: registry.register(metric!(
+            ), role)),
+            reconciliation_reused_dataflows_count_total: registry.register(with_role(metric!(
                 name: "mz_compute_reconciliation_reused_dataflows_count_total",
                 help: "The total number of dataflows that were reused during compute reconciliation.",
                 var_labels: ["worker_id"],
-            )),
-            reconciliation_replaced_dataflows_count_total: registry.register(metric!(
+            ), role)),
+            reconciliation_replaced_dataflows_count_total: registry.register(with_role(metric!(
                 name: "mz_compute_reconciliation_replaced_dataflows_count_total",
                 help: "The total number of dataflows that were replaced during compute reconciliation.",
                 var_labels: ["worker_id", "reason"],
-            )),
-            arrangement_maintenance_seconds_total: registry.register(metric!(
+            ), role)),
+            arrangement_maintenance_seconds_total: registry.register(with_role(metric!(
                 name: "mz_arrangement_maintenance_seconds_total",
                 help: "The total time spent maintaining arrangements.",
                 var_labels: ["worker_id"],
                 visibility: MetricVisibility::Public,
                 tags: [MetricTag::Compute],
-            )),
-            arrangement_maintenance_active_info: registry.register(metric!(
+            ), role)),
+            arrangement_maintenance_active_info: registry.register(with_role(metric!(
                 name: "mz_arrangement_maintenance_active_info",
                 help: "Whether maintenance is currently occuring.",
                 var_labels: ["worker_id"],
-            )),
-            timely_step_duration_seconds: registry.register(metric!(
+            ), role)),
+            timely_step_duration_seconds: registry.register(with_role(metric!(
                 name: "mz_timely_step_duration_seconds",
                 help: "The time spent in each compute step_or_park call",
                 const_labels: {"cluster" => "compute"},
                 var_labels: ["worker_id"],
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 32.0),
-            )),
-            shared_row_heap_capacity_bytes: registry.register(metric!(
+            ), role)),
+            shared_row_heap_capacity_bytes: registry.register(with_role(metric!(
                 name: "mz_dataflow_shared_row_heap_capacity_bytes",
                 help: "The heap capacity of the shared row.",
                 var_labels: ["worker_id"],
-            )),
-            persist_peek_seconds: registry.register(metric!(
+            ), role)),
+            persist_peek_seconds: registry.register(with_role(metric!(
                 name: "mz_persist_peek_seconds",
                 help: "Time spent in (experimental) Persist fast-path peeks.",
                 var_labels: ["worker_id"],
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
-            )),
-            stashed_peek_seconds: registry.register(metric!(
+            ), role)),
+            stashed_peek_seconds: registry.register(with_role(metric!(
                 name: "mz_stashed_peek_seconds",
                 help: "Time spent reading a peek result and stashing it in the peek result stash (aka. persist blob).",
                 var_labels: ["worker_id"],
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
-            )),
-            handle_command_duration_seconds: registry.register(metric!(
+            ), role)),
+            handle_command_duration_seconds: registry.register(with_role(metric!(
                 name: "mz_cluster_handle_command_duration_seconds",
                 help: "Time spent in handling commands.",
                 const_labels: {"cluster" => "compute"},
                 var_labels: ["worker_id", "command_type"],
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
-            )),
-            index_peek_total_seconds: registry.register(metric!(
+            ), role)),
+            index_peek_total_seconds: registry.register(with_role(metric!(
                 name: "mz_index_peek_total_seconds",
                 help: "Total time processing index peeks, from process_peek entry to response. Excluding peeks that use the peek response stash.",
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
-            )),
-            index_peek_seek_fulfillment_seconds: registry.register(metric!(
+            ), role)),
+            index_peek_seek_fulfillment_seconds: registry.register(with_role(metric!(
                 name: "mz_index_peek_seek_fulfillment_seconds",
                 help: "Time in seek_fulfillment method including frontier checks and data collection.",
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
-            )),
-            index_peek_error_scan_seconds: registry.register(metric!(
+            ), role)),
+            index_peek_error_scan_seconds: registry.register(with_role(metric!(
                 name: "mz_index_peek_error_scan_seconds",
                 help: "Time scanning the error trace for errors.",
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
-            )),
-            index_peek_cursor_setup_seconds: registry.register(metric!(
+            ), role)),
+            index_peek_cursor_setup_seconds: registry.register(with_role(metric!(
                 name: "mz_index_peek_cursor_setup_seconds",
                 help: "Time setting up cursor and literal constraints.",
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
-            )),
-            index_peek_row_iteration_seconds: registry.register(metric!(
+            ), role)),
+            index_peek_row_iteration_seconds: registry.register(with_role(metric!(
                 name: "mz_index_peek_row_iteration_seconds",
                 help: "Time iterating rows and evaluating MFP.",
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
-            )),
-            index_peek_result_sort_seconds: registry.register(metric!(
+            ), role)),
+            index_peek_result_sort_seconds: registry.register(with_role(metric!(
                 name: "mz_index_peek_result_sort_seconds",
                 help: "Time sorting intermediate results during peek collection.",
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
-            )),
-            index_peek_frontier_check_seconds: registry.register(metric!(
+            ), role)),
+            index_peek_frontier_check_seconds: registry.register(with_role(metric!(
                 name: "mz_index_peek_frontier_check_seconds",
                 help: "Time checking trace frontiers.",
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
-            )),
-            index_peek_row_collection_seconds: registry.register(metric!(
+            ), role)),
+            index_peek_row_collection_seconds: registry.register(with_role(metric!(
                 name: "mz_index_peek_row_collection_seconds",
                 help: "Time constructing RowCollection from peek results.",
                 buckets: mz_ore::stats::histogram_seconds_buckets(0.000_128, 8.0),
-            )),
-            replica_expiration_timestamp_seconds: registry.register(metric!(
+            ), role)),
+            replica_expiration_timestamp_seconds: registry.register(with_role(metric!(
                 name: "mz_dataflow_replica_expiration_timestamp_seconds",
                 help: "The replica expiration timestamp in seconds since epoch.",
                 var_labels: ["worker_id"],
-            )),
-            replica_expiration_remaining_seconds: registry.register(metric!(
+            ), role)),
+            replica_expiration_remaining_seconds: registry.register(with_role(metric!(
                 name: "mz_dataflow_replica_expiration_remaining_seconds",
                 help: "The remaining seconds until replica expiration. Can go negative, can lag behind.",
                 var_labels: ["worker_id"],
-            )),
-            collection_count: registry.register(metric!(
+            ), role)),
+            collection_count: registry.register(with_role(metric!(
                 name: "mz_compute_collection_count",
                 help: "The number and hydration status of maintained compute collections.",
                 var_labels: ["worker_id", "type", "hydrated"],
-            )),
-            subscribe_snapshots_skipped_total: registry.register(metric!(
+            ), role)),
+            subscribe_snapshots_skipped_total: registry.register(with_role(metric!(
                 name: "mz_subscribe_snapshots_skipped_total",
                 help: "The number of collection snapshots that were skipped by the subscribe snapshot optimization.",
-            )),
+            ), role)),
         }
     }
 
@@ -514,3 +546,64 @@ impl Drop for CollectionMetrics {
             .dec_collection_count(self.collection_type, self.collection_hydrated);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeSet;
+
+    use mz_ore::metrics::MetricsRegistry;
+
+    use super::ComputeMetrics;
+    use crate::server::ComputeRuntimeRole;
+
+    /// The `Solo` (single-runtime) role registers exactly as compute did before a second runtime
+    /// existed: no metric carries a `role` label, so single-runtime dashboards and exact-match
+    /// alerts are byte-unchanged.
+    #[mz_ore::test]
+    fn solo_runtime_omits_role_label() {
+        let registry = MetricsRegistry::new();
+        let metrics = ComputeMetrics::register_with(&registry, ComputeRuntimeRole::Solo);
+        // Instantiate the per-worker children so the `*Vec` families emit rows to inspect.
+        let _worker = metrics.for_worker(0);
+
+        for family in registry.gather() {
+            for metric in family.get_metric() {
+                for label in metric.get_label() {
+                    assert_ne!(
+                        label.name(),
+                        "role",
+                        "solo metric {} unexpectedly carries a role label",
+                        family.name(),
+                    );
+                }
+            }
+        }
+    }
+
+    /// The two named roles each carry their own `role` label, so two runtimes in one process
+    /// register distinct series rather than colliding. Registering both on one registry also
+    /// exercises the non-collision that lets them coexist.
+    #[mz_ore::test]
+    fn named_roles_carry_distinct_role_label() {
+        let registry = MetricsRegistry::new();
+        let maintenance = ComputeMetrics::register_with(&registry, ComputeRuntimeRole::Maintenance);
+        let interactive = ComputeMetrics::register_with(&registry, ComputeRuntimeRole::Interactive);
+        let _maintenance_worker = maintenance.for_worker(0);
+        let _interactive_worker = interactive.for_worker(0);
+
+        let mut roles = BTreeSet::new();
+        for family in registry.gather() {
+            for metric in family.get_metric() {
+                let role = metric
+                    .get_label()
+                    .iter()
+                    .find(|label| label.name() == "role")
+                    .unwrap_or_else(|| panic!("metric {} missing a role label", family.name()));
+                roles.insert(role.value().to_string());
+            }
+        }
+
+        assert!(roles.contains("maintenance"), "roles seen: {roles:?}");
+        assert!(roles.contains("interactive"), "roles seen: {roles:?}");
+    }
+}