diff --git a/src/adapter/src/coord.rs b/src/adapter/src/coord.rs index 721ba8975bcd4..bdc4ab5196caa 100644 --- a/src/adapter/src/coord.rs +++ b/src/adapter/src/coord.rs @@ -200,6 +200,7 @@ use crate::coord::appends::{ use crate::coord::caught_up::CaughtUpCheckContext; use crate::coord::id_bundle::CollectionIdBundle; use crate::coord::introspection::IntrospectionSubscribe; +use crate::coord::metric_sink::{CuratedMetricSink, InstalledMetricSink, PlannedMetricSink}; use crate::coord::peek::PendingPeek; use crate::coord::statement_logging::StatementLogging; use crate::coord::timeline::{TimelineContext, TimelineState}; @@ -243,6 +244,7 @@ mod indexes; mod info_metrics; mod introspection; mod message_handler; +mod metric_sink; mod privatelink_status; mod sql; mod validity; @@ -449,6 +451,10 @@ pub enum Message { span: Span, stage: IntrospectionSubscribeStage, }, + MetricSinkStageReady { + span: Span, + stage: MetricSinkStage, + }, SecretStageReady { ctx: ExecuteContext, span: Span, @@ -575,6 +581,7 @@ impl Message { Message::IntrospectionSubscribeStageReady { .. } => { "introspection_subscribe_stage_ready" } + Message::MetricSinkStageReady { .. } => "metric_sink_stage_ready", Message::SecretStageReady { .. } => "secret_stage_ready", Message::ClusterStageReady { .. } => "cluster_stage_ready", Message::DrainStatementLog => "drain_statement_log", @@ -1181,6 +1188,36 @@ pub struct IntrospectionSubscribeFinish { replica_id: ReplicaId, } +#[derive(Debug)] +pub enum MetricSinkStage { + Optimize(MetricSinkOptimize), + Finish(MetricSinkFinish), +} + +#[derive(Debug)] +pub struct MetricSinkOptimize { + validity: PlanValidity, + definition: &'static CuratedMetricSink, + /// The transient id of the sink's compute export. Recorded in + /// [`Coordinator::metric_sinks`] once the finish stage ships the dataflow. + sink_id: GlobalId, + /// The planned `source_sql`, and the shape it produces. + expr: HirRelationExpr, + desc: RelationDesc, + cluster_id: ComputeInstanceId, + replica_id: ReplicaId, +} + +#[derive(Debug)] +pub struct MetricSinkFinish { + validity: PlanValidity, + definition: &'static CuratedMetricSink, + sink_id: GlobalId, + global_lir_plan: optimize::metric_sink::GlobalLirPlan, + cluster_id: ComputeInstanceId, + replica_id: ReplicaId, +} + #[derive(Debug)] pub enum SecretStage { CreateEnsure(CreateSecretEnsure), @@ -2117,6 +2154,14 @@ pub struct Coordinator { hydration_history_replica_cursor: Option, /// Hydration-history sweep owned by the coordinator while one is in flight. hydration_history_sweep: Option>, + /// The curated metric sinks installed on each replica. + /// + /// Keyed replica-first so a replica's installs form one contiguous range: teardown on replica + /// drop is the only lookup that is not by exact key. + metric_sinks: BTreeMap<(ReplicaId, &'static str), InstalledMetricSink>, + /// Curated metric-sink plans, cached per definition so each is planned once rather than once + /// per replica. See [`Coordinator::plan_metric_sink`]. + metric_sink_plans: BTreeMap<&'static str, PlannedMetricSink>, /// Locks that grant access to a specific object, populated lazily as objects are written to. write_locks: BTreeMap>>, @@ -3112,6 +3157,9 @@ impl Coordinator { // Initialize unified introspection. self.bootstrap_introspection_subscribes().await; + // Install the curated metric sinks on every replica. + self.bootstrap_metric_sinks().await; + info!( "startup: coordinator init: bootstrap: migrate builtin tables in read-only mode complete in {:?}", final_steps_start.elapsed() @@ -3909,9 +3957,12 @@ impl Coordinator { // MIR ⇒ MIR optimization (global) let metric_sink_plan = optimize::metric_sink::MetricSink::new( - entry.name().clone(), - metric_sink.from, + self.catalog() + .resolve_full_name(entry.name(), None) + .to_string(), + optimize::metric_sink::MetricSinkFrom::Id(metric_sink.from), metric_sink.prefix.clone(), + None, ); let global_mir_plan = optimizer.optimize(metric_sink_plan)?; let optimized_plan = global_mir_plan.df_desc().clone(); @@ -5341,6 +5392,8 @@ pub fn serve( introspection_subscribes: BTreeMap::new(), hydration_history_replica_cursor: None, hydration_history_sweep: None, + metric_sinks: BTreeMap::new(), + metric_sink_plans: BTreeMap::new(), write_locks: BTreeMap::new(), deferred_write_ops: BTreeMap::new(), pending_writes: Vec::new(), diff --git a/src/adapter/src/coord/catalog_implications.rs b/src/adapter/src/coord/catalog_implications.rs index 432865aa6c757..4acb9fa7885cd 100644 --- a/src/adapter/src/coord/catalog_implications.rs +++ b/src/adapter/src/coord/catalog_implications.rs @@ -1688,6 +1688,7 @@ impl Coordinator { self.install_introspection_subscribes(cluster_id, replica_id) .await; + self.install_metric_sinks(cluster_id, replica_id).await; } } diff --git a/src/adapter/src/coord/ddl.rs b/src/adapter/src/coord/ddl.rs index 01d91d0da4c34..a451443eb559a 100644 --- a/src/adapter/src/coord/ddl.rs +++ b/src/adapter/src/coord/ddl.rs @@ -719,6 +719,7 @@ impl Coordinator { pub(crate) fn drop_replica(&mut self, cluster_id: ClusterId, replica_id: ReplicaId) { self.drop_introspection_subscribes(replica_id); + self.drop_metric_sinks(replica_id); self.controller .drop_replica(cluster_id, replica_id) diff --git a/src/adapter/src/coord/introspection.rs b/src/adapter/src/coord/introspection.rs index 2f50e4625f739..0f62f754cfb56 100644 --- a/src/adapter/src/coord/introspection.rs +++ b/src/adapter/src/coord/introspection.rs @@ -98,18 +98,27 @@ impl IntrospectionSubscribe { } impl Coordinator { + /// Every `(cluster, replica)` pair currently in the catalog. + /// + /// The set a per-replica feature (introspection subscribes, curated metric sinks) must install + /// onto the replicas that already exist when the coordinator starts. Shared so those callers + /// cannot drift on what "all replicas" means. + pub(super) fn all_cluster_replicas(&self) -> Vec<(ClusterId, ReplicaId)> { + self.catalog + .clusters() + .flat_map(|cluster| { + cluster + .replicas() + .map(move |replica| (cluster.id, replica.replica_id)) + }) + .collect() + } + /// Installs introspection subscribes on all existing replicas. /// /// Meant to be invoked during coordinator bootstrapping. pub(super) async fn bootstrap_introspection_subscribes(&mut self) { - let mut cluster_replicas = Vec::new(); - for cluster in self.catalog.clusters() { - for replica in cluster.replicas() { - cluster_replicas.push((cluster.id, replica.replica_id)); - } - } - - for (cluster_id, replica_id) in cluster_replicas { + for (cluster_id, replica_id) in self.all_cluster_replicas() { self.install_introspection_subscribes(cluster_id, replica_id) .await; } diff --git a/src/adapter/src/coord/message_handler.rs b/src/adapter/src/coord/message_handler.rs index bb8fc01546ec3..b92d20d4ab0a3 100644 --- a/src/adapter/src/coord/message_handler.rs +++ b/src/adapter/src/coord/message_handler.rs @@ -231,6 +231,9 @@ impl Coordinator { Message::IntrospectionSubscribeStageReady { span, stage } => { self.sequence_staged((), span, stage).boxed_local().await; } + Message::MetricSinkStageReady { span, stage } => { + self.sequence_staged((), span, stage).boxed_local().await; + } Message::ExplainTimestampStageReady { ctx, span, stage } => { self.sequence_staged(ctx, span, stage).boxed_local().await; } diff --git a/src/adapter/src/coord/metric_sink.rs b/src/adapter/src/coord/metric_sink.rs new file mode 100644 index 0000000000000..f0e4a8ea38ea6 --- /dev/null +++ b/src/adapter/src/coord/metric_sink.rs @@ -0,0 +1,759 @@ +// 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. + +//! Coordinator-installed metric sinks, the curated counterpart to `CREATE METRIC SINK`. +//! +//! A curated metric sink is a [`CURATED`] entry rendered on every replica, publishing its series +//! into that replica's process-local Prometheus registry. Unlike a user's `CREATE METRIC SINK` it +//! is not a catalog item: it gets a transient [`GlobalId`], targets one replica rather than a +//! cluster, and is re-created from the static list on every boot. Modelling the curated set this +//! way keeps it out of the catalog, so adding or removing a definition needs no builtin migration. +//! +//! Every replica means every replica of every cluster, user clusters included. Each definition is +//! therefore a dataflow, with its arrangements, on customer compute, charged to that customer's +//! cluster, and the cost scales with `CURATED`. `coord::introspection` already accepts this for its +//! subscribes. +//! +//! # Lifecycle +//! +//! * After a new replica is created, the coordinator calls `install_metric_sinks` to install every +//! definition on it. `bootstrap_metric_sinks` does the same for the replicas that already exist +//! when the coordinator starts. +//! * Before a replica is dropped, the coordinator calls `drop_metric_sinks` to drop the sinks +//! installed on it. +//! * A replica that disconnects and reconnects (a crash, an OOM) has its dataflows re-rendered from +//! the controller's state, so unlike an introspection subscribe there is nothing to reinstall. +//! +//! This mirrors [`crate::coord::introspection`], which installs introspection subscribes on the +//! same triggers. + +use std::collections::{BTreeMap, BTreeSet}; + +use anyhow::bail; +use mz_catalog::memory::objects::CatalogItem; +use mz_cluster_client::ReplicaId; +use mz_controller_types::ClusterId; +use mz_ore::collections::CollectionExt; +use mz_ore::{instrument, soft_panic_or_log}; +use mz_repr::optimize::OverrideFrom; +use mz_repr::{CatalogItemId, GlobalId, RelationDesc}; +use mz_sql::catalog::SessionCatalog; +use mz_sql::plan::{ + HirRelationExpr, Params, Plan, SubscribeFrom, SubscribePlan, validate_metric_sink_desc, + validate_metric_sink_prefix, +}; +use mz_sql::session::user::{MZ_SYSTEM_ROLE_ID, RoleMetadata}; +use mz_sql::session::vars::ENABLE_METRIC_SINK; +use tracing::{Span, info}; + +use crate::catalog::Catalog; +use crate::coord::{ + Coordinator, Message, MetricSinkFinish, MetricSinkOptimize, MetricSinkStage, PlanValidity, + StageResult, Staged, +}; +use crate::optimize::Optimize; +use crate::optimize::dataflows::dataflow_import_id_bundle; +use crate::{AdapterError, ExecuteResponse, optimize}; + +/// A curated metric sink: SQL producing the canonical metric-sink columns, plus the name it is +/// known by in logs. +#[derive(Debug)] +pub(super) struct CuratedMetricSink { + /// Stable identifier for the definition: used in logs, as the [`Coordinator::metric_sinks`] key, + /// and as the `sink` label on the health gauges (the `GlobalId` is transient, the name is not). + /// Must be unique within [`CURATED`]. + name: &'static str, + /// A `SELECT` producing the canonical metric-sink columns (`metric_name`, `metric_type`, + /// `labels`, `value`, `help`), the contract `mz_sql::plan::validate_metric_sink_desc` checks. + /// + /// The query must read only introspection relations. A catalog-backed relation would put + /// envd's write frontier on the sink's emission path, which is exactly the coupling these + /// sinks exist to avoid: the sink would stall whenever envd did, taking the freshness signal + /// with it. + source_sql: &'static str, + /// Prepended to every row's `metric_name` to form the published name, exactly as a user's + /// `CREATE METRIC SINK ... WITH (PREFIX = ...)`. Must start with `mz_metric_sink_` so the + /// published families land in the reserved lane (see `validate_metric_sink_prefix`). + prefix: &'static str, +} + +/// The curated metric sinks, installed on every replica. +const CURATED: &[CuratedMetricSink] = &[]; + +/// A [`CuratedMetricSink`] installed on one replica. +#[derive(Debug)] +pub(super) struct InstalledMetricSink { + /// The cluster the replica belongs to, needed to drop the sink's compute collection. + cluster_id: ClusterId, + /// The transient id of the sink's compute export. + sink_id: GlobalId, +} + +/// A [`CuratedMetricSink`] planned once and shared across the replicas it installs on. See +/// [`Coordinator::plan_metric_sink`]. +#[derive(Clone, Debug)] +pub(super) struct PlannedMetricSink { + /// The shaped source query. + expr: HirRelationExpr, + /// The shape `expr` produces. + desc: RelationDesc, + /// The catalog items the source reads. + dependencies: BTreeSet, +} + +impl Coordinator { + /// Installs the curated metric sinks on all existing replicas. + /// + /// Meant to be invoked during coordinator bootstrapping. + pub(super) async fn bootstrap_metric_sinks(&mut self) { + for (cluster_id, replica_id) in self.all_cluster_replicas() { + self.install_metric_sinks(cluster_id, replica_id).await; + } + } + + /// Installs the curated metric sinks on the given replica. + /// + /// Turning `enable_metric_sink` off stops installing on replicas created from then on. It does + /// not tear down what is already installed: those keep running until their replica is dropped + /// or envd restarts. A replica that merely reconnects re-renders them from the controller's + /// command history, so a replica restart does not clear them either. + pub(super) async fn install_metric_sinks( + &mut self, + cluster_id: ClusterId, + replica_id: ReplicaId, + ) { + if !ENABLE_METRIC_SINK.enabled(self.catalog().system_config()) { + return; + } + + // TODO: Skip replicas created with introspection disabled. Their logging dataflows never + // run, so a `source_sql` reading introspection relations there never advances. That is not + // just wasted work: the sink publishes its input frontier as its write frontier, so a + // never-advancing input stalls the sink's frontier at its as-of and pins the read holds it + // takes on those collections for the replica's whole life (replica-local, released on + // drop). `coord::introspection` installs subscribes on the same triggers and has the same + // gap. + for definition in CURATED { + self.install_metric_sink(cluster_id, replica_id, definition) + .await; + } + } + + async fn install_metric_sink( + &mut self, + cluster_id: ClusterId, + replica_id: ReplicaId, + definition: &'static CuratedMetricSink, + ) { + // Cheap duplicate check before planning: if the definition is already installed on this + // replica, there is nothing to do. `metric_sink_finish` keeps a backstop for a double + // install still in flight (not yet recorded here). + if self + .metric_sinks + .contains_key(&(replica_id, definition.name)) + { + return; + } + + let Some(planned) = self.plan_metric_sink(definition) else { + return; + }; + + let (_, sink_id) = self.allocate_transient_id(); + // Logged only once the definition is known good, so an abandoned install leaves no + // misleading "installing" line. + info!(%sink_id, %replica_id, name = definition.name, "installing metric sink"); + + let validity = PlanValidity::new( + &self.catalog, + planned.dependencies.clone(), + Some(cluster_id), + Some(replica_id), + RoleMetadata::new(MZ_SYSTEM_ROLE_ID), + ); + let stage = MetricSinkStage::Optimize(MetricSinkOptimize { + validity, + definition, + sink_id, + expr: planned.expr.clone(), + desc: planned.desc.clone(), + cluster_id, + replica_id, + }); + self.sequence_staged((), Span::current(), stage).await; + } + + /// Plans a curated definition once, caching the result in [`Coordinator::metric_sink_plans`]. + /// + /// The plan depends only on the catalog, never on the replica, so it is shared across every + /// replica the definition installs on rather than re-planned per replica. Curated sources read + /// only builtins (enforced by [`ensure_reads_only_logs`]), which do not change while envd runs, + /// so a cached plan stays valid for envd's lifetime. Returns `None` for an invalid definition, + /// having soft-panicked. + fn plan_metric_sink( + &mut self, + definition: &'static CuratedMetricSink, + ) -> Option { + if let Some(planned) = self.metric_sink_plans.get(definition.name) { + return Some(planned.clone()); + } + + // A user sink's prefix is validated at plan time; a curated one has no such gate, so enforce + // the same contract here. The prefix keeps published families in the reserved lane and + // supplies the leading character the row shaping's name validation needs. A failure is a bug + // in our own definition, hence `soft_panic_or_log!`. + // + // NOTE: This checks only the prefix format, not collisions. A curated prefix is not checked + // against user sinks (`ensure_metric_sink_prefix_is_free` cannot see a non-catalog item) nor + // against other curated definitions, yet both share the `mz_metric_sink_` lane and could + // overlap. Deferred until `CURATED` is populated. + if let Err(err) = validate_metric_sink_prefix(definition.prefix) { + soft_panic_or_log!( + "invalid curated metric sink prefix (name={}): {err}", + definition.name + ); + return None; + } + + let catalog = self.catalog().for_system_session(); + let (expr, desc, dependencies) = match definition.plan_source(&catalog) { + Ok(planned) => planned, + Err(err) => { + soft_panic_or_log!( + "invalid curated metric sink (name={}): {err}", + definition.name + ); + return None; + } + }; + + // Enforce the introspection-only contract before any optimization work, against what the + // definition reads rather than how the optimizer imports it. + if let Err(err) = ensure_reads_only_logs(&self.catalog, &dependencies) { + soft_panic_or_log!( + "invalid curated metric sink (name={}): {err}", + definition.name + ); + return None; + } + + let planned = PlannedMetricSink { + expr, + desc, + dependencies, + }; + self.metric_sink_plans + .insert(definition.name, planned.clone()); + Some(planned) + } + + #[instrument] + fn metric_sink_optimize( + &self, + stage: MetricSinkOptimize, + ) -> Result>, AdapterError> { + let MetricSinkOptimize { + mut validity, + definition, + sink_id, + expr, + desc, + cluster_id, + replica_id, + } = stage; + + let compute_instance = self + .instance_snapshot(cluster_id) + .expect("compute instance exists"); + // A transient id for the view the optimizer builds to shape the source rows, scoped to this + // dataflow. See `optimize::metric_sink::shape_metric_sink_source`. + let (_, view_id) = self.allocate_transient_id(); + + let optimizer_config = optimize::OptimizerConfig::from(self.catalog().system_config()) + .override_from(&self.catalog.get_cluster(cluster_id).config.features()) + .override_from(&self.cluster_scoped_optimizer_overrides(cluster_id)); + + let mut optimizer = optimize::metric_sink::Optimizer::new( + self.owned_catalog(), + compute_instance, + view_id, + sink_id, + optimizer_config, + self.optimizer_metrics(), + ); + let catalog = self.owned_catalog(); + + let span = Span::current(); + Ok(StageResult::Handle(mz_ore::task::spawn_blocking( + || "optimize metric sink", + move || { + span.in_scope(|| { + let metric_sink = optimize::metric_sink::MetricSink::new( + format!("metric-sink-{}-{replica_id}", definition.name), + optimize::metric_sink::MetricSinkFrom::Query { expr, desc }, + definition.prefix.to_string(), + Some(definition.name.to_string()), + ); + + // Both steps run inside one closure so either failure hits the same log. + // `sequence_staged` has no session to report to for a coordinator-driven + // install, so an error would otherwise vanish. + let global_lir_plan = (|| { + // MIR ⇒ MIR optimization (global) + let global_mir_plan = optimizer.catch_unwind_optimize(metric_sink)?; + // The optimizer imports indexes the SQL never named. Fold them into + // validity so one dropped before the finish stage fails the recheck rather + // than shipping a dataflow that imports a gone collection. + let id_bundle = + dataflow_import_id_bundle(global_mir_plan.df_desc(), cluster_id); + let item_ids = id_bundle.iter().map(|id| catalog.resolve_item_id(&id)); + validity.extend_dependencies(&catalog, item_ids); + // MIR ⇒ LIR lowering and LIR ⇒ LIR optimization (global) + optimizer.catch_unwind_optimize(global_mir_plan) + })() + .inspect_err(|err| { + soft_panic_or_log!( + "curated metric sink failed to optimize (name={}): {err}", + definition.name + ) + })?; + + let stage = MetricSinkStage::Finish(MetricSinkFinish { + validity, + definition, + sink_id, + global_lir_plan, + cluster_id, + replica_id, + }); + Ok(Box::new(stage)) + }) + }, + ))) + } + + #[instrument] + async fn metric_sink_finish( + &mut self, + stage: MetricSinkFinish, + ) -> Result>, AdapterError> { + let MetricSinkFinish { + validity: _, + definition, + sink_id, + global_lir_plan, + cluster_id, + replica_id, + } = stage; + + // `sequence_staged` rechecked validity before this stage ran, so the replica still exists. + // The coordinator handles one message at a time, so no replica drop runs between that check + // and the ship below. + + // The metainfo is dropped rather than persisted: a curated sink is not a catalog item, so + // there is nothing for `mz_optimizer_notices` to hang its notices off. + let (mut df_desc, _df_meta) = global_lir_plan.unapply(); + + let id_bundle = dataflow_import_id_bundle(&df_desc, cluster_id); + + // Backstop for the introspection-only contract; the real gate is `ensure_reads_only_logs` + // at install time. A log-only source imports only compute collections, so this should never + // fire, but a storage import would couple the sink's frontier to envd. + if !id_bundle.storage_ids.is_empty() { + soft_panic_or_log!( + "curated metric sink reads non-introspection relations (name={}): {:?}", + definition.name, + id_bundle.storage_ids + ); + return Ok(StageResult::Response(ExecuteResponse::CreatedMetricSink)); + } + + // Hold a read on the imports across shipping, so their since cannot advance past the as-of + // just picked. Compute takes its own holds during `create_dataflow`. + let read_holds = self.acquire_read_holds(&id_bundle); + df_desc.set_as_of(read_holds.least_valid_read()); + + // Record the install now that the dataflow is about to ship, so a definition that fails to + // plan or optimize leaves no entry behind. `drop_metric_sinks` uses this entry to release the + // sink's instance-global collection state when the replica is dropped. Recording after the + // ship (an introspection subscribe records before sequencing) is safe: validity was rechecked + // at this stage and nothing awaits before the ship, so no replica drop can interleave. + let install = InstalledMetricSink { + cluster_id, + sink_id, + }; + if let Some(previous) = self + .metric_sinks + .insert((replica_id, definition.name), install) + { + // The key is already taken. `curated_names_are_unique` rules out two definitions + // colliding, so the reachable cause is `install_metric_sinks` running twice for one + // replica. Restore the first install and abandon this one: shipping both would leak the + // first's collection (now unreachable to `drop_metric_sinks`) and register a second + // collector under the same `sink` label. + self.metric_sinks + .insert((replica_id, definition.name), previous); + soft_panic_or_log!( + "metric sink installed twice (name={}, replica_id={replica_id})", + definition.name + ); + return Ok(StageResult::Response(ExecuteResponse::CreatedMetricSink)); + } + + self.ship_dataflow(df_desc, cluster_id, Some(replica_id)) + .await; + + drop(read_holds); + // Nobody is waiting on this: `StagedContext for ()` drops the result. Reuses the + // `CREATE METRIC SINK` response rather than adding a variant no client ever sees. + Ok(StageResult::Response(ExecuteResponse::CreatedMetricSink)) + } + + /// Drops the curated metric sinks installed on the given replica. + /// + /// Called before the replica itself is dropped. Dropping the replica would tear the sink + /// dataflows down anyway, but the controller's collection state for them is instance-global, + /// so it has to be released explicitly. + pub(super) fn drop_metric_sinks(&mut self, replica_id: ReplicaId) { + for (name, cluster_id, sink_id) in metric_sinks_on_replica(&self.metric_sinks, replica_id) { + info!(%sink_id, %replica_id, name, "dropping metric sink"); + self.metric_sinks.remove(&(replica_id, name)); + + // The collection exists (the entry is recorded only after the dataflow ships), so this + // drop succeeds. Result ignored: a failure during replica teardown is not worth a panic. + let _ = self + .controller + .compute + .drop_collections(cluster_id, vec![sink_id]); + } + } +} + +/// The registry entries installed on `replica_id`, as `(name, cluster, sink)` in key order. +/// +/// The map is keyed replica-first, so a replica's installs are one contiguous range. +fn metric_sinks_on_replica( + metric_sinks: &BTreeMap<(ReplicaId, &'static str), InstalledMetricSink>, + replica_id: ReplicaId, +) -> Vec<(&'static str, ClusterId, GlobalId)> { + metric_sinks + .range((replica_id, "")..) + .take_while(|((id, _), _)| *id == replica_id) + .map(|((_, name), install)| (*name, install.cluster_id, install.sink_id)) + .collect() +} + +/// Enforces the introspection-only contract from [`CuratedMetricSink::source_sql`]: every relation +/// the definition reads, walking views transitively, must be a log collection. A storage-backed +/// read would put envd's write frontier on the sink's emission path, the coupling these sinks exist +/// to avoid. +/// +/// Checked here against what the definition reads rather than by import kind after optimization: the +/// import split (storage vs index) depends on which indexes the target cluster happens to have, so +/// it gives the same definition different verdicts on different clusters. +fn ensure_reads_only_logs( + catalog: &Catalog, + dependencies: &BTreeSet, +) -> Result<(), anyhow::Error> { + let mut to_visit: Vec<_> = dependencies.iter().copied().collect(); + let mut visited = BTreeSet::new(); + while let Some(id) = to_visit.pop() { + if !visited.insert(id) { + continue; + } + let entry = catalog.get_entry(&id); + match entry.item() { + // The only data leaf allowed. + CatalogItem::Log(_) => {} + // Allowed only if everything it reads is, so walk its dependencies. + CatalogItem::View(_) => to_visit.extend(entry.uses()), + // No data dependency; a view over logs still references these. + CatalogItem::Type(_) | CatalogItem::Func(_) => {} + _ => bail!( + "curated metric sink reads {}, which is not an introspection log relation \ + (only logs and views over logs are allowed)", + catalog.resolve_full_name(entry.name(), None) + ), + } + } + Ok(()) +} + +impl CuratedMetricSink { + /// Plans `source_sql` against a session-less catalog, returning the query, its output shape, + /// and the catalog items it reads. + fn plan_source( + &self, + catalog: &dyn SessionCatalog, + ) -> Result<(HirRelationExpr, RelationDesc, BTreeSet), anyhow::Error> { + // A definition is a single statement. Reject the count explicitly for a clear error. + let statements = mz_sql::parse::parse(self.source_sql)?; + if statements.len() != 1 { + bail!( + "source SQL must be exactly one statement, got {}", + statements.len() + ); + } + + // A metric sink's source is a continuously maintained dataflow, like a SUBSCRIBE, so plan it + // as one. A maintained lifetime folds any finishing into the expression (an ORDER BY over a + // maintained collection is dropped, a LIMIT becomes a TopK) rather than leaving it beside the + // query, so `MetricSinkFrom::Query` gets a self-contained expression whose arity matches its + // `desc`. This mirrors `coord::introspection`, which plans its specs as subscribes too. + let subscribe_sql = format!("SUBSCRIBE ({})", self.source_sql); + let parsed = mz_sql::parse::parse(&subscribe_sql)?.into_element(); + let (stmt, resolved_ids) = mz_sql::names::resolve(catalog, parsed.ast)?; + let (plan, sql_impl_ids) = + mz_sql::plan::plan(None, catalog, stmt, &Params::empty(), &resolved_ids)?; + let Plan::Subscribe(SubscribePlan { + from: SubscribeFrom::Query { expr, desc }, + .. + }) = plan + else { + bail!("source SQL must be a single SELECT"); + }; + validate_metric_sink_desc(&desc)?; + + // Fold in ids from SQL-implemented function bodies. `plan` keeps them out of `resolved_ids` + // since a one-shot statement doesn't depend on a function's body, but a metric sink inlines + // that body into its dataflow, so the body's reads are real imports the gate must check. + let dependencies = resolved_ids + .items() + .chain(sql_impl_ids.items()) + .copied() + .collect(); + Ok((expr, desc, dependencies)) + } +} + +impl Staged for MetricSinkStage { + type Ctx = (); + + fn validity(&mut self) -> &mut PlanValidity { + match self { + Self::Optimize(stage) => &mut stage.validity, + Self::Finish(stage) => &mut stage.validity, + } + } + + async fn stage( + self, + coord: &mut Coordinator, + _ctx: &mut (), + ) -> Result>, AdapterError> { + match self { + Self::Optimize(stage) => coord.metric_sink_optimize(stage), + Self::Finish(stage) => coord.metric_sink_finish(stage).await, + } + } + + fn message(self, _ctx: (), span: Span) -> Message { + Message::MetricSinkStageReady { span, stage: self } + } + + fn cancel_enabled(&self) -> bool { + false + } +} + +#[cfg(test)] +mod tests { + use std::collections::{BTreeMap, BTreeSet}; + + use mz_catalog::memory::objects::CatalogItem; + use mz_cluster_client::ReplicaId; + use mz_controller_types::ClusterId; + use mz_repr::GlobalId; + use mz_sql::plan::validate_metric_sink_prefix; + + use crate::catalog::Catalog; + use crate::coord::metric_sink::{ + CURATED, CuratedMetricSink, InstalledMetricSink, ensure_reads_only_logs, + metric_sinks_on_replica, + }; + + /// `drop_metric_sinks` relies on this range scan returning exactly one replica's installs, with + /// no bleed into a neighbouring replica's contiguous range. + #[mz_ore::test] + fn metric_sinks_on_replica_scans_one_replica() { + let cluster = ClusterId::user(1).expect("valid cluster id"); + let install = |sink_id| InstalledMetricSink { + cluster_id: cluster, + sink_id: GlobalId::Transient(sink_id), + }; + let r = ReplicaId::User; + + let mut sinks = BTreeMap::new(); + sinks.insert((r(1), "a"), install(10)); + sinks.insert((r(2), "a"), install(20)); + sinks.insert((r(2), "b"), install(21)); + sinks.insert((r(2), "c"), install(22)); + sinks.insert((r(4), "a"), install(40)); + + // A replica with several installs: all of them, in key order, and nothing from r(1)/r(4). + assert_eq!( + metric_sinks_on_replica(&sinks, r(2)), + vec![ + ("a", cluster, GlobalId::Transient(20)), + ("b", cluster, GlobalId::Transient(21)), + ("c", cluster, GlobalId::Transient(22)), + ] + ); + // First and last replicas in the map: the scan stops at each boundary. + assert_eq!( + metric_sinks_on_replica(&sinks, r(1)), + vec![("a", cluster, GlobalId::Transient(10))] + ); + assert_eq!( + metric_sinks_on_replica(&sinks, r(4)), + vec![("a", cluster, GlobalId::Transient(40))] + ); + // A replica with no installs, whether ordered between present ones (the r(3) gap) or past + // the end, returns nothing rather than the next replica's range. + assert!(metric_sinks_on_replica(&sinks, r(3)).is_empty()); + assert!(metric_sinks_on_replica(&sinks, r(5)).is_empty()); + } + + /// Every curated definition's prefix must satisfy the same contract a user's `PREFIX` does. + /// Unlike the user path, nothing validates a curated prefix at runtime before this guards it, + /// so a malformed one would escape the reserved lane or break the shaping's name validation. + #[mz_ore::test] + fn curated_prefixes_are_valid() { + for definition in CURATED { + validate_metric_sink_prefix(definition.prefix).unwrap_or_else(|err| { + panic!( + "curated metric sink {:?} has an invalid prefix {:?}: {err}", + definition.name, definition.prefix + ) + }); + } + } + + /// The registry is keyed on the name, so a duplicate would make one definition's install + /// unreachable to teardown and both collectors collide on the `sink` label. Guarded at runtime + /// (`metric_sink_finish`) too, but caught here at build time before it can ship. + #[mz_ore::test] + fn curated_names_are_unique() { + let mut seen = BTreeSet::new(); + for definition in CURATED { + assert!( + seen.insert(definition.name), + "duplicate curated metric sink name {:?}", + definition.name + ); + } + } + + /// Every curated definition must plan against the system catalog. A definition that does not + /// would soft-panic at boot, so catch it here as a failing test instead. `CURATED` is empty + /// today, so this iterates nothing until the first definition lands. + #[mz_ore::test(tokio::test)] + #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `TLS_client_method` + async fn curated_definitions_plan() { + Catalog::with_debug(|catalog| async move { + let session_catalog = catalog.for_system_session(); + for definition in CURATED { + definition + .plan_source(&session_catalog) + .unwrap_or_else(|err| { + panic!( + "curated metric sink {:?} does not plan: {err}", + definition.name + ) + }); + } + }) + .await + } + + /// The five canonical columns, no finishing: the shape a definition must produce. + const VALID_SOURCE: &str = "SELECT 'n'::text AS metric_name, 'gauge'::text AS metric_type, \ + NULL::map[text=>text] AS labels, NULL::double AS value, 'h'::text AS help"; + + /// `VALID_SOURCE` with an ORDER BY appended. Maintained-lifetime planning folds it away rather + /// than rejecting it, since ordering has no meaning for a continuously-consumed collection. + const ORDERED_SOURCE: &str = "SELECT 'n'::text AS metric_name, 'gauge'::text AS metric_type, \ + NULL::map[text=>text] AS labels, NULL::double AS value, 'h'::text AS help ORDER BY 1"; + + /// `plan_source` accepts the canonical column contract (including a source with a finishing, + /// which maintained-lifetime planning folds in) and rejects a source missing the columns. + #[mz_ore::test(tokio::test)] + #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `TLS_client_method` + async fn plan_source_enforces_the_metric_sink_contract() { + Catalog::with_debug(|catalog| async move { + let session_catalog = catalog.for_system_session(); + let plan = |source_sql: &'static str| { + CuratedMetricSink { + name: "test", + source_sql, + prefix: "mz_metric_sink_test_", + } + .plan_source(&session_catalog) + }; + + assert!(plan(VALID_SOURCE).is_ok()); + + // An ORDER BY is folded away by maintained-lifetime planning, not rejected. + assert!(plan(ORDERED_SOURCE).is_ok()); + + // Missing the canonical columns: rejected by `validate_metric_sink_desc`. + assert!(plan("SELECT 1 AS foo").is_err()); + + // Not exactly one statement: rejected by the explicit count guard. + assert!(plan("").is_err()); + assert!(plan("SELECT 1; SELECT 2").is_err()); + }) + .await + } + + /// A SQL-implemented builtin hides its reads: `pg_get_viewdef`'s body reads + /// `mz_catalog.mz_views`, which the dataflow imports but the statement's resolved ids omit. + /// `plan_source` must surface those reads so the gate rejects them. + #[mz_ore::test(tokio::test)] + #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `TLS_client_method` + async fn ensure_reads_only_logs_sees_sql_impl_function_reads() { + Catalog::with_debug(|catalog| async move { + let session_catalog = catalog.for_system_session(); + let (_, _, dependencies) = CuratedMetricSink { + name: "test", + source_sql: "SELECT pg_get_viewdef('x') AS metric_name, 'gauge'::text AS metric_type, \ + NULL::map[text=>text] AS labels, NULL::double AS value, 'h'::text AS help", + prefix: "mz_metric_sink_test_", + } + .plan_source(&session_catalog) + .expect("plans against the system catalog"); + assert!(ensure_reads_only_logs(&catalog, &dependencies).is_err()); + }) + .await + } + + /// The introspection-only contract: a log dependency is accepted, a storage-backed one is + /// rejected. Checked against what the definition reads, so the verdict does not depend on the + /// target cluster's index layout. + #[mz_ore::test(tokio::test)] + #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `TLS_client_method` + async fn ensure_reads_only_logs_accepts_logs_rejects_storage() { + Catalog::with_debug(|catalog| async move { + let log_id = catalog + .entries() + .find(|e| matches!(e.item(), CatalogItem::Log(_))) + .expect("debug catalog has a builtin log") + .id(); + assert!(ensure_reads_only_logs(&catalog, &BTreeSet::from([log_id])).is_ok()); + + let storage_id = catalog + .entries() + .find(|e| matches!(e.item(), CatalogItem::Table(_) | CatalogItem::Source(_))) + .expect("debug catalog has a builtin table or source") + .id(); + assert!(ensure_reads_only_logs(&catalog, &BTreeSet::from([storage_id])).is_err()); + }) + .await + } +} diff --git a/src/adapter/src/coord/sequencer/inner/create_metric_sink.rs b/src/adapter/src/coord/sequencer/inner/create_metric_sink.rs index e476fe3f47328..896a6ddf2173f 100644 --- a/src/adapter/src/coord/sequencer/inner/create_metric_sink.rs +++ b/src/adapter/src/coord/sequencer/inner/create_metric_sink.rs @@ -136,6 +136,10 @@ impl Coordinator { .override_from(&self.catalog.get_cluster(cluster_id).config.features()) .override_from(&self.cluster_scoped_optimizer_overrides(cluster_id)); let optimizer_features = optimizer_config.features.clone(); + let debug_name = self + .catalog() + .resolve_full_name(&plan.name, None) + .to_string(); // Build an optimizer for this METRIC SINK. let mut optimizer = optimize::metric_sink::Optimizer::new( @@ -152,9 +156,10 @@ impl Coordinator { move || { span.in_scope(|| { let metric_sink = optimize::metric_sink::MetricSink::new( - plan.name.clone(), - plan.metric_sink.from, + debug_name, + optimize::metric_sink::MetricSinkFrom::Id(plan.metric_sink.from), plan.metric_sink.prefix.clone(), + None, ); // MIR ⇒ MIR optimization (global) diff --git a/src/adapter/src/optimize/metric_sink.rs b/src/adapter/src/optimize/metric_sink.rs index 7baa79e90ead9..1e5e247217056 100644 --- a/src/adapter/src/optimize/metric_sink.rs +++ b/src/adapter/src/optimize/metric_sink.rs @@ -29,8 +29,8 @@ use mz_repr::explain::trace_plan; use mz_repr::{ ColumnName, Datum, GlobalId, RelationDesc, ReprRelationType, ReprScalarType, Row, SqlScalarType, }; -use mz_sql::names::QualifiedItemName; use mz_sql::optimizer_metrics::OptimizerMetrics; +use mz_sql::plan::{HirRelationExpr, HirToMirConfig}; use mz_transform::TransformCtx; use mz_transform::dataflow::DataflowMetainfo; use mz_transform::normalize_lets::normalize_lets; @@ -51,11 +51,13 @@ use crate::optimize::{ /// operator's hot path. const METRIC_NAME_PATTERN: &str = "^[a-zA-Z_:][a-zA-Z0-9_:]*$"; -/// Optimizer for `CREATE METRIC SINK`. +/// Optimizer for metric sinks, both `CREATE METRIC SINK` and the coordinator-installed curated +/// sinks. /// -/// Like `CREATE INDEX`, the pipeline starts directly from the `GlobalId` of the collection to -/// export rather than lowering a new relational expression from HIR. Unlike a materialized view -/// sink, there is no persist shard, so there is no storage-metadata stage. +/// The source is either an existing collection (like `CREATE INDEX`, no HIR to lower) or a planned +/// query (like a materialized view), see [`MetricSinkFrom`]. Either way the row-wise shaping is +/// appended in MIR and the dataflow exports a single `MetricSink`. Unlike a materialized view sink +/// there is no persist shard, so there is no storage-metadata stage. pub struct Optimizer { /// A representation typechecking context to use throughout the optimizer pipeline. typecheck_ctx: SharedTypecheckingContext, @@ -100,20 +102,51 @@ impl Optimizer { /// A wrapper of metric sink parts needed to start the optimization process. pub struct MetricSink { - name: QualifiedItemName, - from: GlobalId, - /// Prepended to every row's `metric_name` to form the published name. Validated at plan time - /// as a valid start of a Prometheus metric name. + /// Names the assembled dataflow, for debugging. + debug_name: String, + /// The collection whose rows the sink exports. + from: MetricSinkFrom, + /// Prepended to every row's `metric_name` to form the published name. Validated as a valid start + /// of a Prometheus metric name before it reaches here (plan time for a user sink, install time + /// for a curated one). prefix: String, + /// Value for the `sink` label on the sink's health gauges. `None` defaults to the sink's + /// `GlobalId`, which is what a user sink wants. A curated sink passes its stable name. + label: Option, } impl MetricSink { /// Construct a new [`MetricSink`]. Arguments are recorded as-is. - pub fn new(name: QualifiedItemName, from: GlobalId, prefix: String) -> Self { - Self { name, from, prefix } + pub fn new( + debug_name: String, + from: MetricSinkFrom, + prefix: String, + label: Option, + ) -> Self { + Self { + debug_name, + from, + prefix, + label, + } } } +/// Where a metric sink's rows come from. +/// +/// Either way the source must expose the canonical metric-sink columns (see +/// [`shape_metric_sink_source`]). +pub enum MetricSinkFrom { + /// An existing catalog collection, as `CREATE METRIC SINK ... FROM ` resolves to. + Id(GlobalId), + /// A planned query, as a coordinator-installed sink built from curated SQL uses. The query is + /// not a catalog item, so it is lowered and locally optimized here rather than imported. + Query { + expr: HirRelationExpr, + desc: RelationDesc, + }, +} + /// The (sealed intermediate) result after embedding a [`MetricSink`] into a /// [`MirDataflowDescription`], inlining referenced views, and jointly optimizing the `MIR` plans. #[derive(Clone, Debug)] @@ -148,30 +181,37 @@ impl Optimize for Optimizer { fn optimize(&mut self, metric_sink: MetricSink) -> Result { let time = Instant::now(); - let from_entry = self.catalog.get_entry(&metric_sink.from); - let full_name = self - .catalog - .resolve_full_name(&metric_sink.name, from_entry.conn_id()); - let from_desc = from_entry - .relation_desc() - .expect("can only create a metric sink on items with a valid description") - .into_owned(); - let mut df_builder = { let compute = self.compute_instance.clone(); DataflowBuilder::new(&*self.catalog, compute).with_config(&self.config) }; - let mut df_desc = MirDataflowDescription::new(full_name.to_string()); + let mut df_desc = MirDataflowDescription::new(metric_sink.debug_name); let mut df_meta = DataflowMetainfo::default(); - df_builder.import_into_dataflow(&metric_sink.from, &mut df_desc, &self.config.features)?; - df_builder.maybe_reoptimize_imported_views(&mut df_desc, &self.config)?; + let (source_expr, source_desc) = match metric_sink.from { + MetricSinkFrom::Id(from) => { + let from_desc = self + .catalog + .get_entry(&from) + .relation_desc() + .expect("can only create a metric sink on items with a valid description") + .into_owned(); + let repr_typ = ReprRelationType::from(from_desc.typ()); + (MirRelationExpr::global_get(from, repr_typ), from_desc) + } + MetricSinkFrom::Query { expr, desc } => { + // HIR ⇒ MIR lowering and decorrelation. The result is inlined under the shaping + // below rather than becoming its own build, so the whole source is one view. + let expr = expr.lower(HirToMirConfig::from(&self.config), Some(&self.metrics))?; + (expr, desc) + } + }; // Push the pure row-wise shaping (coalesce identity elements, classify the metric kind, // validate the metric name) into MIR, so the operator only does the cross-row logic // (dedup/collision/family-conflict) that needs the fold. See `shape_metric_sink_source`. let (shaped_expr, shaped_desc) = - shape_metric_sink_source(metric_sink.from, &from_desc, &metric_sink.prefix); + shape_metric_sink_source(source_expr, &source_desc, &metric_sink.prefix); let mut local_ctx = TransformCtx::local( &self.config.features, &self.typecheck_ctx, @@ -181,6 +221,8 @@ impl Optimize for Optimizer { ); let shaped_expr = optimize_mir_local(shaped_expr, &mut local_ctx)?; + // Imports the source's dependencies (the `Id` variant's collection, or the query's leaf + // collections) before inserting the shaped view that reads them. df_builder.import_view_into_dataflow( &self.view_id, &shaped_expr, @@ -192,7 +234,11 @@ impl Optimize for Optimizer { let sink_description = ComputeSinkDesc { from: self.view_id, from_desc: shaped_desc, - connection: ComputeSinkConnection::MetricSink(MetricSinkConnection {}), + connection: ComputeSinkConnection::MetricSink(MetricSinkConnection { + label: metric_sink + .label + .unwrap_or_else(|| self.sink_id.to_string()), + }), with_snapshot: true, up_to: Antichain::new(), non_null_assertions: Vec::new(), @@ -266,8 +312,8 @@ impl GlobalLirPlan { } } -/// Extends the metric sink's imported relation with the row-wise shaping the operator otherwise -/// has to do in Rust: prepends the user `prefix` to `metric_name` to form the published name, +/// Extends the metric sink's source expression with the row-wise shaping the operator otherwise +/// has to do in Rust: prepends the configured `prefix` to `metric_name` to form the published name, /// coalesces `labels`/`help` to their identity element, and adds two columns the operator reads /// instead of parsing strings on its hot path: /// @@ -288,16 +334,16 @@ impl GlobalLirPlan { /// full move is deferred: the tiebreak fidelity that logic needs is easier to keep correct /// hand-written and unit-tested for now. fn shape_metric_sink_source( - from_id: GlobalId, - from_desc: &RelationDesc, + source: MirRelationExpr, + source_desc: &RelationDesc, prefix: &str, ) -> (MirRelationExpr, RelationDesc) { - // Precondition: the source relation exposes the canonical metric-sink columns (`metric_name`, - // `metric_type`, `labels`, `value`, `help`). `CREATE METRIC SINK` planning enforces this (see - // `validate_metric_sink_desc` in `mz_sql::plan::statement::ddl`), so a missing column here is - // a planner bug, not user error. + // Precondition: `source_desc` describes `source` and exposes the canonical metric-sink columns + // (`metric_name`, `metric_type`, `labels`, `value`, `help`). + // `mz_sql::plan::validate_metric_sink_desc` enforces this for both `CREATE METRIC SINK` and + // the coordinator-installed curated sinks, so a missing column here is a caller bug. let get_idx = |name: &str| { - from_desc + source_desc .get_by_name(&ColumnName::from(name)) .expect("metric-sink source relation must expose the canonical columns") }; @@ -307,8 +353,12 @@ fn shape_metric_sink_source( let (value_idx, value_ct) = get_idx("value"); let (help_idx, help_ct) = get_idx("help"); - let repr_typ = ReprRelationType::from(from_desc.typ()); - let arity = repr_typ.column_types.len(); + let arity = source_desc.typ().columns().len(); + // The mapped columns are appended at `arity + N` and the `Project` indexes into `source` by + // position, so `source` must have exactly the arity `source_desc` describes. Guaranteed by the + // callers (a trivial finishing over the planned query, or a direct `Get` of the source), but a + // mismatch would silently read the wrong columns, so assert it here. + mz_ore::soft_assert_eq_or_log!(source.arity(), arity); let labels_repr_type = ReprScalarType::from(&labels_ct.scalar_type); let empty_map_row = { @@ -345,8 +395,9 @@ fn shape_metric_sink_source( ), ); - // The published name is `prefix + metric_name`. Planning requires the prefix to start with the - // reserved marker (see `validate_metric_sink_prefix`), so every published family lands in the + // The published name is `prefix + metric_name`. The prefix is validated to start with the + // reserved marker (see `validate_metric_sink_prefix`, run at plan time for a user sink and at + // install time for a curated one), so every published family lands in the // `mz_metric_sink_` lane nothing else in the process registry writes. `TextConcat` (the `||` // operator) propagates nulls, so a null `metric_name` stays null and is skipped, never // published as the bare prefix. @@ -371,7 +422,7 @@ fn shape_metric_sink_source( func::IsRegexpMatchCaseSensitive, )); - let shaped_expr = MirRelationExpr::global_get(from_id, repr_typ) + let shaped_expr = source .map(vec![ labels_coalesced, help_coalesced, @@ -421,8 +472,8 @@ mod tests { VersionedRelationDesc, }; use mz_sql::names::{ - FullItemName, ItemQualifiers, RawDatabaseSpecifier, ResolvedDatabaseSpecifier, ResolvedIds, - SchemaId, SchemaSpecifier, + FullItemName, ItemQualifiers, QualifiedItemName, RawDatabaseSpecifier, + ResolvedDatabaseSpecifier, ResolvedIds, SchemaId, SchemaSpecifier, }; use mz_sql::session::vars::SystemVars; @@ -449,10 +500,15 @@ mod tests { .finish() } + /// A bare `Get` of `TABLE_GID`, the source expression the `MetricSinkFrom::Id` path shapes. + fn source_get(desc: &RelationDesc) -> MirRelationExpr { + MirRelationExpr::global_get(TABLE_GID, ReprRelationType::from(desc.typ())) + } + #[mz_ore::test] fn shaped_desc_column_contract() { let (_expr, desc) = - shape_metric_sink_source(GlobalId::Transient(0), &source_desc(), "app_"); + shape_metric_sink_source(source_get(&source_desc()), &source_desc(), "app_"); let cols: Vec<(String, SqlColumnType)> = desc .iter() @@ -499,7 +555,7 @@ mod tests { #[mz_ore::test] fn shaped_expr_projects_seven_columns() { let (expr, _desc) = - shape_metric_sink_source(GlobalId::Transient(0), &source_desc(), "app_"); + shape_metric_sink_source(source_get(&source_desc()), &source_desc(), "app_"); // The shaping is a `Map` of five new columns followed by a `Project` down to the seven // canonical columns. @@ -514,7 +570,7 @@ mod tests { /// The five scalars the shaping `Map` appends, in order: /// `[labels_coalesced, help_coalesced, prefixed_name, metric_kind, name_valid]`. fn shaped_map_scalars(desc: &RelationDesc, prefix: &str) -> Vec { - let (expr, _desc) = shape_metric_sink_source(GlobalId::Transient(0), desc, prefix); + let (expr, _desc) = shape_metric_sink_source(source_get(desc), desc, prefix); match expr { MirRelationExpr::Project { input, .. } => match *input { MirRelationExpr::Map { scalars, .. } => scalars, @@ -684,41 +740,43 @@ mod tests { } } - /// The assembled dataflow exports exactly one `MetricSink`, reading the shaped view rather - /// than the source relation directly. - #[mz_ore::test] - fn optimizer_exports_one_metric_sink() { + const VIEW_GID: GlobalId = GlobalId::Transient(1); + + /// Runs the whole pipeline over `from` and returns the assembled dataflow. + fn optimize_from(from: MetricSinkFrom, metric_label: Option) -> LirDataflowDescription { let catalog = Arc::new(SingleTableCatalog::new()); let cluster_id = ClusterId::user(1).expect("valid cluster id"); let compute_instance = ComputeInstanceSnapshot::new_without_collections(cluster_id); - let view_id = GlobalId::Transient(1); let config = OptimizerConfig::from(&SystemVars::default()); let metrics = OptimizerMetrics::register_into(&MetricsRegistry::new(), Duration::MAX); let mut optimizer = Optimizer::new( catalog, compute_instance, - view_id, + VIEW_GID, SINK_GID, config, metrics, ); - let name = QualifiedItemName { - qualifiers: ItemQualifiers { - database_spec: ResolvedDatabaseSpecifier::Ambient, - schema_spec: SchemaSpecifier::Id(SchemaId::User(1)), - }, - item: "s".to_string(), - }; let global_mir_plan = optimizer - .optimize(MetricSink::new(name, TABLE_GID, "app_".to_string())) + .optimize(MetricSink::new( + "metric-sink-test".to_string(), + from, + "app_".to_string(), + metric_label, + )) .expect("MIR optimization succeeds"); let global_lir_plan = optimizer .optimize(global_mir_plan) .expect("LIR optimization succeeds"); + let (df_desc, _df_meta) = global_lir_plan.unapply(); + df_desc + } - let df_desc = global_lir_plan.df_desc(); + /// Asserts the dataflow exports exactly one `MetricSink` over the shaped view, whose desc + /// carries the operator's column contract. + fn assert_one_shaped_metric_sink_export(df_desc: &LirDataflowDescription) { assert!(df_desc.index_exports.is_empty()); let sink_exports: Vec<_> = df_desc.sink_exports.iter().collect(); assert_eq!(sink_exports.len(), 1); @@ -728,8 +786,7 @@ mod tests { sink_desc.connection, ComputeSinkConnection::MetricSink(_) )); - // The sink reads the shaped view, whose desc carries the operator's column contract. - assert_eq!(sink_desc.from, view_id); + assert_eq!(sink_desc.from, VIEW_GID); let shaped_names: Vec<&str> = sink_desc .from_desc .iter_names() @@ -748,4 +805,68 @@ mod tests { ] ); } + + /// The `sink` label carried by the export's connection. + fn sink_label(df_desc: &LirDataflowDescription) -> &str { + match &df_desc + .sink_exports + .values() + .next() + .expect("one export") + .connection + { + ComputeSinkConnection::MetricSink(conn) => &conn.label, + other => panic!("expected a metric sink connection, got {other:?}"), + } + } + + #[mz_ore::test] + fn optimizer_exports_one_metric_sink() { + let df_desc = optimize_from(MetricSinkFrom::Id(TABLE_GID), None); + assert_one_shaped_metric_sink_export(&df_desc); + // The source collection is imported, not rebuilt: the only build is the shaped view. + assert!(df_desc.source_imports.contains_key(&TABLE_GID)); + let build_ids: Vec<_> = df_desc.objects_to_build.iter().map(|b| b.id).collect(); + assert_eq!(build_ids, vec![VIEW_GID]); + } + + /// The `Query` source path (what a coordinator-installed curated sink takes) assembles the + /// same shape, with the query lowered under the shaping instead of a `Get` of a catalog item. + #[mz_ore::test] + fn optimizer_shapes_a_query_source() { + let desc = source_desc(); + // The simplest query over the canonical columns. Building richer HIR by hand buys nothing: + // what is under test is that a query source is lowered and shaped, not the lowering itself. + let expr = HirRelationExpr::Get { + id: mz_expr::Id::Global(TABLE_GID), + typ: desc.typ().clone(), + }; + + let df_desc = optimize_from( + MetricSinkFrom::Query { + expr, + desc: desc.clone(), + }, + None, + ); + assert_one_shaped_metric_sink_export(&df_desc); + // The query's leaf collection is imported by the shaped view's dependency walk. + assert!(df_desc.source_imports.contains_key(&TABLE_GID)); + let build_ids: Vec<_> = df_desc.objects_to_build.iter().map(|b| b.id).collect(); + assert_eq!(build_ids, vec![VIEW_GID]); + } + + /// With no explicit label a sink is tagged by its `GlobalId`, what a user's `CREATE METRIC + /// SINK` relies on. An explicit label (a curated sink's stable name) is used verbatim. + #[mz_ore::test] + fn metric_sink_label_defaults_to_sink_id_else_override() { + let df_desc = optimize_from(MetricSinkFrom::Id(TABLE_GID), None); + assert_eq!(sink_label(&df_desc), SINK_GID.to_string()); + + let df_desc = optimize_from( + MetricSinkFrom::Id(TABLE_GID), + Some("mz_curated".to_string()), + ); + assert_eq!(sink_label(&df_desc), "mz_curated"); + } } diff --git a/src/clusterd-test-driver/src/dataflow.rs b/src/clusterd-test-driver/src/dataflow.rs index b0218ad7fe8ff..20d4c9bfc6b10 100644 --- a/src/clusterd-test-driver/src/dataflow.rs +++ b/src/clusterd-test-driver/src/dataflow.rs @@ -403,7 +403,9 @@ impl DataflowBuilder { let desc = ComputeSinkDesc { from: from_id, from_desc, - connection: ComputeSinkConnection::MetricSink(MetricSinkConnection {}), + connection: ComputeSinkConnection::MetricSink(MetricSinkConnection { + label: sink_id.to_string(), + }), with_snapshot: true, up_to: Antichain::new(), non_null_assertions: vec![], @@ -1048,7 +1050,7 @@ mod tests { // The metric sink carries a payload-free connection and no storage metadata. assert!(matches!( sink.connection, - ComputeSinkConnection::MetricSink(MetricSinkConnection {}) + ComputeSinkConnection::MetricSink(MetricSinkConnection { .. }) )); } diff --git a/src/compute-client/src/controller.rs b/src/compute-client/src/controller.rs index 405cb3aa417fa..9da9f289e5348 100644 --- a/src/compute-client/src/controller.rs +++ b/src/compute-client/src/controller.rs @@ -810,10 +810,10 @@ impl ComputeController { /// Creates the described dataflow and initializes state for its output. /// - /// Only sink exports are allowed to have a `target_replica`: materialized views and subscribes. - /// Metric sinks are sink exports too, and nothing here forbids targeting one, but by caller - /// convention they always pass `target_replica: None` so each replica renders the sink into its - /// own registry for per-replica introspection. + /// Only sink exports are allowed to have a `target_replica`: materialized views, subscribes, + /// and metric sinks. A user's `CREATE METRIC SINK` runs untargeted, so every replica renders it + /// into its own registry. The coordinator's curated metric sinks are installed per replica and + /// do target one, so each replica's series are attributable to it. /// /// Panics if called with a dataflow description that has index exports /// when `target_replica` is set. diff --git a/src/compute-types/src/sinks.rs b/src/compute-types/src/sinks.rs index 162110a3526e6..c9444aeec4103 100644 --- a/src/compute-types/src/sinks.rs +++ b/src/compute-types/src/sinks.rs @@ -79,10 +79,14 @@ pub struct SubscribeSinkConnection { /// Connection for a sink that publishes rows into the in-process Prometheus metrics registry. /// -/// Carries no payload: the identity of the metric to update is the sink's `GlobalId`, and the -/// sink does not write to persist, so there is no storage metadata to parameterize over. +/// The sink does not write to persist, so there is no storage metadata to parameterize over. #[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)] -pub struct MetricSinkConnection {} +pub struct MetricSinkConnection { + /// Value of the `sink` label on the sink's health gauges. A user sink passes its `GlobalId`, + /// which is durable. A coordinator-installed curated sink passes its definition name, because + /// its `GlobalId` is transient and would churn the label on every boot. + pub label: String, +} /// Connection attributes required to do a oneshot copy to s3. #[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)] diff --git a/src/compute/src/sink/metric_sink.rs b/src/compute/src/sink/metric_sink.rs index 355e74cf7c3a3..725cb1a480e16 100644 --- a/src/compute/src/sink/metric_sink.rs +++ b/src/compute/src/sink/metric_sink.rs @@ -93,7 +93,7 @@ impl<'scope> SinkRender<'scope> for MetricSinkConnection { // So registration collides only on a genuine logic error, where the soft-panic is // the intended backstop. let drop_handle = (worker_id == active_worker_id).then(|| { - let collector = SinkCollector::new(sink_id, Arc::clone(&state)); + let collector = SinkCollector::new(&self.label, Arc::clone(&state)); compute_state .metrics_registry .register_collector_with_dropper(collector) @@ -680,9 +680,9 @@ struct SinkCollector { } impl SinkCollector { - fn new(sink_id: GlobalId, state: Arc>) -> Self { + fn new(label: &str, state: Arc>) -> Self { let gauge = |name: &str, help: &str| { - Gauge::with_opts(Opts::new(name, help).const_label("sink", sink_id.to_string())) + Gauge::with_opts(Opts::new(name, help).const_label("sink", label)) .expect("static metric sink companion gauge options are valid") }; SinkCollector { @@ -1157,4 +1157,29 @@ mod tests { st.publish_if_healthy(); assert_eq!(st.published[&key_m()].0, 9.0); } + + /// Every companion gauge carries the label the collector was built with. A curated sink passes + /// its stable name here, so its health series stay identifiable across boots even though the + /// sink's `GlobalId` is transient. + #[mz_ore::test] + fn collector_labels_gauges_with_sink_label() { + let state = Arc::new(Mutex::new(SinkState::default())); + let collector = SinkCollector::new("mz_curated_example", state); + + let families = collector.collect(); + // The point is that every emitted family carries the label, so guard only that there is + // something to check. Pinning the exact count churns on every added gauge without telling + // the next reader whether they broke labelling or just added a family. + assert!(!families.is_empty()); + for family in &families { + for metric in family.get_metric() { + let sink = metric + .get_label() + .iter() + .find(|l| l.name() == "sink") + .expect("sink label present"); + assert_eq!(sink.value(), "mz_curated_example"); + } + } + } } diff --git a/src/sql/src/plan.rs b/src/sql/src/plan.rs index 15355536ae858..f883199ac6204 100644 --- a/src/sql/src/plan.rs +++ b/src/sql/src/plan.rs @@ -118,7 +118,7 @@ pub use side_effecting_func::SideEffectingFunc; pub use statement::ddl::{ AlterSourceAddSubsourceOptionExtracted, MySqlConfigOptionExtracted, PgConfigOptionExtracted, PlannedAlterRoleOption, PlannedRoleAttributes, PlannedRoleVariable, - SqlServerConfigOptionExtracted, + SqlServerConfigOptionExtracted, validate_metric_sink_desc, validate_metric_sink_prefix, }; pub use statement::{ StatementClassification, StatementContext, StatementDesc, describe, plan, plan_copy_from, diff --git a/src/sql/src/plan/statement/ddl.rs b/src/sql/src/plan/statement/ddl.rs index e8dbf023512f3..bd64cb4118f84 100644 --- a/src/sql/src/plan/statement/ddl.rs +++ b/src/sql/src/plan/statement/ddl.rs @@ -4259,12 +4259,16 @@ generate_extracted_config!(CreateMetricSinkOption, (Prefix, String)); const METRIC_SINK_PREFIX_MARKER: &str = "mz_metric_sink_"; /// Rejects a prefix that could not start a Prometheus metric name, or that escapes the reserved -/// `mz_metric_sink_` lane (see [`METRIC_SINK_PREFIX_MARKER`]). +/// `mz_metric_sink_` lane (see `METRIC_SINK_PREFIX_MARKER`). /// /// The sink prepends this to every name it publishes, so `prefix + name` must stay a legal /// family name (`[a-zA-Z_:][a-zA-Z0-9_:]*`, the same grammar the runtime checks each row's /// `metric_name` against). The prefix must therefore be at least one character long. -fn validate_metric_sink_prefix(prefix: &str) -> Result<(), PlanError> { +/// +/// Enforced for a user's `CREATE METRIC SINK` at plan time, and for a coordinator-installed curated +/// sink at install time. Both paths depend on the guarantees this gives the row shaping: the +/// reserved leading character is what lets a bare `metric_name` start with a digit or be empty. +pub fn validate_metric_sink_prefix(prefix: &str) -> Result<(), PlanError> { if prefix.is_empty() { return Err(sql_err!("metric sink prefix must not be empty")); } @@ -4291,7 +4295,12 @@ fn validate_metric_sink_prefix(prefix: &str) -> Result<(), PlanError> { Ok(()) } -fn validate_metric_sink_desc(desc: &RelationDesc) -> Result<(), PlanError> { +/// Checks that `desc` exposes the canonical metric-sink columns, the contract +/// `mz_adapter::optimize::metric_sink`'s row shaping and the compute-side operator both rely on. +/// +/// Every metric-sink source has to pass this, whether it is the `FROM` relation of a +/// `CREATE METRIC SINK` or the query behind a coordinator-installed curated sink. +pub fn validate_metric_sink_desc(desc: &RelationDesc) -> Result<(), PlanError> { for (name, type_ok) in METRIC_SINK_SOURCE_COLUMNS { let col = ColumnName::from(*name); let (_, column_type) = desc diff --git a/src/sql/src/session/vars.rs b/src/sql/src/session/vars.rs index f6ad7a9446eb7..2e7d3b18086f0 100644 --- a/src/sql/src/session/vars.rs +++ b/src/sql/src/session/vars.rs @@ -2541,10 +2541,15 @@ pub struct FeatureFlag { } impl FeatureFlag { + /// Returns whether the feature flag is enabled in the provided `system_vars`. + pub fn enabled(&'static self, system_vars: &SystemVars) -> bool { + *system_vars.expect_value::(self.flag) + } + /// Returns an error unless the feature flag is enabled in the provided /// `system_vars`. pub fn require(&'static self, system_vars: &SystemVars) -> Result<(), VarError> { - match *system_vars.expect_value::(self.flag) { + match self.enabled(system_vars) { true => Ok(()), false => Err(VarError::RequiresFeatureFlag { feature_flag: self }), } diff --git a/test/testdrive/metric-sink.td b/test/testdrive/metric-sink.td index cb6f05705bbd3..06498ea31a482 100644 --- a/test/testdrive/metric-sink.td +++ b/test/testdrive/metric-sink.td @@ -22,6 +22,10 @@ # the default timeout is too tight here. $ set-sql-timeout duration=60s +# The churn cluster below takes its size from this default, so multi-size +# testdrive configs can vary it like the rest of the suite. +$ set-arg-default default-replica-size=scale=1,workers=1 + # `mz_cluster_prometheus_metrics` is a per-replica log source, so every read of it # below has to name a replica. The sink runs on each replica of its cluster and # each replica owns its own registry, so targeting one replica is also what keeps @@ -224,3 +228,61 @@ contains:still depended upon by metric sink "s" 0 > DROP TABLE lt CASCADE + +# Replica churn drives the coordinator's curated metric-sink install and +# teardown (`install_metric_sinks` / `drop_metric_sinks`). The curated list is +# empty, so there is no curated series to assert on: what is under test is that +# both hooks run over an empty list without erroring, and that a replica added +# and removed under a live sink leaves that sink publishing on the replica that +# remains. +# +# On its own cluster, so the churn does not disturb the replica serving the rest +# of this file. `quickstart` is a managed cluster anyway, which rejects +# `CREATE CLUSTER REPLICA`, so replication factor is the way to add and remove a +# replica. +> CREATE CLUSTER churn_c (SIZE '${arg.default-replica-size}') + +> CREATE TABLE churn_t (k text NOT NULL, val double) + +> INSERT INTO churn_t VALUES ('a', 1) + +> CREATE VIEW churn_v AS + SELECT 'value'::text AS metric_name, + 'gauge'::text AS metric_type, + map_build(LIST[ROW('k', k)])::map[text=>text] AS labels, + val AS value, + 'a metric emitted across replica churn'::text AS help + FROM churn_t + +> CREATE METRIC SINK s_churn IN CLUSTER churn_c FROM churn_v WITH (PREFIX = 'mz_metric_sink_churn_') + +# Read `churn_c`'s registry, not `quickstart`'s: the relation is per-replica +# introspection, served by the session's active cluster. +> SET cluster = churn_c + +> SELECT count(*) FROM mz_introspection.mz_cluster_prometheus_metrics + WHERE metric_name = 'mz_metric_sink_churn_value' +1 + +> ALTER CLUSTER churn_c SET (REPLICATION FACTOR 2) + +# `cluster_replica = r1` (set at the top of the file) targets churn_c's r1 for +# every read here, so the churn is observable rather than hidden behind +# `UntargetedLogRead`: r1 survives both factor changes, and adding a second replica +# leaves its own registry, and the sink on it, untouched. +> SELECT count(*) FROM mz_introspection.mz_cluster_prometheus_metrics + WHERE metric_name = 'mz_metric_sink_churn_value' +1 + +> ALTER CLUSTER churn_c SET (REPLICATION FACTOR 1) + +# And after the second replica is torn down, r1 is still publishing. +> SELECT count(*) FROM mz_introspection.mz_cluster_prometheus_metrics + WHERE metric_name = 'mz_metric_sink_churn_value' +1 + +> RESET cluster + +> DROP CLUSTER churn_c CASCADE + +> DROP TABLE churn_t CASCADE