diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index 276de9f8288df..1f086a0e000e9 100644 --- a/doc/user/data/metrics.yml +++ b/doc/user/data/metrics.yml @@ -453,28 +453,6 @@ metrics: help: Total number of started transactions. source: src/catalog/src/durable/metrics.rs visibility: internal -- name: mz_check_scheduling_policies_seconds_bucket - help: The time each policy in `check_scheduling_policies` takes. - labels: - - le - - policy - - thread - source: src/adapter/src/metrics.rs - visibility: internal -- name: mz_check_scheduling_policies_seconds_count - help: The time each policy in `check_scheduling_policies` takes. - labels: - - policy - - thread - source: src/adapter/src/metrics.rs - visibility: internal -- name: mz_check_scheduling_policies_seconds_sum - help: The time each policy in `check_scheduling_policies` takes. - labels: - - policy - - thread - source: src/adapter/src/metrics.rs - visibility: internal - name: mz_cluster_handle_command_duration_seconds_bucket help: Time spent in handling commands. labels: @@ -971,25 +949,6 @@ metrics: help: The time it takes to advance the catalog shard upper for a txns-shard write (group commits and table register/forget). source: src/adapter/src/metrics.rs visibility: internal -- name: mz_handle_scheduling_decisions_seconds_bucket - help: The time `handle_scheduling_decisions` takes. - labels: - - altered_a_cluster - - le - source: src/adapter/src/metrics.rs - visibility: internal -- name: mz_handle_scheduling_decisions_seconds_count - help: The time `handle_scheduling_decisions` takes. - labels: - - altered_a_cluster - source: src/adapter/src/metrics.rs - visibility: internal -- name: mz_handle_scheduling_decisions_seconds_sum - help: The time `handle_scheduling_decisions` takes. - labels: - - altered_a_cluster - source: src/adapter/src/metrics.rs - visibility: internal - name: mz_index_peek_cursor_setup_seconds_bucket help: Time setting up cursor and literal constraints. labels: diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 9fb80b0a1b169..18afa831d4b44 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -97,14 +97,9 @@ def get_minimal_system_parameters( "enable_refresh_every_mvs": "true", "enable_replacement_materialized_views": "true", "enable_cluster_schedule_refresh": "true", - # The cluster controller and background ALTER CLUSTER dyncfgs default on - # in current versions. Pin them explicitly so runs against older versions - # (which predate the flags or defaulted them off) exercise the legacy - # paths while current versions exercise the controller owning the - # managed-cluster replica set. - "enable_cluster_controller": ( - "true" if version >= MzVersion.parse_mz("v26.29.0-dev") else "false" - ), + # Pinned explicitly so runs against older versions (which predate the + # flag or defaulted it off) behave like current ones, where it defaults + # on. "enable_background_alter_cluster": ( "true" if version >= MzVersion.parse_mz("v26.29.0-dev") else "false" ), @@ -127,6 +122,15 @@ def get_minimal_system_parameters( if version < MzVersion.parse_mz("v0.163.0-dev"): config["enable_compute_active_dataflow_cancelation"] = "true" + # The cluster controller's break-glass gate. Removed in v26.38, where the + # controller runs unconditionally. Older binaries still read it, and + # defaulted it off before v26.29, so pin it on for them to keep mixed-version + # runs exercising the same path as current versions. + if version < MzVersion.parse_mz("v26.38.0-dev"): + config["enable_cluster_controller"] = ( + "true" if version >= MzVersion.parse_mz("v26.29.0-dev") else "false" + ) + return config diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index c585e9bccb09a..e33e80231b316 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -3270,7 +3270,6 @@ def __init__( "oidc_group_role_sync_strict", "console_oidc_client_id", "console_oidc_scopes", - "enable_cluster_controller", "cluster_controller_tick_interval", "enable_background_alter_cluster", "default_cluster_reconfiguration_timeout", diff --git a/src/adapter-types/src/dyncfgs.rs b/src/adapter-types/src/dyncfgs.rs index 2a370901583f8..327af7a7c464e 100644 --- a/src/adapter-types/src/dyncfgs.rs +++ b/src/adapter-types/src/dyncfgs.rs @@ -353,23 +353,7 @@ pub const ENABLE_SCOPED_SYSTEM_PARAMETERS: Config = Config::new( "Whether per-cluster and per-replica scoped system parameters are evaluated and applied.", ); -/// Top-level gate for the cluster controller. When on, the controller owns the -/// managed-cluster replica set and the legacy paths (the graceful 3-stage -/// machine and `cluster_scheduling.rs`) are bypassed. The replica set cannot -/// have two writers, so this is a clean switch, not a per-strategy toggle. -/// -/// Defaults on. This is the break-glass switch to fall back to the legacy -/// paths if the controller misbehaves. -pub const ENABLE_CLUSTER_CONTROLLER: Config = Config::new( - "enable_cluster_controller", - true, - "Whether the cluster controller owns the managed-cluster replica set. When false, the legacy scheduling and graceful-reconfiguration paths run instead.", -); - /// Cadence of the cluster controller's reconcile tick. -/// -/// Replaces `cluster_check_scheduling_policies_interval` once the controller is -/// the sole owner; while the controller is dark both intervals exist. pub const CLUSTER_CONTROLLER_TICK_INTERVAL: Config = Config::new( "cluster_controller_tick_interval", Duration::from_secs(5), @@ -380,9 +364,6 @@ pub const CLUSTER_CONTROLLER_TICK_INTERVAL: Config = Config::new( /// controller converging in the background, or blocks the session on a /// wait-shim until the reconfiguration completes or its deadline passes. /// -/// Only consulted while [`ENABLE_CLUSTER_CONTROLLER`] is on, when the -/// controller owns the reconfiguration. -/// /// Defaults on. This is the break-glass switch back to the blocking wait-shim /// if returning immediately causes trouble. pub const ENABLE_BACKGROUND_ALTER_CLUSTER: Config = Config::new( @@ -404,9 +385,9 @@ pub const DEFAULT_CLUSTER_RECONFIGURATION_TIMEOUT: Config = Config::ne /// runs a burst replica; graceful reconfiguration and `ON REFRESH` scheduling /// are unaffected. /// -/// Only consulted while [`ENABLE_CLUSTER_CONTROLLER`] is on. A cluster can only -/// carry an `AUTO SCALING STRATEGY` while its SQL acceptance feature flag is -/// on, so this is the second of the two gates burst sits behind. +/// A cluster can only carry an `AUTO SCALING STRATEGY` while its SQL acceptance +/// feature flag is on, so this is the second of the two gates burst sits +/// behind. pub const ENABLE_HYDRATION_BURST: Config = Config::new( "enable_hydration_burst", true, @@ -426,7 +407,6 @@ pub const DEFAULT_HYDRATION_BURST_LINGER: Config = Config::new( pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { configs .add(&ALLOW_USER_SESSIONS) - .add(&ENABLE_CLUSTER_CONTROLLER) .add(&CLUSTER_CONTROLLER_TICK_INTERVAL) .add(&ENABLE_BACKGROUND_ALTER_CLUSTER) .add(&DEFAULT_CLUSTER_RECONFIGURATION_TIMEOUT) diff --git a/src/adapter/src/catalog/open.rs b/src/adapter/src/catalog/open.rs index afe4f8f1729c7..c328b0882f6b8 100644 --- a/src/adapter/src/catalog/open.rs +++ b/src/adapter/src/catalog/open.rs @@ -1231,9 +1231,9 @@ fn reconcile_builtin_cluster_replicas( } // Reading the cluster's factor is what makes this compose with the other - // writers of a replica set. The refresh scheduler parks a scheduled cluster - // by writing its factor to 0, so converging on the factor honors that - // instead of resurrecting a replica the scheduler just dropped. + // writers of a replica set. The controller's on-refresh strategy parks a + // scheduled cluster by writing its factor to 0, so converging on the + // factor honors that instead of resurrecting a replica it just dropped. let mut surplus = replicas_by_cluster.remove(&cluster.id).unwrap_or_default(); for index in 0..managed.replication_factor { let replica_name = managed_cluster_replica_name(index); diff --git a/src/adapter/src/catalog/transact.rs b/src/adapter/src/catalog/transact.rs index c69eb846f998a..947ab6818c43c 100644 --- a/src/adapter/src/catalog/transact.rs +++ b/src/adapter/src/catalog/transact.rs @@ -86,7 +86,6 @@ use crate::catalog::{ use crate::config::{ScopedParameters, ScopedParametersScope}; use crate::coord::ConnMeta; use crate::coord::catalog_implications::parsed_state_updates::ParsedStateUpdate; -use crate::coord::cluster_scheduling::SchedulingDecision; use crate::util::ResultExt; /// A manually injected audit event. @@ -360,9 +359,6 @@ pub enum ReplicaCreateDropReason { /// - ALTERing various options on a managed cluster, /// - CREATE/DROP CLUSTER REPLICA on an unmanaged cluster. Manual, - /// The automated cluster scheduling initiated the replica create or drop, e.g., a - /// materialized view is needing a refresh on a SCHEDULE ON REFRESH cluster. - ClusterScheduling(Vec), /// The cluster controller's graceful-reconfiguration strategy created the replica while /// converging a cluster onto an in-flight `reconfiguration` target (a background /// `ALTER CLUSTER`). @@ -373,11 +369,7 @@ pub enum ReplicaCreateDropReason { /// The cluster controller's on-refresh strategy created the replica for a refresh window on /// a `SCHEDULE = ON REFRESH` cluster. Audited as the `schedule` reason, carrying the tick's /// window decision (which MVs needed a refresh or compaction time, and the hydration-time - /// estimate) as the `scheduling_policies` detail, the same detail the legacy scheduler's - /// [`ReplicaCreateDropReason::ClusterScheduling`] records. Deliberately not that variant - /// itself: its legacy shape carries a per-policy `Vec` and an on/off flag for auditing - /// off-decisions, neither of which the controller has (controller drops are uniformly - /// `Retired`), and it is removed together with the legacy scheduler. + /// estimate) as the `scheduling_policies` detail. OnRefresh(RefreshWindowDecision), /// The cluster controller dropped the replica because the cluster's configuration no longer /// calls for it. The uniform reason on every controller-emitted drop (e.g. a @@ -394,12 +386,6 @@ impl ReplicaCreateDropReason { ) { match self { ReplicaCreateDropReason::Manual => (CreateOrDropClusterReplicaReasonV1::Manual, None), - ReplicaCreateDropReason::ClusterScheduling(scheduling_decisions) => ( - CreateOrDropClusterReplicaReasonV1::Schedule, - Some(SchedulingDecision::reasons_to_audit_log_reasons( - &scheduling_decisions, - )), - ), ReplicaCreateDropReason::GracefulReconfiguration => { (CreateOrDropClusterReplicaReasonV1::Reconfiguration, None) } @@ -416,8 +402,8 @@ impl ReplicaCreateDropReason { } /// Convert the controller's on-refresh window decision into the audit log's -/// `scheduling_policies` detail, the same shape the legacy scheduler records: -/// ids as strings and the hydration-time estimate as an interval string. +/// `scheduling_policies` detail: ids as strings and the hydration-time estimate +/// as an interval string. fn refresh_window_decision_to_audit_log( decision: RefreshWindowDecision, ) -> SchedulingDecisionsWithReasonsV2 { @@ -514,8 +500,8 @@ impl Catalog { /// status change, a fresh record, or the drop of an in-progress record. /// /// Every such movement is an audit-log transition, so a write performing - /// one must declare the matching intent. Status-preserving copies (legacy - /// paths carrying a record forward, re-targets that stay in progress with a + /// one must declare the matching intent. Status-preserving copies (a write + /// carrying a record forward, re-targets that stay in progress with a /// declared `Started`) and drops of already-settled records move nothing. fn reconfiguration_lifecycle_moved( old_config: &ClusterConfig, @@ -3827,8 +3813,8 @@ mod tests { &unmanaged, )); - // Not movements: no record at all, a status-preserving copy (legacy - // paths carry the record forward), and dropping a settled record. + // Not movements: no record at all, a status-preserving copy (a write + // that carries the record forward), and dropping a settled record. assert!(!Catalog::reconfiguration_lifecycle_moved( &managed(None), &managed(None), @@ -3985,9 +3971,8 @@ mod tests { use crate::catalog::ReplicaCreateDropReason; - // `OnRefresh` shares the `schedule` audit word with the legacy - // `ClusterScheduling` variant and converts the controller's window - // decision into the same `scheduling_policies` detail blob: ids as + // `OnRefresh` audits the `schedule` word and converts the controller's + // window decision into the `scheduling_policies` detail blob: ids as // strings, the hydration-time estimate as an interval string, and the // decision hardcoded `on` (the controller produces a create, and so // this detail, only for an open window). diff --git a/src/adapter/src/coord.rs b/src/adapter/src/coord.rs index 66650756c64ef..116ff747c4f41 100644 --- a/src/adapter/src/coord.rs +++ b/src/adapter/src/coord.rs @@ -195,7 +195,6 @@ use crate::coord::appends::{ PendingWriteTxn, }; use crate::coord::caught_up::CaughtUpCheckContext; -use crate::coord::cluster_scheduling::SchedulingDecision; use crate::coord::id_bundle::CollectionIdBundle; use crate::coord::introspection::IntrospectionSubscribe; use crate::coord::peek::PendingPeek; @@ -220,7 +219,6 @@ use crate::{AdapterNotice, ReadHolds, flags}; pub(crate) mod appends; pub(crate) mod catalog_serving; pub(crate) mod cluster_controller; -pub(crate) mod cluster_scheduling; pub(crate) mod consistency; pub(crate) mod id_bundle; pub(crate) mod in_memory_oracle; @@ -446,13 +444,6 @@ pub enum Message { }, DrainStatementLog, PrivateLinkVpcEndpointEvents(Vec), - CheckSchedulingPolicies, - - /// Scheduling policy decisions about turning clusters On/Off. - /// `Vec<(policy name, Vec of decisions by the policy)>` - /// A cluster will be On if and only if there is at least one On decision for it. - /// Scheduling decisions for clusters that have `SCHEDULE = MANUAL` are ignored. - SchedulingDecisions(Vec<(&'static str, Vec<(ClusterId, SchedulingDecision)>)>), /// One pull/apply call from the cluster controller task, answered on the main /// coordinator message loop from the catalog and live controller signals. @@ -560,8 +551,6 @@ impl Message { Message::DrainStatementLog => "drain_statement_log", Message::AlterConnectionValidationReady(..) => "alter_connection_validation_ready", Message::PrivateLinkVpcEndpointEvents(_) => "private_link_vpc_endpoint_events", - Message::CheckSchedulingPolicies => "check_scheduling_policies", - Message::SchedulingDecisions { .. } => "scheduling_decision", Message::ClusterControllerRequest(_) => "cluster_controller_request", Message::DeferredStatementReady => "deferred_statement_ready", } @@ -2134,14 +2123,6 @@ pub struct Coordinator { /// a timestamp oracle backend is configured. timestamp_oracle_config: Option, - /// Periodically asks cluster scheduling policies to make their decisions. - check_cluster_scheduling_policies_interval: Interval, - - /// This keeps the last On/Off decision for each cluster and each scheduling policy. - /// (Clusters that have been dropped or are otherwise out of scope for automatic scheduling are - /// periodically cleaned up from this Map.) - cluster_scheduling_decisions: BTreeMap>, - /// When doing 0dt upgrades/in read-only mode, periodically ask all known /// clusters/collections whether they are caught up. caught_up_check_interval: Interval, @@ -4041,13 +4022,6 @@ impl Coordinator { linearize_reads_notified.set(linearize_reads_notify.notified()); messages.push(Message::LinearizeReads); } - // `tick()` on `Interval` is cancel-safe: - // https://docs.rs/tokio/1.19.2/tokio/time/struct.Interval.html#cancel-safety - // Receive a single command. - _ = self.check_cluster_scheduling_policies_interval.tick() => { - messages.push(Message::CheckSchedulingPolicies); - }, - // `tick()` on `Interval` is cancel-safe: // https://docs.rs/tokio/1.19.2/tokio/time/struct.Interval.html#cancel-safety // Receive a single command. @@ -4928,12 +4902,6 @@ pub fn serve( let coord_now = now.clone(); let advance_timelines_interval = tokio::time::interval(catalog.system_config().default_timestamp_interval()); - let mut check_scheduling_policies_interval = tokio::time::interval( - catalog - .system_config() - .cluster_check_scheduling_policies_interval(), - ); - check_scheduling_policies_interval.set_missed_tick_behavior(MissedTickBehavior::Delay); let clusters_caught_up_check_interval = if read_only_controllers { let dyncfgs = catalog.system_config().dyncfgs(); @@ -5122,8 +5090,6 @@ pub fn serve( statement_logging: StatementLogging::new(coord_now.clone()), webhook_concurrency_limit, timestamp_oracle_config, - check_cluster_scheduling_policies_interval: check_scheduling_policies_interval, - cluster_scheduling_decisions: BTreeMap::new(), caught_up_check_interval: clusters_caught_up_check_interval, caught_up_check: clusters_caught_up_check, installed_watch_sets: BTreeMap::new(), diff --git a/src/adapter/src/coord/cluster_controller.rs b/src/adapter/src/coord/cluster_controller.rs index be53782f19608..6b2803ffbf63d 100644 --- a/src/adapter/src/coord/cluster_controller.rs +++ b/src/adapter/src/coord/cluster_controller.rs @@ -18,19 +18,16 @@ //! signals are pulled on demand, so a tick's round-trips scale with the number of //! managed clusters that need a live signal, not with a constant. //! -//! Everything here is gated by [`ENABLE_CLUSTER_CONTROLLER`] (default on). With -//! the gate off the task does not tick, so the legacy scheduling and graceful -//! paths remain the sole writers of the replica set. With the gate on the -//! controller owns the *user* managed-cluster replica set; the legacy entry -//! points no-op. (System/builtin clusters are excluded here. Their config-implied -//! replicas are materialized by `reconcile_builtin_cluster_replicas` at catalog -//! open, which derives the same target from the same config.) +//! The controller owns the *user* managed-cluster replica set. (System/builtin +//! clusters are excluded here. Their config-implied replicas are materialized +//! by `reconcile_builtin_cluster_replicas` at catalog open, which derives the +//! same target from the same config.) use std::collections::BTreeSet; use std::sync::Arc; use std::time::Duration; -use mz_adapter_types::dyncfgs::{CLUSTER_CONTROLLER_TICK_INTERVAL, ENABLE_CLUSTER_CONTROLLER}; +use mz_adapter_types::dyncfgs::CLUSTER_CONTROLLER_TICK_INTERVAL; use mz_catalog::memory::objects::{ClusterConfig, ClusterVariant}; use mz_cluster_controller::ClusterController; use mz_cluster_controller::ctx::{ @@ -203,13 +200,10 @@ impl ClusterControllerCtx for CoordCtx { impl Coordinator { /// Spawn the cluster controller task. /// - /// The task ticks at [`CLUSTER_CONTROLLER_TICK_INTERVAL`] and reconciles when - /// [`ENABLE_CLUSTER_CONTROLLER`] is on; while the gate is off it ticks but - /// each tick is an early no-op. Both the gate and the interval are re-read - /// each tick (the interval via a [`ClusterControllerRequest::TickInterval`] - /// round-trip), so a runtime change to either takes effect without a restart. - /// It owns the controller and a [`CoordCtx`] that marshals back to this - /// Coordinator. + /// The task ticks at [`CLUSTER_CONTROLLER_TICK_INTERVAL`], re-read each tick + /// via a [`ClusterControllerRequest::TickInterval`] round-trip so a runtime + /// change takes effect without a restart. It owns the controller and a + /// [`CoordCtx`] that marshals back to this Coordinator. /// /// The interval is the fallback cadence: `reconcile_now` cuts the /// sleep short after a catalog transaction changes durable cluster state. @@ -255,22 +249,20 @@ impl Coordinator { /// Handle one [`ClusterControllerRequest`] on the coordinator loop. /// - /// The controller is inactive when the gate is off, or while the deployment - /// is in read-only mode (a 0dt upgrade, where it must not write the catalog). - /// When inactive, reads report no managed clusters (so the controller finds - /// nothing to reconcile) and applies are rejected: the task still wakes each - /// tick and sends one `ManagedClusterIds` request, but that request - /// early-returns here and no catalog state is read or written, so the legacy - /// paths remain the sole writers of the replica set. The task keeps ticking, - /// so the controller reactivates on its own once the deployment promotes out - /// of read-only mode. + /// The controller is inactive while the deployment is in read-only mode (a + /// 0dt upgrade, where it must not write the catalog). When inactive, reads + /// report no managed clusters (so the controller finds nothing to + /// reconcile) and applies are rejected: the task still wakes each tick and + /// sends one `ManagedClusterIds` request, but that request early-returns + /// here and no catalog state is read or written. The task keeps ticking, so + /// the controller reactivates on its own once the deployment promotes out of + /// read-only mode. #[mz_ore::instrument(level = "debug")] pub(crate) async fn handle_cluster_controller_request( &mut self, request: ClusterControllerRequest, ) { - let active = ENABLE_CLUSTER_CONTROLLER.get(self.catalog().system_config().dyncfgs()) - && !self.controller.read_only(); + let active = !self.controller.read_only(); match request { ClusterControllerRequest::ManagedClusterIds { tx } => { @@ -331,8 +323,7 @@ impl Coordinator { // then complete the reply from a spawned task: the oracle // read is a network round-trip (to the Postgres/CRDB-backed // timestamp oracle) and must never run on the serial - // coordinator loop. The legacy `check_refresh_policy` makes - // the same split. + // coordinator loop. match self.refresh_window_catalog_inputs(cluster_id) { None => { let _ = tx.send(None); @@ -511,8 +502,7 @@ impl Coordinator { /// cluster (the system compaction estimate and each bound REFRESH /// materialized view's storage write frontier and refresh schedule), or /// `None` if the cluster is missing, unmanaged, or not scheduled `ON - /// REFRESH`. These are the same signals the legacy `check_refresh_policy` - /// reads. + /// REFRESH`. /// /// The oracle read timestamp completing [`RefreshWindowInputs`] is /// deliberately not fetched here: this runs on the coordinator loop, and @@ -520,8 +510,8 @@ impl Coordinator { /// a spawned task instead. /// /// The MV write frontier is carried through with full fidelity as the - /// `Antichain` the storage controller reports, matching the legacy refresh - /// policy; the on-refresh strategy compares against it directly. + /// `Antichain` the storage controller reports. The on-refresh strategy + /// compares against it directly. fn refresh_window_catalog_inputs( &self, cluster_id: ClusterId, diff --git a/src/adapter/src/coord/cluster_scheduling.rs b/src/adapter/src/coord/cluster_scheduling.rs deleted file mode 100644 index 88c5c9f6872e5..0000000000000 --- a/src/adapter/src/coord/cluster_scheduling.rs +++ /dev/null @@ -1,566 +0,0 @@ -// 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. - -use itertools::Itertools; -use mz_adapter_types::dyncfgs::ENABLE_CLUSTER_CONTROLLER; -use mz_audit_log::SchedulingDecisionsWithReasonsV2; -use mz_catalog::memory::objects::{CatalogItem, ClusterVariant, ClusterVariantManaged}; -use mz_controller_types::ClusterId; -use mz_ore::collections::CollectionExt; -use mz_ore::{soft_assert_or_log, soft_panic_or_log}; -use mz_repr::adt::interval::Interval; -use mz_repr::{GlobalId, TimestampManipulation}; -use mz_sql::catalog::CatalogCluster; -use mz_sql::plan::{AlterClusterPlanStrategy, ClusterSchedule}; -use std::time::{Duration, Instant}; -use tracing::{debug, warn}; - -use crate::AdapterError; -use crate::coord::sequencer::cancel_carried_reconfiguration; -use crate::coord::{Coordinator, Message}; - -const POLICIES: &[&str] = &[REFRESH_POLICY_NAME]; - -const REFRESH_POLICY_NAME: &str = "refresh"; - -/// A policy's decision for whether it wants a certain cluster to be On, along with its reason. -/// (Among the reasons there can be settings of the policy as well as other information about the -/// state of the system.) -#[derive(Clone, Debug)] -pub enum SchedulingDecision { - /// The reason for the refresh policy for wanting to turn a cluster On or Off. - Refresh(RefreshDecision), -} - -impl SchedulingDecision { - /// Extract the On/Off decision from the policy-specific structs. - pub fn cluster_on(&self) -> bool { - match &self { - SchedulingDecision::Refresh(RefreshDecision { cluster_on, .. }) => cluster_on.clone(), - } - } -} - -#[derive(Clone, Debug)] -pub struct RefreshDecision { - /// Whether the ON REFRESH policy wants a certain cluster to be On. - cluster_on: bool, - /// Objects that currently need a refresh on the cluster (taking into account the rehydration - /// time estimate), and therefore should keep the cluster On. - objects_needing_refresh: Vec, - /// Objects for which we estimate that they currently need Persist compaction, and therefore - /// should keep the cluster On. - objects_needing_compaction: Vec, - /// The HYDRATION TIME ESTIMATE setting of the cluster. - hydration_time_estimate: Duration, -} - -impl SchedulingDecision { - pub fn reasons_to_audit_log_reasons<'a, I>(reasons: I) -> SchedulingDecisionsWithReasonsV2 - where - I: IntoIterator, - { - SchedulingDecisionsWithReasonsV2 { - on_refresh: reasons - .into_iter() - .filter_map(|r| match r { - SchedulingDecision::Refresh(RefreshDecision { - cluster_on, - objects_needing_refresh, - objects_needing_compaction, - hydration_time_estimate, - }) => { - soft_assert_or_log!( - !cluster_on - || !objects_needing_refresh.is_empty() - || !objects_needing_compaction.is_empty(), - "`cluster_on = true` should have an explanation" - ); - let mut hydration_time_estimate_str = String::new(); - mz_repr::strconv::format_interval( - &mut hydration_time_estimate_str, - Interval::from_duration(hydration_time_estimate).expect( - "planning ensured that this is convertible back to Interval", - ), - ); - Some(mz_audit_log::RefreshDecisionWithReasonV2 { - decision: (*cluster_on).into(), - objects_needing_refresh: objects_needing_refresh - .iter() - .map(|id| id.to_string()) - .collect(), - objects_needing_compaction: objects_needing_compaction - .iter() - .map(|id| id.to_string()) - .collect(), - hydration_time_estimate: hydration_time_estimate_str, - }) - } - }) - .into_element(), // Each policy should have exactly one opinion on each cluster. - } - } -} - -impl Coordinator { - #[mz_ore::instrument(level = "debug")] - /// Call each scheduling policy. - /// - /// No-ops when the cluster controller owns the replica set - /// ([`ENABLE_CLUSTER_CONTROLLER`]): the controller's `OnRefreshStrategy` is - /// then the sole authority over scheduled clusters, so the legacy policy must - /// not also toggle their replication factor (two writers of the replica set is - /// not allowed). The legacy path remains in place to drive scheduling while the - /// gate is off. - pub(crate) async fn check_scheduling_policies(&self) { - if ENABLE_CLUSTER_CONTROLLER.get(self.catalog().system_config().dyncfgs()) { - return; - } - // (So far, we have only this one policy.) - self.check_refresh_policy(); - } - - /// Runs the `SCHEDULE = ON REFRESH` cluster scheduling policy, which makes cluster On/Off - /// decisions based on REFRESH materialized view write frontiers and the current time (the local - /// oracle read ts), and sends `Message::SchedulingDecisions` with these decisions. - /// (Queries the timestamp oracle on a background task.) - fn check_refresh_policy(&self) { - let start_time = Instant::now(); - - // Collect information about REFRESH MVs: - // - cluster - // - hydration_time_estimate of the cluster - // - MV's id - // - MV's write frontier - // - MV's refresh schedule - let mut refresh_mv_infos = Vec::new(); - for cluster in self.catalog().clusters() { - if let ClusterVariant::Managed(ref config) = cluster.config.variant { - match config.schedule { - ClusterSchedule::Manual => { - // Nothing to do, user manages this cluster manually. - } - ClusterSchedule::Refresh { - hydration_time_estimate, - } => { - let mvs = cluster - .bound_objects() - .iter() - .filter_map(|id| { - if let CatalogItem::MaterializedView(mv) = - self.catalog().get_entry(id).item() - { - mv.refresh_schedule.clone().map(|refresh_schedule| { - let (_since, write_frontier) = self - .controller - .storage - .collection_frontiers(mv.global_id_writes()) - .expect("the storage controller should know about MVs that exist in the catalog"); - (mv.global_id_writes(), write_frontier, refresh_schedule) - }) - } else { - None - } - }) - .collect_vec(); - debug!(%cluster.id, ?refresh_mv_infos, "check_refresh_policy"); - refresh_mv_infos.push((cluster.id, hydration_time_estimate, mvs)); - } - } - } - } - - // Spawn a background task that queries the timestamp oracle for the current read timestamp, - // compares this ts with the REFRESH MV write frontiers, thus making On/Off decisions per - // cluster, and sends a `Message::SchedulingDecisions` with these decisions. - let ts_oracle = self.get_local_timestamp_oracle(); - let internal_cmd_tx = self.internal_cmd_tx.clone(); - let check_scheduling_policies_seconds_cloned = - self.metrics.check_scheduling_policies_seconds.clone(); - let compaction_estimate = self - .catalog() - .system_config() - .cluster_refresh_mv_compaction_estimate() - .try_into() - .expect("should be configured to a reasonable value"); - mz_ore::task::spawn(|| "refresh policy get ts and make decisions", async move { - let task_start_time = Instant::now(); - let local_read_ts = ts_oracle.read_ts().await; - debug!(%local_read_ts, ?refresh_mv_infos, "check_refresh_policy background task"); - let decisions = refresh_mv_infos - .into_iter() - .map(|(cluster_id, hydration_time_estimate, refresh_mv_info)| { - // 1. check that - // write_frontier < local_read_ts + hydration_time_estimate - let hydration_estimate = &hydration_time_estimate - .try_into() - .expect("checked during planning"); - let local_read_ts_adjusted = local_read_ts.step_forward_by(hydration_estimate); - let mvs_needing_refresh = refresh_mv_info - .iter() - .cloned() - .filter_map(|(id, frontier, _refresh_schedule)| { - if frontier.less_than(&local_read_ts_adjusted) { - Some(id) - } else { - None - } - }) - .collect_vec(); - - // 2. check that - // prev_refresh + compaction_estimate > local_read_ts - let mvs_needing_compaction = refresh_mv_info - .into_iter() - .filter_map(|(id, frontier, refresh_schedule)| { - let frontier = frontier.as_option(); - // `prev_refresh` will be None in two cases: - // 1. When there is no previous refresh, because we haven't yet had - // the first refresh. In this case, there is no need to schedule - // time now for compaction. - // 2. In the niche case where a `REFRESH EVERY` MV's write frontier - // is empty. In this case, it's not impossible that there would be a - // need for compaction. But I can't see any easy way to correctly - // handle this case, because we don't have any info handy about when - // the last refresh happened in wall clock time, because the - // frontiers have no relation to wall clock time. So, we'll not - // schedule any compaction time. - // (Note that `REFRESH AT` MVs with empty frontiers, which is a more - // common case, are fine, because `last_refresh` will return - // Some(...) for them.) - let prev_refresh = match frontier { - Some(frontier) => frontier.round_down_minus_1(&refresh_schedule), - None => refresh_schedule.last_refresh(), - }; - prev_refresh - .map(|prev_refresh| { - if prev_refresh.step_forward_by(&compaction_estimate) - > local_read_ts - { - Some(id) - } else { - None - } - }) - .flatten() - }) - .collect_vec(); - - let cluster_on = - !mvs_needing_refresh.is_empty() || !mvs_needing_compaction.is_empty(); - ( - cluster_id, - SchedulingDecision::Refresh(RefreshDecision { - cluster_on, - objects_needing_refresh: mvs_needing_refresh, - objects_needing_compaction: mvs_needing_compaction, - hydration_time_estimate, - }), - ) - }) - .collect(); - if let Err(e) = internal_cmd_tx.send(Message::SchedulingDecisions(vec![( - REFRESH_POLICY_NAME, - decisions, - )])) { - // It is not an error for this task to be running after `internal_cmd_rx` is dropped. - warn!("internal_cmd_rx dropped before we could send: {:?}", e); - } - check_scheduling_policies_seconds_cloned - .with_label_values(&[REFRESH_POLICY_NAME, "background"]) - .observe((Instant::now() - task_start_time).as_secs_f64()); - }); - - self.metrics - .check_scheduling_policies_seconds - .with_label_values(&[REFRESH_POLICY_NAME, "main"]) - .observe((Instant::now() - start_time).as_secs_f64()); - } - - /// Handles `SchedulingDecisions`: - /// 1. Adds the newly made decisions to `cluster_scheduling_decisions`. - /// 2. Cleans up old decisions that are for clusters no longer in scope of automated scheduling - /// decisions. - /// 3. For each cluster, it sums up `cluster_scheduling_decisions`, checks the summed up decision - /// against the cluster state, and turns cluster On/Off if needed. - #[mz_ore::instrument(level = "debug")] - pub(crate) async fn handle_scheduling_decisions( - &mut self, - decisions: Vec<(&'static str, Vec<(ClusterId, SchedulingDecision)>)>, - ) { - // When the cluster controller owns the replica set it is the sole writer - // for scheduled clusters. Drop any legacy decisions still in flight from a - // background task spawned before the gate flipped on, so the two never - // contend. (`check_scheduling_policies` already stops spawning new ones.) - if ENABLE_CLUSTER_CONTROLLER.get(self.catalog().system_config().dyncfgs()) { - return; - } - - let start_time = Instant::now(); - - // 1. Add the received decisions to `cluster_scheduling_decisions`. - for (policy_name, decisions) in decisions.iter() { - for (cluster_id, decision) in decisions { - self.cluster_scheduling_decisions - .entry(*cluster_id) - .or_insert_with(Default::default) - .insert(policy_name, decision.clone()); - } - } - - // 2. Clean up those clusters from `scheduling_decisions` that - // - have been dropped, or - // - were switched to unmanaged, or - // - were switched to `SCHEDULE = MANUAL`. - for cluster_id in self - .cluster_scheduling_decisions - .keys() - .cloned() - .collect_vec() - { - match self.get_managed_cluster_config(cluster_id) { - None => { - // Cluster have been dropped or switched to unmanaged. - debug!( - "handle_scheduling_decisions: \ - Removing cluster {} from cluster_scheduling_decisions, \ - because get_managed_cluster_config returned None", - cluster_id - ); - self.cluster_scheduling_decisions.remove(&cluster_id); - } - Some(managed_config) => { - if matches!(managed_config.schedule, ClusterSchedule::Manual) { - debug!( - "handle_scheduling_decisions: \ - Removing cluster {} from cluster_scheduling_decisions, \ - because schedule is Manual", - cluster_id - ); - self.cluster_scheduling_decisions.remove(&cluster_id); - } - } - } - } - - // 3. Act on `scheduling_decisions` where needed. - let mut altered_a_cluster = false; - for (cluster_id, decisions) in self.cluster_scheduling_decisions.clone() { - // We touch a cluster only when all policies have made a decision about it. This is - // to ensure that after an envd restart all policies have a chance to run at least once - // before we turn off a cluster, to avoid spuriously turning off a cluster and possibly - // losing a hydrated state. - if POLICIES.iter().all(|policy| decisions.contains_key(policy)) { - // Check whether the cluster's state matches the needed state. - // If any policy says On, then we need a replica. - let needs_replica = decisions - .values() - .map(|decision| decision.cluster_on()) - .contains(&true); - let cluster = self.catalog().get_cluster(cluster_id); - let cluster_name = cluster.name().to_string(); - let cluster_config = cluster.config.clone(); - // NOTE: the durable replication factor is not a reliable - // on/off signal here. The cluster controller runs a scheduled - // cluster's replica while holding the factor at 0, so after a - // controller gate-off the factor can disagree with the replica - // set that actually exists. Decide from the physical replicas, - // with the same exclusions as the controller's ownership test - // (`ObservedReplica::owned_shape`): internal and billed-as - // replicas are manually managed, pending ones belong to an - // in-flight reconfiguration. In a pure legacy world the factor - // and the replica set always agree, so both signals give the - // same answer there. - let owned_replicas: Vec<_> = cluster - .replicas() - .filter(|r| { - !r.config.location.internal() - && r.config.location.billed_as().is_none() - && !r.config.location.pending() - }) - .map(|r| r.replica_id) - .collect(); - let has_pending_replica = cluster.replicas().any(|r| r.config.location.pending()); - let mut new_config = cluster_config.clone(); - let ClusterVariant::Managed(managed_config) = &mut new_config.variant else { - panic!("cleaned up unmanaged clusters above"); - }; - let replication_factor = managed_config.replication_factor; - let has_replica = !owned_replicas.is_empty(); // Is it On? - let reason = crate::catalog::ReplicaCreateDropReason::ClusterScheduling( - decisions.values().cloned().collect(), - ); - if has_pending_replica { - // A graceful reconfiguration owns the replica set until it - // finalizes. The turn-on alter below would reject this - // case itself, the direct drop and adopt paths must not - // race the finalization either. This covers only legacy - // `-pending` replicas: controller-created overlap replicas - // are not pending and are handled by the adopt and - // turn-off branches instead. - debug!( - "handle_scheduling_decisions skipped cluster {} because it is \ - undergoing a graceful reconfiguration", - cluster_id - ); - } else if needs_replica && !has_replica { - // Turn the cluster On. - altered_a_cluster = true; - managed_config.replication_factor = 1; - if let Err(e) = self - .sequence_alter_cluster_managed_to_managed( - None, - cluster_id, - new_config.clone(), - reason, - AlterClusterPlanStrategy::None, - ) - .await - { - if let AdapterError::AlterClusterWhilePendingReplicas = e { - debug!( - "handle_scheduling_decisions tried to alter a cluster that is undergoing a graceful reconfiguration" - ); - } else { - soft_panic_or_log!( - "handle_scheduling_decisions couldn't alter cluster {}. \ - Old config: {:?}, \ - New config: {:?}, \ - Error: {}", - cluster_id, - cluster_config, - new_config, - e - ); - } - } - } else if !needs_replica && has_replica { - // Turn the cluster Off. Drop the replicas by id rather - // than altering the factor down: a replica handed over by - // the controller exists while the factor is already 0 (an - // alter to 0 would be a no-op there), and it may not sit - // at the canonical `r` name a factor-derived drop - // would look for. - altered_a_cluster = true; - let drops = owned_replicas - .into_iter() - .map(|replica_id| { - crate::catalog::DropObjectInfo::ClusterReplica(( - cluster_id, - replica_id, - reason.clone(), - )) - }) - .collect(); - managed_config.replication_factor = 0; - let reconfiguration_audit = cancel_carried_reconfiguration(&mut new_config); - let mut ops = vec![crate::catalog::Op::DropObjects(drops)]; - // After a controller handoff the factor is already 0 and - // there is usually no record to retire, so the config - // write would be a no-op. Push it only when something - // actually changed. - if new_config != cluster_config || reconfiguration_audit.is_some() { - ops.push(crate::catalog::Op::UpdateClusterConfig { - id: cluster_id, - name: cluster_name, - config: new_config.clone(), - reconfiguration_audit, - burst_audit: None, - }); - } - if let Err(e) = self.catalog_transact(None, ops).await { - soft_panic_or_log!( - "handle_scheduling_decisions couldn't turn off cluster {}. \ - Old config: {:?}, \ - New config: {:?}, \ - Error: {}", - cluster_id, - cluster_config, - new_config, - e - ); - } - } else if needs_replica && replication_factor == 0 { - // The controller left in-window replicas behind on - // gate-off (`has_replica` is true here). Adopt exactly - // one: the scheduled-cluster invariant caps the factor at - // 1 (the planner refuses higher, and `unplan` asserts it), - // so the lowest-id replica is kept, the factor is aligned - // with it so later decisions and user `ALTER`s see a - // consistent on-state, and any surplus is retired in the - // same transaction. Surplus replicas are possible when a - // pre-schedule reconfiguration's overlap replica was live - // at gate-off, those are not marked pending. Nothing is - // created. - altered_a_cluster = true; - let mut owned_replicas = owned_replicas; - owned_replicas.sort_unstable(); - let surplus = owned_replicas.split_off(1); - managed_config.replication_factor = 1; - let reconfiguration_audit = cancel_carried_reconfiguration(&mut new_config); - let mut ops = Vec::new(); - if !surplus.is_empty() { - let drops = surplus - .into_iter() - .map(|replica_id| { - crate::catalog::DropObjectInfo::ClusterReplica(( - cluster_id, - replica_id, - reason.clone(), - )) - }) - .collect(); - ops.push(crate::catalog::Op::DropObjects(drops)); - } - ops.push(crate::catalog::Op::UpdateClusterConfig { - id: cluster_id, - name: cluster_name, - config: new_config.clone(), - reconfiguration_audit, - burst_audit: None, - }); - if let Err(e) = self.catalog_transact(None, ops).await { - soft_panic_or_log!( - "handle_scheduling_decisions couldn't adopt replicas of cluster {}. \ - Old config: {:?}, \ - New config: {:?}, \ - Error: {}", - cluster_id, - cluster_config, - new_config, - e - ); - } - } - } else { - debug!( - "handle_scheduling_decisions: \ - Not all policies have made a decision about cluster {}. decisions: {:?}", - cluster_id, decisions, - ); - } - } - - self.metrics - .handle_scheduling_decisions_seconds - .with_label_values(&[altered_a_cluster.to_string().as_str()]) - .observe((Instant::now() - start_time).as_secs_f64()); - } - - /// Returns the managed config for a cluster. Returns None if the cluster doesn't exist or if - /// it's an unmanaged cluster. - fn get_managed_cluster_config(&self, cluster_id: ClusterId) -> Option { - let cluster = self.catalog().try_get_cluster(cluster_id)?; - if let ClusterVariant::Managed(managed_config) = cluster.config.variant.clone() { - Some(managed_config) - } else { - None - } - } -} diff --git a/src/adapter/src/coord/message_handler.rs b/src/adapter/src/coord/message_handler.rs index 988000c1abfb4..71c81667b1055 100644 --- a/src/adapter/src/coord/message_handler.rs +++ b/src/adapter/src/coord/message_handler.rs @@ -245,14 +245,6 @@ impl Coordinator { ); } } - Message::CheckSchedulingPolicies => { - self.check_scheduling_policies().boxed_local().await; - } - Message::SchedulingDecisions(decisions) => { - self.handle_scheduling_decisions(decisions) - .boxed_local() - .await; - } Message::ClusterControllerRequest(request) => { self.handle_cluster_controller_request(request) .boxed_local() diff --git a/src/adapter/src/coord/sequencer.rs b/src/adapter/src/coord/sequencer.rs index c2c22fc996c83..f6f56f204796b 100644 --- a/src/adapter/src/coord/sequencer.rs +++ b/src/adapter/src/coord/sequencer.rs @@ -94,7 +94,6 @@ use crate::util::ClientTransmitter; // big refactoring after the old peek sequencing is removed. mod inner; -pub(crate) use inner::cancel_carried_reconfiguration; impl Coordinator { /// BOXED FUTURE: As of Nov 2023 the returned Future from this function was 34KB. This would diff --git a/src/adapter/src/coord/sequencer/inner.rs b/src/adapter/src/coord/sequencer/inner.rs index 4b62df13d3edb..f979d56383ef1 100644 --- a/src/adapter/src/coord/sequencer/inner.rs +++ b/src/adapter/src/coord/sequencer/inner.rs @@ -126,7 +126,6 @@ use crate::{PeekResponseUnary, ReadHolds}; type RtrTimestampFuture = BoxFuture<'static, Result>; mod cluster; -pub(crate) use cluster::cancel_carried_reconfiguration; mod copy_from; mod create_index; mod create_materialized_view; diff --git a/src/adapter/src/coord/sequencer/inner/cluster.rs b/src/adapter/src/coord/sequencer/inner/cluster.rs index 4fd7b174d467a..a05c760b561e0 100644 --- a/src/adapter/src/coord/sequencer/inner/cluster.rs +++ b/src/adapter/src/coord/sequencer/inner/cluster.rs @@ -48,7 +48,6 @@ use tracing::{Instrument, Span, debug}; use mz_adapter_types::dyncfgs::{ DEFAULT_CLUSTER_RECONFIGURATION_TIMEOUT, ENABLE_BACKGROUND_ALTER_CLUSTER, - ENABLE_CLUSTER_CONTROLLER, }; use super::return_if_err; @@ -316,10 +315,8 @@ impl Coordinator { // path below). A system/builtin cluster is never converged by the // controller, so it must not be reshaped into a durable reconfiguration // record nobody would cut over. It takes the direct realized-config path - // below, exactly as it does with the controller off. - let cluster_controller_owns = ENABLE_CLUSTER_CONTROLLER - .get(self.catalog().system_config().dyncfgs()) - && cluster_id.is_user(); + // below. + let cluster_controller_owns = cluster_id.is_user(); let reconfiguration_in_flight = matches!( &config.variant, Managed(managed) if managed @@ -2007,21 +2004,18 @@ impl Coordinator { } } - // When the controller owns the managed replica set (master gate on, user - // cluster), a non-record change reaching this path is replication-factor - // only. Config-shape changes (size/logging/AZ) are reshaped into a durable + // When the controller owns the managed replica set (a user cluster), a + // non-record change reaching this path is replication-factor only. + // Config-shape changes (size/logging/AZ) are reshaped into a durable // reconfiguration record before they get here. The controller reconciles // the replica set to the realized config's new count on its next tick, so // we update only the realized config and emit no create/drop here. Doing // both fights the controller. It derives replica names from the observed // set, so an adapter create by canonical `rN` can collide with a // controller-chosen name, and an adapter drop by canonical `rN` can miss a - // churned one. With the gate off (or a system cluster, which the - // controller never owns) the legacy path below still does the create/drop - // directly. - let controller_owns = ENABLE_CLUSTER_CONTROLLER - .get(self.catalog().system_config().dyncfgs()) - && cluster_id.is_user(); + // churned one. For a system cluster, which the controller never owns, the + // direct path below still does the create/drop itself. + let controller_owns = cluster_id.is_user(); // Count exactly as many replica ids as the branches below consume. The // config-changed branches recreate all replicas. A pure scale-up creates @@ -2244,15 +2238,9 @@ impl Coordinator { // config. Otherwise the config write happens here. With the controller // owning the cluster, a record still in progress belongs to a live, // converging reconfiguration this write didn't touch: carry it through - // untouched. Without (gate off, or a system cluster), such a record is - // orphaned, so retain it as cancelled with the matching audit intent - // rather than risk a bogus revival if the gate comes back on. - // - // NOTE: `handle_scheduling_decisions` also calls this function and - // bypasses the sequencer's guards. It runs only while the controller - // gate is off, where the cancel-carried write below retires any - // in-progress record instead of leaving it behind for a controller - // that is not running. + // untouched. On a system cluster, which the controller never owns, such + // a record is orphaned, so retain it as cancelled with the matching + // audit intent rather than leave it behind for nothing to settle. match finalization_needed { NeedsFinalization::No => { let mut new_config = new_config; @@ -2608,20 +2596,15 @@ struct ReconfigurationDimensionsUnchanged { arrangement_compression: bool, } -/// Retains a stale in-progress reconfiguration record carried by a legacy-path +/// Retains a stale in-progress reconfiguration record carried by a direct /// config write as cancelled, returning the audit intent to declare with the /// write. /// -/// The legacy write paths (controller gate off), the ALTER sequencer and the -/// legacy scheduler, change the realized config directly and know nothing -/// about reconfiguration records. Nothing on those -/// paths ever settles a record, and carrying an in-progress one forward invites -/// a bogus revival, up to a forced cut-over to an obsolete target, if the gate -/// is turned back on later. A record can only be in progress here if it was -/// written while the gate was on. -pub(crate) fn cancel_carried_reconfiguration( - config: &mut ClusterConfig, -) -> Option { +/// The direct write paths change the realized config themselves and settle no +/// record. Carrying an in-progress one forward would leave it for a controller +/// that does not own this cluster, so nothing would ever drive it to a terminal +/// status. +fn cancel_carried_reconfiguration(config: &mut ClusterConfig) -> Option { let ClusterVariant::Managed(managed) = &mut config.variant else { return None; }; diff --git a/src/adapter/src/metrics.rs b/src/adapter/src/metrics.rs index b5168a4916b98..d6c31194d6e4c 100644 --- a/src/adapter/src/metrics.rs +++ b/src/adapter/src/metrics.rs @@ -43,8 +43,6 @@ pub struct Metrics { pub append_table_duration_seconds: Histogram, pub webhook_validation_reduce_failures: IntCounterVec, pub webhook_get_appender: IntCounter, - pub check_scheduling_policies_seconds: HistogramVec, - pub handle_scheduling_decisions_seconds: HistogramVec, pub row_set_finishing_seconds: Histogram, pub session_startup_table_writes_seconds: Histogram, pub parse_seconds: Histogram, @@ -198,18 +196,6 @@ impl Metrics { name: "mz_webhook_get_appender_count", help: "Count of getting a webhook appender from the Coordinator.", )), - check_scheduling_policies_seconds: registry.register(metric!( - name: "mz_check_scheduling_policies_seconds", - help: "The time each policy in `check_scheduling_policies` takes.", - var_labels: ["policy", "thread"], - buckets: histogram_seconds_buckets(0.000_128, 8.0), - )), - handle_scheduling_decisions_seconds: registry.register(metric!( - name: "mz_handle_scheduling_decisions_seconds", - help: "The time `handle_scheduling_decisions` takes.", - var_labels: ["altered_a_cluster"], - buckets: histogram_seconds_buckets(0.000_128, 8.0), - )), row_set_finishing_seconds: registry.register(metric!( name: "mz_row_set_finishing_seconds", help: "The time it takes to run RowSetFinishing::finish.", diff --git a/src/cluster-controller/src/ctx.rs b/src/cluster-controller/src/ctx.rs index beca4fd3d9215..e27a5f2ba4974 100644 --- a/src/cluster-controller/src/ctx.rs +++ b/src/cluster-controller/src/ctx.rs @@ -88,9 +88,8 @@ impl ObservedReplica { /// compares it against the read timestamp (`less_than`) to decide whether the MV /// still needs a refresh. For the compaction window it reads the frontier's lone /// element via `as_option` to find the previous refresh time, falling back to the -/// schedule's last refresh on the empty/sealed frontier `[]`, mirroring the -/// legacy refresh policy. The frontier of a single-input total-order MV holds at -/// most one element. +/// schedule's last refresh on the empty/sealed frontier `[]`. The frontier of a +/// single-input total-order MV holds at most one element. #[derive(Clone, Debug, PartialEq, Eq)] pub struct RefreshMvInfo { /// The MV's writes-`GlobalId`: the identity the window decision records in diff --git a/src/cluster-controller/src/strategy.rs b/src/cluster-controller/src/strategy.rs index 59cbd9e3744f5..c8584c713bef7 100644 --- a/src/cluster-controller/src/strategy.rs +++ b/src/cluster-controller/src/strategy.rs @@ -384,8 +384,8 @@ impl Strategy for GracefulReconfigurationStrategy { /// replica at the cluster's realized shape while the cluster is inside a refresh /// window, and nothing otherwise. The window decision keys on the bound REFRESH /// materialized views' write frontiers, their refresh schedules, the configured -/// hydration-time estimate, and the current read timestamp (the same signals the -/// legacy scheduler reads), all carried in [`RefreshWindowInputs`]. +/// hydration-time estimate, and the current read timestamp, all carried in +/// [`RefreshWindowInputs`]. /// /// The controller (not the user's `replication_factor`) owns a scheduled /// cluster's replica set, so [`Strategy::update_state`] normalizes the realized @@ -394,12 +394,10 @@ impl Strategy for GracefulReconfigurationStrategy { /// a scheduled cluster, with `mz_cluster_replicas` authoritative for what is /// actually running. /// -/// NB: the decision is re-derived purely from the live signals each tick; there -/// is no "all policies have decided" latch like `cluster_scheduling.rs` needs. -/// That scheduler collects policy decisions asynchronously and across ticks, so -/// turning a cluster off is only safe once every policy has reported. We pull a -/// complete decision from durable + storage state on every tick, so the first -/// tick after a restart already decides from the same inputs as a steady tick. +/// NB: the decision is re-derived purely from the live signals each tick, with +/// no cross-tick latch. We pull a complete decision from durable and storage +/// state on every tick, so the first tick after a restart already decides from +/// the same inputs as a steady tick. #[derive(Clone, Copy, Debug, Default)] pub struct OnRefreshStrategy; @@ -485,9 +483,10 @@ impl Strategy for OnRefreshStrategy { _now: Timestamp, ) -> StateWrite { // The controller owns a scheduled cluster's replica set, so hold the - // realized `replication_factor` at `0`. A stale non-zero value (e.g. left - // by the legacy scheduler toggling 0↔1) would otherwise have the implicit - // baseline desire a replica the on-refresh strategy does not, a flap. + // realized `replication_factor` at `0`. A stale non-zero value (e.g. + // carried over from a cluster that was just given a schedule) would + // otherwise have the implicit baseline desire a replica the on-refresh + // strategy does not, a flap. // Only write when it is actually non-zero, to keep steady ticks no-ops. if matches!(state.schedule, ClusterSchedule::Manual) || state.replication_factor == 0 { return StateWrite::default(); @@ -540,9 +539,8 @@ impl Strategy for OnRefreshStrategy { return Vec::new(); } // One replica at the realized shape (`cluster.size` plus the cluster's AZ - // pool and logging), matching what the legacy scheduler brings up. The - // window decision rides inside the reason so the create it may produce - // can carry the audit detail. + // pool and logging). The window decision rides inside the reason so the + // create it may produce can carry the audit detail. vec![DesiredReplica { shape: state.realized_shape(), reason: CreateReason::OnRefresh(decision), diff --git a/src/cluster-controller/src/tests.rs b/src/cluster-controller/src/tests.rs index 5c8db3dd1f7bd..19e51834eac15 100644 --- a/src/cluster-controller/src/tests.rs +++ b/src/cluster-controller/src/tests.rs @@ -2278,7 +2278,7 @@ fn on_refresh_window_decision_lists_due_mvs() { fn on_refresh_caught_up_at_read_ts_is_off() { // Frontier exactly at the read ts (and no hydration lead, no compaction // window): the MV is caught up, so the cluster is Off. The needs-refresh check - // is strict (`frontier < read_ts + estimate`), matching the legacy scheduler. + // is strict (`frontier < read_ts + estimate`). let c = cluster(1); let inputs = window_inputs(100, 0, Some(100), refresh_at(50)); let (state, signals) = scheduled_state(c, "100cc", 0, 0, Vec::new(), Some(inputs)); @@ -2294,8 +2294,7 @@ fn on_refresh_caught_up_at_read_ts_is_off() { fn on_refresh_empty_frontier_needs_no_refresh() { // An empty (sealed) write frontier `[]` is the "complete past every timestamp" // state: `Antichain::less_than` is `false` for every timestamp, so the MV never - // reads as needing a refresh on that count, exactly as the legacy refresh - // policy decides it with `Antichain::less_than`. The compaction window is also + // reads as needing a refresh on that count. The compaction window is also // closed here (read ts 1000 is well past the last `AT 200` plus the compaction // estimate), so the cluster is Off. // diff --git a/src/sql/src/session/vars.rs b/src/sql/src/session/vars.rs index 1cdc5dd702056..1922337e0e2d8 100644 --- a/src/sql/src/session/vars.rs +++ b/src/sql/src/session/vars.rs @@ -1309,7 +1309,6 @@ impl SystemVars { &cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY, &cluster_scheduling::CLUSTER_SOFTEN_AZ_AFFINITY_WEIGHT, &cluster_scheduling::CLUSTER_ALTER_CHECK_READY_INTERVAL, - &cluster_scheduling::CLUSTER_CHECK_SCHEDULING_POLICIES_INTERVAL, &cluster_scheduling::CLUSTER_SECURITY_CONTEXT_ENABLED, &cluster_scheduling::CLUSTER_REFRESH_MV_COMPACTION_ESTIMATE, &grpc_client::HTTP2_KEEP_ALIVE_TIMEOUT, @@ -2258,10 +2257,6 @@ impl SystemVars { *self.expect_value(&cluster_scheduling::CLUSTER_ALTER_CHECK_READY_INTERVAL) } - pub fn cluster_check_scheduling_policies_interval(&self) -> Duration { - *self.expect_value(&cluster_scheduling::CLUSTER_CHECK_SCHEDULING_POLICIES_INTERVAL) - } - pub fn cluster_security_context_enabled(&self) -> bool { *self.expect_value(&cluster_scheduling::CLUSTER_SECURITY_CONTEXT_ENABLED) } diff --git a/src/sql/src/session/vars/definitions.rs b/src/sql/src/session/vars/definitions.rs index 09e49d767be7e..8bc9d8d9cdddb 100644 --- a/src/sql/src/session/vars/definitions.rs +++ b/src/sql/src/session/vars/definitions.rs @@ -1681,17 +1681,6 @@ pub mod cluster_scheduling { false, ); - const DEFAULT_CHECK_SCHEDULING_POLICIES_INTERVAL: Duration = Duration::from_secs(3); - - pub static CLUSTER_CHECK_SCHEDULING_POLICIES_INTERVAL: VarDefinition = VarDefinition::new( - "cluster_check_scheduling_policies_interval", - value!(Duration; DEFAULT_CHECK_SCHEDULING_POLICIES_INTERVAL), - "How often policies are invoked to automatically start/stop clusters, e.g., \ - for REFRESH EVERY materialized views.", - false, - ) - .with_constraint(&NON_ZERO_DURATION); - pub static CLUSTER_SECURITY_CONTEXT_ENABLED: VarDefinition = VarDefinition::new( "cluster_security_context_enabled", value!(bool; DEFAULT_SECURITY_CONTEXT_ENABLED), diff --git a/src/sqllogictest/src/bin/sqllogictest.rs b/src/sqllogictest/src/bin/sqllogictest.rs index 8e4c7643d0d03..ec6aecd00d419 100644 --- a/src/sqllogictest/src/bin/sqllogictest.rs +++ b/src/sqllogictest/src/bin/sqllogictest.rs @@ -18,7 +18,7 @@ use std::process::ExitCode; use chrono::Utc; use clap::ArgAction; -use mz_adapter_types::dyncfgs::{ENABLE_BACKGROUND_ALTER_CLUSTER, ENABLE_CLUSTER_CONTROLLER}; +use mz_adapter_types::dyncfgs::ENABLE_BACKGROUND_ALTER_CLUSTER; use mz_orchestrator_tracing::{StaticTracingConfig, TracingCliArgs}; use mz_ore::cli::{self, CliConfig, KeyValueArg}; use mz_ore::metrics::MetricsRegistry; @@ -176,19 +176,13 @@ async fn main() -> ExitCode { } } - // The cluster controller and background ALTER CLUSTER land dark in - // production (the dyncfg defaults stay false); force them on for - // sqllogictest so the suite exercises the controller owning the - // managed-cluster replica set. These are dyncfgs (set by name), and a - // caller-provided value wins. - for name in [ - ENABLE_CLUSTER_CONTROLLER.name(), - ENABLE_BACKGROUND_ALTER_CLUSTER.name(), - ] { - system_parameter_defaults - .entry(name.to_string()) - .or_insert_with(|| "true".to_string()); - } + // Pin background ALTER CLUSTER on for the suite so a config-shape + // `ALTER CLUSTER` returns immediately rather than blocking on the + // wait-shim. This is a dyncfg (set by name), and a caller-provided value + // wins. + system_parameter_defaults + .entry(ENABLE_BACKGROUND_ALTER_CLUSTER.name().to_string()) + .or_insert_with(|| "true".to_string()); let config = RunConfig { stdout: &OutputStream::new(io::stdout(), args.timestamps), diff --git a/test/launchdarkly-flag-consistency/mzcompose.py b/test/launchdarkly-flag-consistency/mzcompose.py index 27ab80604948e..094c20f552554 100644 --- a/test/launchdarkly-flag-consistency/mzcompose.py +++ b/test/launchdarkly-flag-consistency/mzcompose.py @@ -195,7 +195,6 @@ aws_prefetch_sts_connect_timeout catalog_info_metrics_reconcile_interval cluster_alter_check_ready_interval - cluster_check_scheduling_policies_interval cluster_controller_tick_interval cluster_enable_topology_spread cluster_multi_process_replica_az_affinity_weight @@ -240,9 +239,7 @@ enable_statement_arrival_logging enable_binary_date_bin enable_bounded_staleness_isolation - enable_cluster_controller enable_coalesce_case_transform - enable_cluster_controller enable_compute_half_join2 enable_compute_render_fueled_as_specific_collection enable_date_bin_hopping diff --git a/test/pg-cdc/cluster-graceful-reconfiguration.td b/test/pg-cdc/cluster-graceful-reconfiguration.td index eb87b1e2feb73..87c50e60f7dbe 100644 --- a/test/pg-cdc/cluster-graceful-reconfiguration.td +++ b/test/pg-cdc/cluster-graceful-reconfiguration.td @@ -17,9 +17,8 @@ # wait for the source to hydrate on the target, but must still wait for the # target's processes to come online before cutting over. # -# Both reconfiguration paths are exercised below, the legacy foreground path -# and the controller-owned path. The path-selection flags are pinned -# explicitly so the test does not depend on the harness defaults. +# The background flag is pinned explicitly so the test does not depend on the +# harness defaults. $ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr} ALTER SYSTEM SET enable_zero_downtime_cluster_reconfiguration = true @@ -60,64 +59,32 @@ CREATE PUBLICATION mz_source FOR TABLE t1; 2 3 -# ----- Legacy foreground path ----- - $ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr} -ALTER SYSTEM SET enable_cluster_controller = false -ALTER SYSTEM SET enable_background_alter_cluster = false - -# The ALTER blocks until readiness passes. Before the fix it would block until -# the deadline rolls the resize back, failing the test at the statement -# timeout. The raised timeout only gives the success path headroom on slow CI. -$ set-sql-timeout duration=120s +ALTER SYSTEM SET enable_background_alter_cluster = true +# With the background flag on the ALTER returns immediately and the controller +# drives readiness and cut-over. Before the fix the resize would roll back at +# its deadline and the realized size below would never advance. > ALTER CLUSTER source_reconfig SET (SIZE 'scale=1,workers=2') WITH (WAIT UNTIL READY (TIMEOUT '300s', ON TIMEOUT 'ROLLBACK')) -$ set-sql-timeout duration=default +# The raised timeout covers replica boot plus the controller's tick cadence. +$ set-sql-timeout duration=120s # The realized size advanced to the target: the cut-over happened. > SELECT size FROM mz_clusters WHERE name = 'source_reconfig' "scale=1,workers=2" -# The source still serves its data after cut-over. -> SELECT * FROM t1 -1 -2 -3 - -# And it keeps ingesting on the promoted replica. -$ postgres-execute connection=postgres://postgres:postgres@postgres -INSERT INTO t1 VALUES (4), (5); +$ set-sql-timeout duration=default +# The source still serves its data after cut-over, and keeps ingesting on the +# promoted replica. > SELECT * FROM t1 1 2 3 -4 -5 - -# ----- Controller-owned path ----- - -$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr} -ALTER SYSTEM SET enable_cluster_controller = true -ALTER SYSTEM SET enable_background_alter_cluster = true - -# With the background flag on the ALTER returns immediately and the controller -# drives readiness and cut-over. Before the fix the resize would roll back at -# its deadline and the realized size below would never advance. -> ALTER CLUSTER source_reconfig SET (SIZE 'scale=1,workers=1') WITH (WAIT UNTIL READY (TIMEOUT '300s', ON TIMEOUT 'ROLLBACK')) -# The raised timeout covers replica boot plus the controller's tick cadence. -$ set-sql-timeout duration=120s - -> SELECT size FROM mz_clusters WHERE name = 'source_reconfig' -"scale=1,workers=1" - -$ set-sql-timeout duration=default - -# The source keeps serving and ingesting after the controller-driven cut-over. $ postgres-execute connection=postgres://postgres:postgres@postgres -INSERT INTO t1 VALUES (6); +INSERT INTO t1 VALUES (4), (5); > SELECT * FROM t1 1 @@ -125,7 +92,6 @@ INSERT INTO t1 VALUES (6); 3 4 5 -6 > DROP SOURCE mz_source CASCADE > DROP CLUSTER source_reconfig diff --git a/test/sqllogictest/mz_cluster_schedules.slt b/test/sqllogictest/mz_cluster_schedules.slt index fbc86a6d98ea6..b03458927c1c6 100644 --- a/test/sqllogictest/mz_cluster_schedules.slt +++ b/test/sqllogictest/mz_cluster_schedules.slt @@ -161,25 +161,3 @@ JOIN mz_clusters c ON s.cluster_id = c.id WHERE c.name = 'c_rt' ---- 0 - -# --- cluster_check_scheduling_policies_interval rejects zero ------------------ - -# The interval at which scheduling policies run is passed to -# tokio::time::interval during coordinator bootstrap, which panics on a zero -# period. A zero value would therefore crash-loop environmentd on every boot, -# so it must be rejected at ALTER SYSTEM SET time. -simple conn=mz_system,user=mz_system -ALTER SYSTEM SET cluster_check_scheduling_policies_interval = '0s'; ----- -db error: ERROR: parameter "cluster_check_scheduling_policies_interval" cannot have value "0ns": only supports non-zero durations - -# A non-zero value is still accepted. -simple conn=mz_system,user=mz_system -ALTER SYSTEM SET cluster_check_scheduling_policies_interval = '1s'; ----- -COMPLETE 0 - -simple conn=mz_system,user=mz_system -ALTER SYSTEM RESET cluster_check_scheduling_policies_interval; ----- -COMPLETE 0 diff --git a/test/sqllogictest/show_clusters.slt b/test/sqllogictest/show_clusters.slt index 294abd45721e0..a235c5231a926 100644 --- a/test/sqllogictest/show_clusters.slt +++ b/test/sqllogictest/show_clusters.slt @@ -44,9 +44,9 @@ mz_system r1 (scale=1,workers=2) # `SHOW CLUSTERS` surfaces any in-flight reconfiguration or autoscaling action in a -# single `activity` column. With the cluster controller at its default (off), no -# reconfiguration or burst record is ever written, so `activity` is NULL for every -# cluster, managed (baz) or unmanaged (foo). +# single `activity` column. Neither cluster below is ever reconfigured or bursts, +# so no record is written and `activity` is NULL for both, managed (baz) and +# unmanaged (foo). statement ok CREATE CLUSTER baz (SIZE 'scale=1,workers=1', REPLICATION FACTOR 1) diff --git a/test/testdrive/cluster-controller.td b/test/testdrive/cluster-controller.td index 2240b9413a743..9060c3672ad50 100644 --- a/test/testdrive/cluster-controller.td +++ b/test/testdrive/cluster-controller.td @@ -7,27 +7,26 @@ # the Business Source License, use of this software will be governed # by the Apache License, Version 2.0. -# Boundary test for the cluster controller. With the master gate forced on and -# the tick interval driven down, the controller reconciles managed clusters many -# times within the test window. Convergence is asynchronous (a config-shape ALTER -# returns before the realized config settles), so every readback below is a -# retrying testdrive query (`>` re-runs until it matches or the SQL timeout -# elapses), never a fixed-duration sleep. Where a transient in-flight state must -# be observed, the controller is frozen first so the state cannot move; where the -# assertion is a negative ("the controller does nothing"), a sibling -# reconfiguration is used as a liveness synchronizer, proving the reconcile loop -# ran many ticks without acting on the cluster under test. +# Boundary test for the cluster controller. With the tick interval driven down, +# the controller reconciles managed clusters many times within the test window. +# Convergence is asynchronous (a config-shape ALTER returns before the realized +# config settles), so every readback below is a retrying testdrive query (`>` +# re-runs until it matches or the SQL timeout elapses), never a fixed-duration +# sleep. Where a transient in-flight state must be observed, the controller is +# held off first (by cranking the tick interval back up) so the state cannot +# move; where the assertion is a negative ("the controller does nothing"), a +# sibling reconfiguration is used as a liveness synchronizer, proving the +# reconcile loop ran many ticks without acting on the cluster under test. # Give convergence (provision + hydrate + cut-over) ample room on a loaded CI host # without ever sleeping the full budget. `>` returns as soon as it matches. $ set-sql-timeout duration=120s -# Force the gate on and drive the tick interval down so the controller ticks -# ~hundreds of times across the waits below. Both are re-read each tick, so the -# flips take effect without a restart. The graceful cases use the WITH (WAIT ...) -# surface, whose planner acceptance is gated on enable_zero_downtime. +# Drive the tick interval down so the controller ticks ~hundreds of times across +# the waits below. It is re-read each tick, so the flip takes effect without a +# restart. The graceful cases use the WITH (WAIT ...) surface, whose planner +# acceptance is gated on enable_zero_downtime. $ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}/materialize -ALTER SYSTEM SET enable_cluster_controller = true ALTER SYSTEM SET cluster_controller_tick_interval = '5ms' ALTER SYSTEM SET enable_zero_downtime_cluster_reconfiguration = true @@ -600,23 +599,27 @@ ALTER SYSTEM RESET max_credit_consumption_rate # The unmanaged variant has no reconfiguration field, so the conversion would # silently drop an in-progress record with no terminal status and no audit # event, and strand any overlap replicas. The conversion is refused instead. -# The controller is frozen (master gate off) right after the record is written -# so the in-flight state holds still while the rejection is asserted. +# +# The controller is held off across the assertion by cranking the tick interval +# up before the reconfiguring ALTER. It still reconciles once right after that +# ALTER (a cluster write wakes it), which is what brings the overlap replica up, +# but from then on nothing wakes it: the refused conversion below writes no +# catalog state, so it cannot race a cut-over. + +$ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}/materialize +ALTER SYSTEM SET cluster_controller_tick_interval = '10s' > CREATE CLUSTER cc_unmanaged (SIZE 'scale=1,workers=1', REPLICATION FACTOR 1) > ALTER CLUSTER cc_unmanaged SET (SIZE 'scale=1,workers=2') -$ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}/materialize -ALTER SYSTEM SET enable_cluster_controller = false - ! ALTER CLUSTER cc_unmanaged SET (MANAGED = false) contains:cannot convert cluster to unmanaged while a reconfiguration is in progress -# Unfreeze and cancel by ALTERing back to the realized size. Once the record -# settles (the retrying `>` rides out the interim), the conversion goes +# Restore the cadence and cancel by ALTERing back to the realized size. Once the +# record settles (the retrying `>` rides out the interim), the conversion goes # through. $ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}/materialize -ALTER SYSTEM SET enable_cluster_controller = true +ALTER SYSTEM SET cluster_controller_tick_interval = '5ms' > ALTER CLUSTER cc_unmanaged SET (SIZE 'scale=1,workers=1') > ALTER CLUSTER cc_unmanaged SET (MANAGED = false) @@ -1343,211 +1346,8 @@ manual scale=1,workers=2 0 > DROP CLUSTER cc_wait_manual -# ----- Break-glass handoff to the legacy scheduler ----- -# -# Disabling the controller hands scheduled clusters back to the legacy -# scheduler. The controller runs a scheduled cluster's replica while holding -# the durable replication factor at 0, so the legacy scheduler must decide -# on/off from the physical replica set, not the factor. Otherwise a -# controller-created replica would leak across the handoff. - -# Gate-off with the window still open: the legacy scheduler adopts the running -# replica by aligning the replication factor with it, and retires it once the -# window closes. -> CREATE CLUSTER cc_handoff (SIZE 'scale=1,workers=1', SCHEDULE = ON REFRESH (HYDRATION TIME ESTIMATE = '60 seconds')) -> CREATE TABLE cc_handoff_t (x int) -> CREATE MATERIALIZED VIEW cc_handoff_mv IN CLUSTER cc_handoff WITH (REFRESH = EVERY '1 second') AS SELECT count(*) FROM cc_handoff_t -> SELECT count(*) FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_handoff' -1 -> SELECT replication_factor FROM mz_catalog.mz_clusters WHERE name = 'cc_handoff' -0 - -$ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}/materialize -ALTER SYSTEM SET enable_cluster_controller = false - -> SELECT replication_factor FROM mz_catalog.mz_clusters WHERE name = 'cc_handoff' -1 -> SELECT count(*) FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_handoff' -1 - -> DROP MATERIALIZED VIEW cc_handoff_mv -> SELECT count(*) FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_handoff' -0 -> SELECT replication_factor FROM mz_catalog.mz_clusters WHERE name = 'cc_handoff' -0 -> DROP CLUSTER cc_handoff -> DROP TABLE cc_handoff_t - -# Gate-off and MV-drop together: the legacy scheduler sees a closed window and -# a replica the factor does not reflect, and still retires it. -$ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}/materialize -ALTER SYSTEM SET enable_cluster_controller = true - -> CREATE CLUSTER cc_handoff2 (SIZE 'scale=1,workers=1', SCHEDULE = ON REFRESH (HYDRATION TIME ESTIMATE = '60 seconds')) -> CREATE TABLE cc_handoff2_t (x int) -> CREATE MATERIALIZED VIEW cc_handoff2_mv IN CLUSTER cc_handoff2 WITH (REFRESH = EVERY '1 second') AS SELECT count(*) FROM cc_handoff2_t -> SELECT count(*) FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_handoff2' -1 - -$ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}/materialize -ALTER SYSTEM SET enable_cluster_controller = false -DROP MATERIALIZED VIEW materialize.public.cc_handoff2_mv - -> SELECT count(*) FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_handoff2' -0 -> SELECT replication_factor FROM mz_catalog.mz_clusters WHERE name = 'cc_handoff2' -0 -> DROP CLUSTER cc_handoff2 -> DROP TABLE cc_handoff2_t - -# Regression: a mid-window shape bounce lands the surviving replica on a -# non-canonical name (r1 -> r2), and the break-glass adopt aligns only the -# replication factor. A legacy config-shape ALTER must still replace that -# replica: it drops the cluster's observed replica set by id, not the -# factor-derived canonical names, so the resize cannot miss the survivor and -# leave a duplicate behind. -$ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}/materialize -ALTER SYSTEM SET enable_cluster_controller = true - -> CREATE CLUSTER cc_handoff3 (SIZE 'scale=1,workers=1', SCHEDULE = ON REFRESH (HYDRATION TIME ESTIMATE = '60 seconds')) -> CREATE TABLE cc_handoff3_t (x int) -> CREATE MATERIALIZED VIEW cc_handoff3_mv IN CLUSTER cc_handoff3 WITH (REFRESH = EVERY '1 second') AS SELECT count(*) FROM cc_handoff3_t -> SELECT count(*) FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_handoff3' -1 - -# The direct-path resize makes the controller bounce the in-window replica, -# which comes back under a fresh, non-canonical name. -> ALTER CLUSTER cc_handoff3 SET (SIZE 'scale=1,workers=2') -> SELECT r.size FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_handoff3' -scale=1,workers=2 -> SELECT count(*) FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_handoff3' -1 - -$ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}/materialize -ALTER SYSTEM SET enable_cluster_controller = false - -# The legacy scheduler adopts the surviving replica. -> SELECT replication_factor FROM mz_catalog.mz_clusters WHERE name = 'cc_handoff3' -1 - -# A legacy resize replaces the adopted replica: still exactly one, at the new -# size. -> ALTER CLUSTER cc_handoff3 SET (SIZE 'scale=1,workers=1') -> SELECT count(*) FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_handoff3' -1 -> SELECT r.size FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_handoff3' -scale=1,workers=1 - -# Once the window closes the legacy scheduler retires the replica and the -# factor follows. -> DROP MATERIALIZED VIEW cc_handoff3_mv -> SELECT count(*) FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_handoff3' -0 -> SELECT replication_factor FROM mz_catalog.mz_clusters WHERE name = 'cc_handoff3' -0 -> DROP CLUSTER cc_handoff3 -> DROP TABLE cc_handoff3_t - -# ----- Break-glass cleanup of a stranded graceful reconfiguration ----- -# -# Disabling the controller mid-flight strands a graceful reconfiguration: the -# controller is the only component that drives a reconfiguration record to a -# terminal status, so once it is off the in-progress record and the overlap -# replica it provisioned are left in place with nothing to retire them. The -# cleanup lever is a config-shape ALTER to a new size. With the controller off -# it takes the legacy path, which drops the entire observed owned replica set by -# id (baseline plus stranded overlap), recreates a dense r1..rN at the new size, -# and cancels the carried record in the same transaction. An ALTER back to the -# realized shape is not a lever here: it is byte-identical to the realized -# config, so the sequencer short-circuits it as a no-op and the record survives. -# -# A sleeping materialized view pins hydration on every replica of the cluster, -# baseline and overlap alike (the same mz_sleep pattern used above), so the -# target set never hydrates and the reconfiguration cannot cut over. The -# in-flight state then holds still while we disable the controller and observe -# the stranding, no matter the tick cadence. - -$ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}/materialize -ALTER SYSTEM SET enable_cluster_controller = true -ALTER SYSTEM SET unsafe_enable_unstable_dependencies = true - -> CREATE CLUSTER cc_strand (SIZE 'scale=1,workers=1', REPLICATION FACTOR 1) - -> CREATE TABLE cc_strand_t (id int) -> INSERT INTO cc_strand_t VALUES (1) - -> CREATE MATERIALIZED VIEW cc_strand_slow IN CLUSTER cc_strand AS - SELECT mz_unsafe.mz_sleep(id * 3600) AS s FROM cc_strand_t - -# Start a graceful reconfiguration to a new size. The controller writes the -# in-progress record and brings up an overlap replica at the target shape -# alongside the baseline replica, but the sleeping view keeps it from hydrating, -# so no cut-over can happen. Wait until both shapes are up: baseline (workers=1) -# and the target-shape overlap (workers=2) run at once while the realized size -# stays at workers=1. -> ALTER CLUSTER cc_strand SET (SIZE 'scale=1,workers=2') - -> SELECT r.size, count(*) FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_strand' GROUP BY r.size -scale=1,workers=1 1 -scale=1,workers=2 1 - -> SELECT recon.status FROM mz_internal.mz_cluster_reconfigurations recon JOIN mz_clusters ON mz_clusters.id = recon.cluster_id WHERE mz_clusters.name = 'cc_strand' -in-progress - -# Break-glass: disable the controller mid-flight. Nothing now drives the record. -$ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}/materialize -ALTER SYSTEM SET enable_cluster_controller = false - -# The stranding holds (the controller is frozen): still two replicas across the -# two shapes, the realized size still at workers=1, and the record still -# in-progress with nothing on its way to retire either. -> SELECT r.size, count(*) FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_strand' GROUP BY r.size -scale=1,workers=1 1 -scale=1,workers=2 1 -> SELECT size FROM mz_clusters WHERE name = 'cc_strand' -scale=1,workers=1 -> SELECT recon.status FROM mz_internal.mz_cluster_reconfigurations recon JOIN mz_clusters ON mz_clusters.id = recon.cluster_id WHERE mz_clusters.name = 'cc_strand' -in-progress - -# Not the lever: an ALTER back to the realized shape is byte-identical, so with -# the controller off the sequencer short-circuits it as a no-op. The record -# stays in-progress and the overlap replica is not retired. -> ALTER CLUSTER cc_strand SET (SIZE 'scale=1,workers=1') -> SELECT recon.status FROM mz_internal.mz_cluster_reconfigurations recon JOIN mz_clusters ON mz_clusters.id = recon.cluster_id WHERE mz_clusters.name = 'cc_strand' -in-progress -> SELECT count(*) FROM mz_cluster_replicas r JOIN mz_clusters c ON r.cluster_id = c.id WHERE c.name = 'cc_strand' -2 - -# The lever: a config-shape ALTER to a genuinely new size. The legacy path drops -# the whole owned replica set (baseline plus stranded overlap) and recreates a -# dense r1..rN at the new size, cancelling the carried record in the same -# transaction. -> ALTER CLUSTER cc_strand SET (SIZE 'scale=1,workers=4') - -# Desired end state: realized size is the new one, exactly one replica named r1 -# at that size (no orphan left behind), and the stranded record is terminal. -> SELECT size FROM mz_clusters WHERE name = 'cc_strand' -scale=1,workers=4 -> SELECT cluster, replica, size FROM (SHOW CLUSTER REPLICAS) WHERE cluster = 'cc_strand' -cc_strand r1 scale=1,workers=4 -> SELECT recon.status FROM mz_internal.mz_cluster_reconfigurations recon JOIN mz_clusters ON mz_clusters.id = recon.cluster_id WHERE mz_clusters.name = 'cc_strand' -cancelled - -# The lifecycle is audited: the record was started, then cancelled by the -# break-glass resize. No finalize or timeout ever ran. -> SELECT row_number() OVER (ORDER BY id), details->>'transition' FROM mz_catalog.mz_audit_events WHERE event_type = 'alter' AND object_type = 'cluster' AND details->>'cluster_name' = 'cc_strand' AND details->>'transition' IS NOT NULL -1 started -2 cancelled - -> DROP CLUSTER cc_strand CASCADE -> DROP TABLE cc_strand_t - -$ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}/materialize -ALTER SYSTEM SET unsafe_enable_unstable_dependencies = false - # Restore pristine server state (including the tick-interval override). $ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}/materialize -ALTER SYSTEM RESET enable_cluster_controller ALTER SYSTEM RESET cluster_controller_tick_interval ALTER SYSTEM RESET enable_zero_downtime_cluster_reconfiguration ALTER SYSTEM RESET enable_background_alter_cluster