diff --git a/src/clusterd/src/lib.rs b/src/clusterd/src/lib.rs index 148cf0471d961..4d202a6fcaabd 100644 --- a/src/clusterd/src/lib.rs +++ b/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, diff --git a/src/compute/src/metrics.rs b/src/compute/src/metrics.rs index 9ca9f085e2ff8..16861f1b378e3 100644 --- a/src/compute/src/metrics.rs +++ b/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; @@ -85,8 +87,32 @@ 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)); let mut index_peek_row_buckets = prometheus::exponential_buckets(1.0, 2.0, 25).expect("valid parameters"); @@ -94,164 +120,170 @@ impl ComputeMetrics { // 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 = - 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 = + 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_row_iteration_rows: registry.register(metric!( + ), role)), + index_peek_row_iteration_rows: registry.register(with_role(metric!( name: "mz_index_peek_row_iteration_rows", help: "Number of arrangement rows evaluated by the index peek result iterator.", buckets: index_peek_row_buckets.clone(), - )), - 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_result_sort_rows: registry.register(metric!( + ), role)), + index_peek_result_sort_rows: registry.register(with_role(metric!( name: "mz_index_peek_result_sort_rows", help: "Number of intermediate result rows sorted during peek collection, summed across sort operations.", buckets: index_peek_row_buckets, - )), - 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)), } } @@ -537,3 +569,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(®istry, 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(®istry, ComputeRuntimeRole::Maintenance); + let interactive = ComputeMetrics::register_with(®istry, 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:?}"); + } +} diff --git a/src/compute/src/server.rs b/src/compute/src/server.rs index 3588044852835..3fdbd78e0060c 100644 --- a/src/compute/src/server.rs +++ b/src/compute/src/server.rs @@ -56,6 +56,70 @@ pub struct ComputeInstanceContext { pub connection_context: ConnectionContext, } +/// Which of a process's compute runtimes a given runtime is. +/// +/// A clusterd process runs a single `Solo` runtime by default. When an interactive runtime is +/// configured, the process instead runs a `Maintenance` and an `Interactive` runtime side by side. +/// The named roles share per-process resources (persist cache, metrics registry, log spans). The +/// role distinguishes them so that only the globals-owning runtime runs the non-idempotent +/// process-global initializers, and so metric series and log spans do not collide. +/// +/// `Solo` exists so the single-runtime default stays behaviorally identical to a deployment without +/// a second runtime: no `role` metric label, and it owns the process globals just as the sole +/// runtime always has. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ComputeRuntimeRole { + /// The sole runtime of a single-runtime process. Owns index maintenance and the process-global + /// initializers. + Solo, + /// The maintenance runtime of a two-runtime process. Owns index maintenance and the + /// process-global initializers. + Maintenance, + /// The interactive runtime of a two-runtime process. Shares the process globals owned by + /// maintenance and serves reads. + /// + /// Test-only until the interactive runtime exists to construct it. It is present because the + /// `role` label's entire purpose is that two named roles register into one process registry + /// without colliding, and nothing else can express that: `Solo` registers the same metric names + /// with no `role` label, so prometheus rejects it alongside a named role for differing label + /// dimensions rather than treating it as a second series. Verifying non-collision therefore + /// needs a second *named* role. + /// + /// TODO: drop the `cfg` when the interactive runtime lands and constructs this. + #[cfg(test)] + Interactive, +} + +impl ComputeRuntimeRole { + /// The `role` metric/log label for this role, or `None` for `Solo`. + /// + /// `Solo` omits the label so a single-runtime deployment registers exactly as it did before a + /// second runtime existed, keeping exact-match dashboards and alerts unchanged. + pub fn label(self) -> Option<&'static str> { + match self { + ComputeRuntimeRole::Solo => None, + ComputeRuntimeRole::Maintenance => Some("maintenance"), + #[cfg(test)] + ComputeRuntimeRole::Interactive => Some("interactive"), + } + } + + /// Whether this role runs the non-idempotent, process-global initializers. + /// + /// `Solo` and `Maintenance` run them. An interactive runtime shares the same process and + /// inherits the globals maintenance installs, so re-running them would either double-apply a + /// non-idempotent effect or race maintenance. + /// + /// NOTE: every role a release build can construct owns the globals, so this is constantly true + /// outside tests. The distinction becomes load-bearing when the interactive runtime lands. + pub fn owns_process_globals(self) -> bool { + matches!( + self, + ComputeRuntimeRole::Solo | ComputeRuntimeRole::Maintenance + ) + } +} + /// Type alias for the storage timely log reader. pub(crate) type StorageTimelyLogReader = Arc>>; @@ -84,6 +148,7 @@ struct Config { /// Initiates a timely dataflow computation, processing compute commands. pub async fn serve( timely_config: TimelyConfig, + role: ComputeRuntimeRole, metrics_registry: &MetricsRegistry, persist_clients: Arc, txns_ctx: TxnsContext, @@ -111,7 +176,7 @@ pub async fn serve( persist_clients, txns_ctx, tracing_handle, - metrics: ComputeMetrics::register_with(metrics_registry), + metrics: ComputeMetrics::register_with(metrics_registry, role), context, metrics_registry: metrics_registry.clone(), workers_per_process,