diff --git a/src/adapter/src/coord.rs b/src/adapter/src/coord.rs index 92b7407c5472e..36b3d600398a2 100644 --- a/src/adapter/src/coord.rs +++ b/src/adapter/src/coord.rs @@ -92,10 +92,7 @@ use mz_adapter_types::bootstrap_builtin_cluster_config::BootstrapBuiltinClusterC use mz_adapter_types::compaction::CompactionWindow; use mz_adapter_types::connection::ConnectionId; use mz_adapter_types::dyncfgs::FRONTEND_READ_THEN_WRITE; -use mz_adapter_types::dyncfgs::{ - ENABLE_0DT_HYDRATE_MIGRATED_BUILTIN_MVS, USER_ID_POOL_BATCH_SIZE, - WITH_0DT_DEPLOYMENT_CAUGHT_UP_CHECK_INTERVAL, -}; +use mz_adapter_types::dyncfgs::{ENABLE_0DT_HYDRATE_MIGRATED_BUILTIN_MVS, USER_ID_POOL_BATCH_SIZE}; use mz_auth::password::Password; use mz_build_info::BuildInfo; use mz_catalog::builtin::{ @@ -182,7 +179,7 @@ use timely::progress::{Antichain, Timestamp as _}; use tokio::runtime::Handle as TokioHandle; use tokio::select; use tokio::sync::{Notify, OwnedMutexGuard, Semaphore, mpsc, oneshot, watch}; -use tokio::time::{Interval, MissedTickBehavior}; +use tokio::time::Interval; use tracing::{Instrument, Level, Span, debug, info, info_span, span, warn}; use tracing_opentelemetry::OpenTelemetrySpanExt; use uuid::Uuid; @@ -407,6 +404,7 @@ pub enum Message { ArrangementSizesPrune(Vec), HydrationHistorySchedule, HydrationHistoryRun, + CaughtUpCheck(caught_up::CaughtUpCheckRequest), /// Performs any cleanup and logging actions necessary for /// finalizing a statement execution. RetireExecute { @@ -568,6 +566,7 @@ impl Message { Message::ArrangementSizesPrune(_) => "arrangement_sizes_prune", Message::HydrationHistorySchedule => "hydration_history_schedule", Message::HydrationHistoryRun => "hydration_history_run", + Message::CaughtUpCheck(_) => "caught_up_check", Message::RetireExecute { .. } => "retire_execute", Message::ExecuteSingleStatementTransaction { .. } => { "execute_single_statement_transaction" @@ -2226,12 +2225,9 @@ pub struct Coordinator { /// a timestamp oracle backend is configured. timestamp_oracle_config: Option, - /// When doing 0dt upgrades/in read-only mode, periodically ask all known - /// clusters/collections whether they are caught up. - caught_up_check_interval: Interval, - /// Context needed to check whether all clusters/collections have caught up. - /// Only used during 0dt deployment, while in read-only mode. + /// Only present during 0dt deployment, while in read-only mode, and taken + /// by [`Coordinator::spawn_caught_up_check_task`]. caught_up_check: Option, /// The metrics registry, handed to the catalog info-metrics background task @@ -4193,6 +4189,7 @@ impl Coordinator { self.spawn_statement_logging_task(); self.spawn_catalog_info_metrics_task(); self.spawn_cluster_controller_task(); + self.spawn_caught_up_check_task(); flags::tracing_config(self.catalog.system_config()).apply(&self.tracing_handle); // Report if the handling of a single message takes longer than this threshold. @@ -4337,18 +4334,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.caught_up_check_interval.tick() => { - // We do this directly on the main loop instead of - // firing off a message. We are still in read-only mode, - // so optimizing for latency, not blocking the main loop - // is not that important. - self.maybe_check_caught_up().await; - - continue; - }, // Process the idle metric at the lowest priority to sample queue non-idle time. // `recv()` on `Receiver` is cancellation safe: @@ -5270,26 +5255,6 @@ pub fn serve( let advance_timelines_interval = tokio::time::interval(catalog.system_config().default_timestamp_interval()); - let clusters_caught_up_check_interval = if read_only_controllers { - let dyncfgs = catalog.system_config().dyncfgs(); - let interval = WITH_0DT_DEPLOYMENT_CAUGHT_UP_CHECK_INTERVAL.get(dyncfgs); - - let mut interval = tokio::time::interval(interval); - interval.set_missed_tick_behavior(MissedTickBehavior::Skip); - interval - } else { - // When not in read-only mode, we don't do hydration checks. But we - // still have to provide _some_ interval. This is large enough that - // it doesn't matter. - // - // TODO(aljoscha): We cannot use Duration::MAX right now because of - // https://github.com/tokio-rs/tokio/issues/6634. Use that once it's - // fixed for good. - let mut interval = tokio::time::interval(Duration::from_secs(60 * 60)); - interval.set_missed_tick_behavior(MissedTickBehavior::Skip); - interval - }; - let clusters_caught_up_check = clusters_caught_up_trigger.map(|trigger| { let mut exclude_collections: BTreeSet = @@ -5331,7 +5296,6 @@ pub fn serve( CaughtUpCheckContext { trigger, exclude_collections, - cluster_stability: BTreeMap::new(), } }); @@ -5467,7 +5431,6 @@ pub fn serve( statement_logging: StatementLogging::new(coord_now.clone()), webhook_concurrency_limit, timestamp_oracle_config, - caught_up_check_interval: clusters_caught_up_check_interval, caught_up_check: clusters_caught_up_check, installed_watch_sets: BTreeMap::new(), connection_watch_sets: BTreeMap::new(), diff --git a/src/adapter/src/coord/caught_up.rs b/src/adapter/src/coord/caught_up.rs index 11cc44619972e..28f07804800de 100644 --- a/src/adapter/src/coord/caught_up.rs +++ b/src/adapter/src/coord/caught_up.rs @@ -12,45 +12,100 @@ //! //! During a zero-downtime upgrade the new `environmentd` boots read-only and //! reports "ready to promote" once its clusters have caught up with the leader -//! generation. [`Coordinator::maybe_check_caught_up`] runs that check on an -//! interval (see `with_0dt_deployment_caught_up_check_interval`). We call one -//! such run a "tick", and the term is used throughout this module. +//! generation. A task spawned by [`Coordinator::spawn_caught_up_check_task`] +//! runs that check on an interval (see +//! [`WITH_0DT_DEPLOYMENT_CAUGHT_UP_CHECK_INTERVAL`]). We call one such run a +//! "tick", and the term is used throughout this module. +//! +//! Each tick the task asks the coordinator to classify the clusters, which +//! needs controller state only the coordinator loop can reach. The task then +//! applies the stability gate itself, because the replica queries it needs +//! call back into the coordinator and would deadlock on its loop. //! //! A point-in-time caught-up check is not enough on its own: a crash- or //! OOM-looping replica can momentarily look hydrated and caught-up, and cutting //! over right then drops us straight into a crashing replica. On top of the -//! per-tick caught-up classification we therefore run a stability gate. Once a -//! cluster is genuinely caught-up it must stay caught-up and have all replicas -//! healthy for a configurable period before we report it ready. Any disruption -//! (a replica not `Online`, a status flap between ticks, or a replica restart) -//! resets the streak, so a crash-looping replica never accumulates the required -//! stable time. [`ClusterStabilityState`] holds the per-cluster gate state -//! across ticks. +//! per-tick caught-up classification we therefore run a stability gate. A +//! caught-up cluster is ready only once every replica is `Online` and has had +//! everything it hosts hydrated for a configurable period. +//! +//! Each replica reports its hydration times in its +//! `mz_compute_hydration_times_per_worker` introspection log. The log lives in +//! the replica process. It survives environmentd restarts, because a restarted +//! environmentd reconciles with the same replica processes and they keep +//! compatible dataflows, so a DDL-triggered restart does not restart the +//! period. A dataflow that changed rehydrates and restarts it. The log dies +//! with the process, so a restarted replica has to hydrate again and then wait +//! out the full period, and a crash-looping replica never becomes ready. +//! +//! Hydration times come from the replica's clock, while the period is measured +//! against environmentd's. A replica clock that runs ahead only lengthens the +//! wait, but one that runs behind shortens the period by the skew. We rely on +//! hosts keeping their clocks in sync. +//! +//! A replica without introspection logging cannot report hydration times. It +//! only has to be `Online`. use std::collections::{BTreeMap, BTreeSet}; +use std::fmt::Write as _; +use std::sync::Arc; use std::time::Duration; -use chrono::{DateTime, Utc}; use differential_dataflow::lattice::Lattice as _; -use futures::StreamExt; +use futures::{StreamExt, future}; use itertools::Itertools; use mz_adapter_types::dyncfgs::{ ENABLE_0DT_CAUGHT_UP_REPLICA_STATUS_CHECK, ENABLE_0DT_CAUGHT_UP_STABILITY_CHECK, WITH_0DT_CAUGHT_UP_CHECK_ALLOWED_LAG, WITH_0DT_CAUGHT_UP_CHECK_CUTOFF, - WITH_0DT_CAUGHT_UP_CHECK_STABILITY_PERIOD, + WITH_0DT_CAUGHT_UP_CHECK_STABILITY_PERIOD, WITH_0DT_DEPLOYMENT_CAUGHT_UP_CHECK_INTERVAL, }; use mz_catalog::builtin::{MZ_CLUSTER_REPLICA_FRONTIERS, MZ_CLUSTER_REPLICA_STATUS_HISTORY}; use mz_catalog::memory::objects::Cluster; use mz_compute_client::controller::CollectionReadiness; -use mz_controller::clusters::{ClusterStatus, ProcessId}; +use mz_compute_client::logging::{ComputeLog, LogVariant}; +use mz_controller::clusters::ClusterStatus; use mz_controller_types::{ClusterId, ReplicaId}; use mz_orchestrator::OfflineReason; use mz_ore::channel::trigger::Trigger; use mz_ore::now::EpochMillis; -use mz_repr::{GlobalId, Timestamp}; +use mz_ore::task; +use mz_repr::{GlobalId, Row, Timestamp}; use timely::progress::{Antichain, Timestamp as _}; +use tokio::sync::oneshot; +use tokio::time::MissedTickBehavior; + +use crate::PeekClient; +use crate::coord::{ClusterReplicaStatuses, Coordinator, Message}; -use crate::coord::{ClusterReplicaStatuses, Coordinator}; +/// How long a check waits for a replica to answer the hydration query. +/// +/// A replica that does not answer in time counts as not hydrated for that +/// check. Replicas are asked concurrently, so this also bounds how long a check +/// waits for replicas. +/// +/// The check drops a query that times out, with the leftovers documented at +/// [`PeekClient::background_peek`]. We accept those: a stuck replica holds at +/// most one peek per check, on a log index that barely changes, and an +/// orphaned registration needs the coordinator to stall at exactly that point. +const HYDRATION_QUERY_TIMEOUT: Duration = Duration::from_secs(5); + +/// Returns a query for when a replica last finished hydrating an export. +/// +/// The one row it returns is NULL if some export, other than a transient one +/// or one in `ignored`, is not hydrated on some worker. It returns no row if +/// the replica has reported nothing yet. +fn hydration_query(ignored: &BTreeSet) -> String { + let mut query = "SELECT hydrated_at \ + FROM mz_introspection.mz_compute_hydration_times_per_worker \ + WHERE export_id NOT LIKE 't%'" + .to_string(); + if !ignored.is_empty() { + let ids = ignored.iter().map(|id| format!("'{id}'")).join(", "); + write!(query, " AND export_id NOT IN ({ids})").expect("writing to a string"); + } + query.push_str(" ORDER BY hydrated_at DESC NULLS FIRST LIMIT 1"); + query +} /// Context needed to check whether clusters/collections are caught up. #[derive(Debug)] @@ -63,19 +118,59 @@ pub struct CaughtUpCheckContext { /// When a caught up check is performed as part of a 0dt upgrade, it makes sense to exclude /// collections of newly added builtin objects, as these might not hydrate in read-only mode. pub exclude_collections: BTreeSet, - /// Per-cluster state for the stability gate, retained across checks. - /// - /// Only genuinely caught-up clusters have an entry. Entries are dropped as - /// soon as a cluster stops being caught-up, so the streak restarts from - /// scratch when it becomes caught-up again. - pub cluster_stability: BTreeMap, } -/// How a cluster relates to the 0dt caught-up check on a given tick. +/// A request from the caught-up check task for the coordinator's view of the +/// clusters, answered by [`Coordinator::handle_caught_up_check_request`]. +#[derive(Debug)] +pub struct CaughtUpCheckRequest { + exclude_collections: Arc>, + tx: oneshot::Sender, +} + +/// The coordinator's view of the clusters at one check. +#[derive(Debug)] +pub struct CaughtUpSnapshot { + now: EpochMillis, + /// Whether every cluster is caught up or ignored. + all_caught_up: bool, + stability_check_enabled: bool, + stability_period_ms: u64, + /// Every caught-up cluster. Empty when the stability check is disabled. + caught_up_clusters: BTreeMap, +} + +/// A caught-up cluster, as the stability gate needs to know it. +#[derive(Debug)] +struct CaughtUpCluster { + /// Compute collections the classification ignored, which therefore need not + /// hydrate. + ignored_compute_collections: BTreeSet, + replicas: BTreeMap, +} + +/// What the coordinator knows about a replica of a caught-up cluster, and so +/// whether the stability gate has to ask it. #[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ReplicaTarget { + /// See [`ReplicaHealth::NotOnline`]. + NotOnline, + /// See [`ReplicaHealth::Exempt`]. + Exempt, + /// The replica is `Online` and has to report its hydration times. + Ask, +} + +/// How a cluster relates to the 0dt caught-up check on a given tick. +#[derive(Debug, Clone, PartialEq, Eq)] enum ClusterCaughtUpStatus { - /// Genuinely hydrated and within lag. Subject to the stability gate. - CaughtUp, + /// Hydrated and within lag, apart from `ignored_compute_collections`. + /// Subject to the stability gate. + CaughtUp { + /// Compute collections the classification ignored, see + /// [`CaughtUpCluster::ignored_compute_collections`]. + ignored_compute_collections: BTreeSet, + }, /// Excluded by the existing checks (no replicas, or hopelessly behind with /// only crash/OOM-looping replicas). Does not block readiness and is not /// health-gated, so we keep ignoring clusters that are already unhealthy in @@ -85,180 +180,257 @@ enum ClusterCaughtUpStatus { NotCaughtUp, } -/// Per-cluster state for the stability gate, retained across caught-up checks. -/// -/// The gate requires a cluster to stay caught-up and fully healthy for a -/// configurable period before we report it ready. A point-in-time check isn't -/// enough: a crash-looping replica can momentarily look hydrated and healthy, so -/// we'd cut over right into a crash. We therefore track health over time here. -#[derive(Debug, Default, Clone)] -pub struct ClusterStabilityState { - /// Wall-clock time (environmentd's `now()`) at which the current - /// uninterrupted caught-up-and-healthy streak began. `None` while the - /// cluster isn't currently in such a streak. - /// - /// We anchor the window with environmentd's clock so the configured period - /// means real wall-clock seconds, independent of orchestrator event times. - stable_since: Option, - /// Max replica-process status-change time observed on the previous tick. - /// - /// Used to detect status transitions that happened and resolved between two - /// ticks (a fast flap we'd otherwise miss by only sampling the current - /// status). This is an orchestrator-supplied timestamp (`process.time`), not - /// a locally measured one, which is why it's a `DateTime` and not an - /// `Instant`. We only ever compare these orchestrator times against each - /// other, so orchestrator/environmentd clock skew doesn't matter. - last_status_change: Option>, - /// Restart count per replica process observed on the previous tick. - /// - /// Any difference from this tick resets the streak: an increased count means - /// a restart, a decreased one means the process was recreated, and an added - /// or removed key means replica/process churn. Restart counts survive gaps - /// in the orchestrator watch, so they catch restarts the status stream can - /// drop. We track them per process rather than as a cluster-wide sum so that - /// offsetting changes across processes can't cancel out and hide a restart. - last_restart_counts: Option>, -} - -/// A point-in-time view of a cluster's replica health, derived from the -/// in-memory mirror of orchestrator-reported replica statuses. -#[derive(Debug, Clone)] -struct ClusterHealthSnapshot { - /// True iff the cluster has replicas and every process of every replica is - /// `Online`. We deliberately require all replicas to be healthy, so we only - /// cut over when the new environment is fully healthy. - all_healthy: bool, - /// Max status-change time across all of the cluster's replica processes. - max_status_change: Option>, - /// Restart count per replica process. - /// - /// Kept per process rather than summed: restart counts are not monotonic (a - /// recreated process resets to zero), so a cluster-wide sum could cancel - /// offsetting changes across processes and hide a restart. Comparing the - /// whole map between ticks also catches replica/process churn. - restart_counts: BTreeMap<(ReplicaId, ProcessId), u64>, +/// What the stability gate knows about one replica on a given tick. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ReplicaHealth { + /// The replica is not currently `Online`. + NotOnline, + /// The replica is `Online`, but has not hydrated everything it hosts, or + /// did not answer the hydration query. + NotHydrated, + /// The replica has been healthy since this wall-clock time. + HealthySince(EpochMillis), + /// The replica is `Online` but cannot report hydration times, so there is + /// nothing to wait on. + Exempt, } /// Why a caught-up cluster is being held back by the stability gate on a given -/// tick. -/// -/// Only ever set when the cluster is not yet ready, so there is no "stable" -/// variant. Recorded so we can log the cause. +/// tick. Recorded so we can log the cause. #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum StabilityBlocker { /// Not all replicas are currently `Online`. - NotHealthy, - /// A status change happened and resolved between two ticks. - StatusFlapped, - /// A replica process restarted between two ticks. - Restarted, - /// Currently caught-up and healthy, but the streak hasn't reached the - /// required period yet. + NotOnline, + /// Some replica has not hydrated everything it hosts. + NotHydrated, + /// Every replica is healthy, but not yet for the required period. WithinPeriod, } -/// Outcome of folding one health snapshot into a [`ClusterStabilityState`]. +/// Outcome of folding a cluster's replica health. #[derive(Debug, Clone, Copy)] struct StabilityObservation { - /// Whether the cluster has now been continuously caught-up and healthy for - /// at least the required period. - ready: bool, - /// How long the current uninterrupted streak has lasted, in milliseconds. - /// `None` when the cluster is not currently in a streak (this tick reset it). + /// How long the cluster has been stable, in milliseconds. `None` when some + /// replica has no evidence of health. stable_for_ms: Option, - /// Why the cluster is being held back, for logging. `None` once it's ready. + /// Why the cluster is held back. `None` once it is ready. blocked_by: Option, } -impl ClusterStabilityState { - /// Folds in the latest health snapshot and returns an observation: whether - /// the cluster has now been continuously caught-up and healthy for at least - /// `period_ms`, how long the current streak has lasted, and (when not ready) - /// what is holding it back. - /// - /// A cluster is "good" on a tick only if all its replicas are currently - /// healthy and nothing changed since the previous tick (no status flap, no - /// restart). Any disruption resets the streak, so a crash-looping replica can - /// never accumulate the required stable time. - fn observe( - &mut self, - snapshot: &ClusterHealthSnapshot, - now: EpochMillis, - period_ms: u64, - ) -> StabilityObservation { - // NOTE: We don't assume orchestrator status events arrive in order or - // that every process of a cluster reports within the same tick. The - // snapshot reflects whatever the in-memory mirror holds right now, and - // the three checks below are deliberately redundant so no single one has - // to be reliable on its own: - // - // - `all_healthy` is a point-in-time check, independent of ordering. - // - a change in the per-process `restart_counts` is the durable signal: - // k8s reports restart counts and they survive gaps in the orchestrator - // watch, so they catch restarts the status stream drops. We compare - // the whole map, never a cluster-wide sum: restart counts are not - // monotonic (a recreated process resets to zero), so a sum could - // cancel offsetting changes across processes and hide a restart. - // - `max_status_change` advancing is a best-effort flap detector. A - // cluster-wide max is enough here, unlike the restart counts, because - // any status change stamps `process.time` at ~now, so a flap pushes - // the max past the previous tick's value. It can still miss a flap if - // an out-of-order event reports an older time, which is why the - // restart counts are the belt-and-suspenders. - // - // We only compare the orchestrator-supplied times against each other, so - // clock skew between the orchestrator and environmentd doesn't matter. - let status_flapped = match (self.last_status_change, snapshot.max_status_change) { - (Some(prev), Some(cur)) => cur > prev, - _ => false, - }; - let restarted = self - .last_restart_counts - .as_ref() - .is_some_and(|prev| prev != &snapshot.restart_counts); - - let good = snapshot.all_healthy && !status_flapped && !restarted; +/// Folds the health of a cluster's replicas into the gate's verdict. +/// +/// A cluster is stable since the latest of its replicas' healthy-since times. +/// These can come from replica clocks, which may run ahead of `now`, so they +/// are clamped to `now`. A replica clock running behind shortens the period by +/// its skew. +fn observe_stability( + replicas: &[ReplicaHealth], + now: EpochMillis, + period_ms: u64, +) -> StabilityObservation { + let blocked = |blocker| StabilityObservation { + stable_for_ms: None, + blocked_by: Some(blocker), + }; + if replicas.is_empty() || replicas.contains(&ReplicaHealth::NotOnline) { + return blocked(StabilityBlocker::NotOnline); + } + let mut stable_since = EpochMillis::MIN; + for health in replicas { + match health { + ReplicaHealth::HealthySince(since) => stable_since = stable_since.max(*since), + ReplicaHealth::Exempt => {} + ReplicaHealth::NotOnline | ReplicaHealth::NotHydrated => { + return blocked(StabilityBlocker::NotHydrated); + } + } + } + let stable_for_ms = now.saturating_sub(stable_since.min(now)); + StabilityObservation { + stable_for_ms: Some(stable_for_ms), + blocked_by: (stable_for_ms < period_ms).then_some(StabilityBlocker::WithinPeriod), + } +} - self.stable_since = if good { - self.stable_since.or(Some(now)) - } else { - None - }; - self.last_status_change = snapshot.max_status_change; - self.last_restart_counts = Some(snapshot.restart_counts.clone()); +/// The part of the caught-up check that runs in its own task: it asks replicas +/// for their hydration times and applies the stability gate. +struct StabilityGate { + client: PeekClient, +} - let stable_for_ms = self.stable_since.map(|since| now.saturating_sub(since)); - let ready = stable_for_ms.is_some_and(|elapsed| elapsed >= period_ms); +impl StabilityGate { + /// Returns whether `snapshot` shows every cluster ready for promotion. + async fn ready(&self, snapshot: CaughtUpSnapshot) -> bool { + let CaughtUpSnapshot { + now, + all_caught_up, + stability_check_enabled, + stability_period_ms, + caught_up_clusters, + } = snapshot; + // Disabling the stability check is a fleet-wide break-glass that makes + // caught-up clusters ready without waiting on replica health. While + // some cluster is not caught up, readiness is blocked anyway, so no + // replica needs to be asked. + if !all_caught_up || !stability_check_enabled { + tracing::info!(all_ready = %all_caught_up, "checked caught-up status of clusters"); + return all_caught_up; + } - let blocked_by = if ready { - None - } else if !snapshot.all_healthy { - Some(StabilityBlocker::NotHealthy) - } else if status_flapped { - Some(StabilityBlocker::StatusFlapped) - } else if restarted { - Some(StabilityBlocker::Restarted) - } else { - Some(StabilityBlocker::WithinPeriod) - }; + let answers = self.ask_replicas(&caught_up_clusters).await; - StabilityObservation { - ready, - stable_for_ms, - blocked_by, + let mut all_ready = true; + for (cluster_id, cluster) in caught_up_clusters { + let health: BTreeMap<_, _> = cluster + .replicas + .iter() + .map(|(replica_id, target)| { + let replica_health = match target { + ReplicaTarget::NotOnline => ReplicaHealth::NotOnline, + ReplicaTarget::Exempt => ReplicaHealth::Exempt, + ReplicaTarget::Ask => answers + .get(replica_id) + .copied() + .unwrap_or(ReplicaHealth::NotHydrated), + }; + (*replica_id, replica_health) + }) + .collect(); + + let replica_health = health.values().copied().collect_vec(); + let observation = observe_stability(&replica_health, now, stability_period_ms); + if let Some(reason) = observation.blocked_by { + all_ready = false; + tracing::info!( + %cluster_id, + ?reason, + replicas = ?health, + stable_for_ms = ?observation.stable_for_ms, + required_period_ms = stability_period_ms, + "cluster is caught up but not yet stable for the required period" + ); + } } + + tracing::info!(%all_ready, "checked caught-up status of clusters"); + all_ready + } + + /// Asks every [`ReplicaTarget::Ask`] replica, concurrently, when it last + /// finished hydrating an export. + async fn ask_replicas( + &self, + clusters: &BTreeMap, + ) -> BTreeMap { + let queries = clusters.iter().flat_map(|(&cluster_id, cluster)| { + let sql = hydration_query(&cluster.ignored_compute_collections); + cluster + .replicas + .iter() + .filter(|(_, target)| **target == ReplicaTarget::Ask) + .map(move |(&replica_id, _)| { + let mut client = self.client.clone(); + let sql = sql.clone(); + async move { + let query = client.background_peek(&sql, cluster_id, replica_id); + let answer = tokio::time::timeout(HYDRATION_QUERY_TIMEOUT, query).await; + let health = match answer { + Ok(Ok(rows)) => replica_health(&rows), + Ok(Err(error)) => { + tracing::warn!( + %cluster_id, %replica_id, %error, "hydration query failed" + ); + ReplicaHealth::NotHydrated + } + Err(_elapsed) => { + tracing::warn!(%cluster_id, %replica_id, "hydration query timed out"); + ReplicaHealth::NotHydrated + } + }; + (replica_id, health) + } + }) + }); + future::join_all(queries).await.into_iter().collect() } } +/// Interprets the rows of a [`hydration_query`]. +fn replica_health(rows: &[Row]) -> ReplicaHealth { + let Some(row) = rows.first() else { + return ReplicaHealth::NotHydrated; + }; + let hydrated_at = row.unpack_first(); + if hydrated_at.is_null() { + return ReplicaHealth::NotHydrated; + } + let millis = hydrated_at.unwrap_timestamptz().timestamp_millis(); + // A time before the epoch is nonsense, and must not read as stable for + // decades. + u64::try_from(millis).map_or(ReplicaHealth::NotHydrated, ReplicaHealth::HealthySince) +} + impl Coordinator { - /// Checks that all clusters/collections are caught up. If so, this will - /// trigger `self.caught_up_check.trigger`. + /// Spawns the task that checks, on an interval, whether all + /// clusters/collections are caught up, and fires the trigger of + /// `self.caught_up_check` once they are. /// - /// This method is a no-op when the trigger has already been fired. - pub async fn maybe_check_caught_up(&mut self) { - if self.caught_up_check.is_none() { + /// A no-op when there is no caught-up check, i.e. outside read-only mode. + pub(crate) fn spawn_caught_up_check_task(&mut self) { + let Some(CaughtUpCheckContext { + trigger, + exclude_collections, + }) = self.caught_up_check.take() + else { return; - } + }; + let exclude_collections = Arc::new(exclude_collections); + let period = WITH_0DT_DEPLOYMENT_CAUGHT_UP_CHECK_INTERVAL + .get(self.catalog().system_config().dyncfgs()); + let internal_cmd_tx = self.internal_cmd_tx.clone(); + let gate = StabilityGate { + client: self.background_peek_client(&self.owned_catalog()), + }; + + task::spawn(|| "caught_up_check", async move { + let mut interval = tokio::time::interval(period); + interval.set_missed_tick_behavior(MissedTickBehavior::Skip); + loop { + interval.tick().await; + let (tx, rx) = oneshot::channel(); + let request = CaughtUpCheckRequest { + exclude_collections: Arc::clone(&exclude_collections), + tx, + }; + // A failed send or a dropped reply means the coordinator is + // gone. + // + // NOTE: Returning drops `trigger`, which fires it, so a + // coordinator shutdown reads as caught up. + if internal_cmd_tx + .send(Message::CaughtUpCheck(request)) + .is_err() + { + return; + } + let Ok(snapshot) = rx.await else { + return; + }; + if gate.ready(snapshot).await { + trigger.fire(); + return; + } + } + }); + } + + /// Answers a [`CaughtUpCheckRequest`] with the classification of every + /// cluster and the replicas the stability gate has to hear from. + pub(crate) async fn handle_caught_up_check_request(&self, request: CaughtUpCheckRequest) { + let CaughtUpCheckRequest { + exclude_collections, + tx, + } = request; let replica_frontier_item_id = self .catalog() @@ -316,9 +488,9 @@ impl Coordinator { }) .collect_vec(); - // We care about each collection being hydrated on _some_ - // replica. We don't check that at least one replica has all - // collections of that cluster hydrated. + // The classification only requires each collection to be hydrated on + // _some_ replica. The stability gate separately requires every replica + // to have hydrated everything it hosts. let live_collection_frontiers: BTreeMap<_, _> = live_frontiers .into_iter() .map(|(oid, _replica_id, upper_ts)| (oid, upper_ts)) @@ -369,16 +541,6 @@ impl Coordinator { // and an operator can still force it via skip-catchup. let stability_period_ms = u64::try_from(stability_period.as_millis()).unwrap_or(u64::MAX); - // We clone the exclude set so we don't hold a borrow of `caught_up_check` - // across the classification, which lets us update the per-cluster - // stability state on it (mutably) afterwards. - let exclude_collections = self - .caught_up_check - .as_ref() - .expect("known to exist") - .exclude_collections - .clone(); - let classification = self .classify_clusters( allowed_lag.into(), @@ -390,111 +552,68 @@ impl Coordinator { ) .await; - // Read the health snapshots for genuinely caught-up clusters now, while we - // only hold a shared borrow of `self`. We update the stability state in a - // separate, mutable pass below. - let health: BTreeMap = classification - .iter() - .filter(|(_, status)| **status == ClusterCaughtUpStatus::CaughtUp) - .map(|(&cluster_id, _)| (cluster_id, self.cluster_health(cluster_id))) - .collect(); - - let ctx = self.caught_up_check.as_mut().expect("known to exist"); + let all_caught_up = classification + .values() + .all(|status| *status != ClusterCaughtUpStatus::NotCaughtUp); + let caught_up_clusters = if stability_check_enabled { + classification + .into_iter() + .filter_map(|(cluster_id, status)| match status { + ClusterCaughtUpStatus::CaughtUp { + ignored_compute_collections, + } => Some(( + cluster_id, + CaughtUpCluster { + ignored_compute_collections, + replicas: self.replica_targets(cluster_id), + }, + )), + _ => None, + }) + .collect() + } else { + BTreeMap::new() + }; - // Drop stability state for clusters that are no longer genuinely caught - // up, so the streak restarts from scratch when they become caught-up - // again. - ctx.cluster_stability.retain(|cluster_id, _| { - classification.get(cluster_id) == Some(&ClusterCaughtUpStatus::CaughtUp) + // A dropped receiver means the task is gone, which leaves nothing to + // answer. + let _ = tx.send(CaughtUpSnapshot { + now, + all_caught_up, + stability_check_enabled, + stability_period_ms, + caught_up_clusters, }); - - let mut all_ready = true; - for (&cluster_id, status) in &classification { - match status { - ClusterCaughtUpStatus::Ignored => {} - ClusterCaughtUpStatus::NotCaughtUp => { - all_ready = false; - } - ClusterCaughtUpStatus::CaughtUp => { - // Break-glass: when disabled, a caught-up cluster is - // immediately ready, with no replica-health requirement, - // i.e. the behavior from before this gate existed. We keep it - // as a config-level, fleet-wide revert. Operators can already - // force a single cutover via skip-catchup/promote, but this - // flag restores prior auto-cutover behavior across all - // environments without per-deploy manual intervention or a - // code release, mirroring - // `enable_0dt_caught_up_replica_status_check`. - if !stability_check_enabled { - continue; - } - let snapshot = health.get(&cluster_id).expect("computed above"); - let state = ctx.cluster_stability.entry(cluster_id).or_default(); - let observation = state.observe(snapshot, now, stability_period_ms); - if !observation.ready { - all_ready = false; - tracing::info!( - %cluster_id, - reason = ?observation.blocked_by, - all_healthy = snapshot.all_healthy, - stable_for_ms = ?observation.stable_for_ms, - required_period_ms = stability_period_ms, - max_status_change = ?snapshot.max_status_change, - // Summed only for a readable log line. The gate - // compares the per-process map, not this total. - restart_total = snapshot.restart_counts.values().sum::(), - "cluster is caught up but not yet stable for the required period" - ); - } - } - } - } - - tracing::info!(%all_ready, "checked caught-up status of clusters"); - - if all_ready { - let ctx = self.caught_up_check.take().expect("known to exist"); - ctx.trigger.fire(); - } } - /// Reads the current health of a cluster's replicas from the in-memory - /// mirror of orchestrator-reported statuses. - /// - /// A cluster with no replica status entries (e.g. a freshly created cluster - /// whose statuses haven't been initialized) is reported as not healthy. - fn cluster_health(&self, cluster_id: ClusterId) -> ClusterHealthSnapshot { - let Some(replicas) = self - .cluster_replica_statuses - .try_get_cluster_statuses(cluster_id) - .filter(|replicas| !replicas.is_empty()) - else { - // A cluster with no replica statuses is treated as not healthy. - return ClusterHealthSnapshot { - all_healthy: false, - max_status_change: None, - restart_counts: BTreeMap::new(), - }; - }; - - let mut all_healthy = true; - let mut max_status_change = None; - let mut restart_counts = BTreeMap::new(); - for (replica_id, processes) in replicas { - if ClusterReplicaStatuses::cluster_replica_status(processes) != ClusterStatus::Online { - all_healthy = false; - } - for (process_id, process) in processes { - max_status_change = max_status_change.max(Some(process.time)); - restart_counts.insert((*replica_id, *process_id), process.restart_count); - } - } - - ClusterHealthSnapshot { - all_healthy, - max_status_change, - restart_counts, - } + /// Classifies the replicas of a cluster for the stability gate. + fn replica_targets(&self, cluster_id: ClusterId) -> BTreeMap { + let cluster = self.catalog().get_cluster(cluster_id); + let has_hydration_log = cluster + .log_indexes + .contains_key(&LogVariant::Compute(ComputeLog::HydrationTime)); + cluster + .replicas() + .map(|replica| { + let online = self + .cluster_replica_statuses + .try_get_cluster_replica_statuses(cluster_id, replica.replica_id) + .is_some_and(|processes| { + !processes.is_empty() + && ClusterReplicaStatuses::cluster_replica_status(processes) + == ClusterStatus::Online + }); + let logging = has_hydration_log && replica.config.compute.logging.enabled(); + let target = if !online { + ReplicaTarget::NotOnline + } else if !logging { + ReplicaTarget::Exempt + } else { + ReplicaTarget::Ask + }; + (replica.replica_id, target) + }) + .collect() } /// Classifies every cluster for the caught-up check. @@ -587,6 +706,12 @@ impl Coordinator { } let mut all_caught_up = true; + let mut ignored_compute_collections: BTreeSet<_> = self + .controller + .compute + .collection_ids(cluster.id)? + .filter(|id| exclude_collections.contains(id)) + .collect(); let storage_frontiers = self .controller @@ -704,6 +829,9 @@ impl Coordinator { "live write frontier of collection {id} is too far behind 'now'; \ ignoring for caught-up checks" ); + if matches!(collection_type, CollectionType::Compute) { + ignored_compute_collections.insert(id); + } continue; } @@ -731,6 +859,11 @@ impl Coordinator { // We don't expect collections to get hydrated, ingestions to be // started, etc. when they are already at the empty write frontier. if live_write_frontier.is_empty() || readiness == CollectionReadiness::Ready { + if readiness != CollectionReadiness::Ready + && matches!(collection_type, CollectionType::Compute) + { + ignored_compute_collections.insert(id); + } // This is a bit spammy, but log caught-up collections while we // investigate why environments are cutting over but then a lot // of compute collections are _not_ in fact hydrated on @@ -762,7 +895,9 @@ impl Coordinator { } Ok(if all_caught_up { - ClusterCaughtUpStatus::CaughtUp + ClusterCaughtUpStatus::CaughtUp { + ignored_compute_collections, + } } else { ClusterCaughtUpStatus::NotCaughtUp }) @@ -914,143 +1049,69 @@ impl Coordinator { #[cfg(test)] mod tests { + use mz_repr::Datum; + use mz_repr::adt::timestamp::CheckedTimestamp; + use super::*; - /// Builds a health snapshot with all restarts attributed to a single - /// replica process. `change_secs` is the max status-change time as a - /// unix-second offset, `restarts` that process's restart count. - fn snapshot(all_healthy: bool, change_secs: i64, restarts: u64) -> ClusterHealthSnapshot { - ClusterHealthSnapshot { - all_healthy, - max_status_change: DateTime::from_timestamp(change_secs, 0), - restart_counts: BTreeMap::from([((ReplicaId::User(1), 0), restarts)]), - } - } + use ReplicaHealth::{Exempt, HealthySince, NotHydrated, NotOnline}; #[mz_ore::test] - fn stability_requires_sustained_health() { + fn stable_since_latest_replica() { let period_ms = 1000; - let mut state = ClusterStabilityState::default(); - - // The first healthy observation starts the streak but isn't yet stable. - assert!(!state.observe(&snapshot(true, 100, 0), 0, period_ms).ready); - // Still within the period. - assert!(!state.observe(&snapshot(true, 100, 0), 500, period_ms).ready); - // Past the period: ready. - assert!( - state - .observe(&snapshot(true, 100, 0), 1000, period_ms) - .ready - ); - } - - #[mz_ore::test] - fn unhealthy_resets_streak() { - let period_ms = 1000; - let mut state = ClusterStabilityState::default(); - - assert!(!state.observe(&snapshot(true, 100, 0), 0, period_ms).ready); - // A currently-unhealthy observation resets the streak. - assert!( - !state - .observe(&snapshot(false, 100, 0), 500, period_ms) - .ready - ); - // Healthy again, but the clock restarts from here. - assert!(!state.observe(&snapshot(true, 100, 0), 600, period_ms).ready); - assert!( - !state - .observe(&snapshot(true, 100, 0), 1599, period_ms) - .ready - ); - assert!( - state - .observe(&snapshot(true, 100, 0), 1600, period_ms) - .ready + let replicas = [HealthySince(100), HealthySince(500), Exempt]; + let observation = observe_stability(&replicas, 1499, period_ms); + assert_eq!(observation.stable_for_ms, Some(999)); + assert_eq!(observation.blocked_by, Some(StabilityBlocker::WithinPeriod)); + assert_eq!( + observe_stability(&replicas, 1500, period_ms).blocked_by, + None ); } #[mz_ore::test] - fn status_flap_between_ticks_resets_streak() { - let period_ms = 1000; - let mut state = ClusterStabilityState::default(); - - assert!(!state.observe(&snapshot(true, 100, 0), 0, period_ms).ready); - // Currently healthy, but the status-change time advanced, so a flap - // happened and resolved between ticks: reset. - assert!( - !state - .observe(&snapshot(true, 200, 0), 1000, period_ms) - .ready - ); - // A clean streak from here. - assert!( - !state - .observe(&snapshot(true, 200, 0), 1500, period_ms) - .ready - ); - assert!( - state - .observe(&snapshot(true, 200, 0), 2500, period_ms) - .ready - ); + fn every_replica_needs_evidence() { + // `NotOnline` wins, because it is the more fundamental problem. + for (replicas, blocker) in [ + (vec![], StabilityBlocker::NotOnline), + ( + vec![HealthySince(0), NotOnline], + StabilityBlocker::NotOnline, + ), + (vec![NotHydrated, NotOnline], StabilityBlocker::NotOnline), + ( + vec![HealthySince(0), NotHydrated], + StabilityBlocker::NotHydrated, + ), + (vec![Exempt, NotHydrated], StabilityBlocker::NotHydrated), + ] { + // Even a zero period requires evidence from every replica. + let observation = observe_stability(&replicas, 10_000, 0); + assert_eq!(observation.blocked_by, Some(blocker), "{replicas:?}"); + assert_eq!(observation.stable_for_ms, None, "{replicas:?}"); + } } #[mz_ore::test] - fn restart_between_ticks_resets_streak() { - let period_ms = 1000; - let mut state = ClusterStabilityState::default(); - - assert!(!state.observe(&snapshot(true, 100, 3), 0, period_ms).ready); - // Healthy with the same status-change time, but the restart count went - // up: a restart happened and recovered between ticks, which the status - // alone would miss. Reset. - assert!( - !state - .observe(&snapshot(true, 100, 4), 1000, period_ms) - .ready - ); - assert!( - !state - .observe(&snapshot(true, 100, 4), 1500, period_ms) - .ready - ); - assert!( - state - .observe(&snapshot(true, 100, 4), 2500, period_ms) - .ready - ); + fn replica_clock_ahead_is_clamped() { + let observation = observe_stability(&[HealthySince(2000)], 1000, 0); + assert_eq!(observation.stable_for_ms, Some(0)); + assert_eq!(observation.blocked_by, None); } #[mz_ore::test] - fn offsetting_restart_changes_reset_streak() { - // Two processes whose restart counts move in opposite directions by the - // same amount. A cluster-wide sum would be unchanged and miss the - // restart, but the per-process map differs, so the streak resets. - let period_ms = 1000; - let mut state = ClusterStabilityState::default(); - - let r = ReplicaId::User(1); - let snapshot = |a: u64, b: u64| ClusterHealthSnapshot { - all_healthy: true, - max_status_change: DateTime::from_timestamp(100, 0), - restart_counts: BTreeMap::from([((r, 0u64), a), ((r, 1u64), b)]), + fn hydration_answer_fails_closed() { + let answer = |millis| { + let time = chrono::DateTime::from_timestamp_millis(millis).expect("valid time"); + let time = CheckedTimestamp::from_timestamplike(time).expect("in range"); + vec![Row::pack_slice(&[Datum::TimestampTz(time)])] }; - - // Start a streak with per-process counts summing to 2. - assert!(!state.observe(&snapshot(1, 1), 0, period_ms).ready); - // One process restarts (+1) while the other is recreated (-1). The sum - // is still 2, but the per-process map changed: reset. - assert!(!state.observe(&snapshot(2, 0), 1000, period_ms).ready); - // A clean streak from here. - assert!(!state.observe(&snapshot(2, 0), 1500, period_ms).ready); - assert!(state.observe(&snapshot(2, 0), 2500, period_ms).ready); - } - - #[mz_ore::test] - fn zero_period_ready_on_first_healthy_tick() { - let mut state = ClusterStabilityState::default(); - // With a zero period a single clean, healthy observation is enough. - assert!(state.observe(&snapshot(true, 100, 0), 0, 0).ready); + assert_eq!(replica_health(&answer(1000)), HealthySince(1000)); + assert_eq!(replica_health(&answer(-1)), NotHydrated, "pre-epoch time"); + assert_eq!( + replica_health(&[Row::pack_slice(&[Datum::Null])]), + NotHydrated + ); + assert_eq!(replica_health(&[]), NotHydrated, "nothing reported yet"); } } diff --git a/src/adapter/src/coord/hydration_history.rs b/src/adapter/src/coord/hydration_history.rs index 26087a805cdff..6caf7c1b88596 100644 --- a/src/adapter/src/coord/hydration_history.rs +++ b/src/adapter/src/coord/hydration_history.rs @@ -33,8 +33,8 @@ use std::time::{Duration, Instant}; use itertools::Itertools; use mz_adapter_types::dyncfgs::{ - FRONTEND_READ_THEN_WRITE, HYDRATION_HISTORY_COLLECTION_INTERVAL, - HYDRATION_HISTORY_RETENTION_PERIOD, REPLICA_HYDRATION_HISTORY_RETENTION_PERIOD, + HYDRATION_HISTORY_COLLECTION_INTERVAL, HYDRATION_HISTORY_RETENTION_PERIOD, + REPLICA_HYDRATION_HISTORY_RETENTION_PERIOD, }; use mz_catalog::builtin::{ MZ_CATALOG_SERVER_CLUSTER, MZ_OBJECT_HYDRATION_HISTORY, MZ_REPLICA_HYDRATION_HISTORY, @@ -55,7 +55,6 @@ use crate::catalog::Catalog; use crate::command::ExecuteResponse; use crate::coord::{Coordinator, Message}; use crate::metrics::Metrics; -use crate::peek_client::CoordinatorClient; use crate::session::Session; use crate::{AdapterError, PeekClient}; @@ -266,26 +265,7 @@ impl Coordinator { let dyncfgs = catalog.system_config().dyncfgs(); let object_cutoff = cutoff(HYDRATION_HISTORY_RETENTION_PERIOD.get(dyncfgs)); let replica_cutoff = cutoff(REPLICA_HYDRATION_HISTORY_RETENTION_PERIOD.get(dyncfgs)); - let build_version = catalog.state().config().build_info.human_version(None); - // Background read-then-write always uses the frontend OCC path. This - // shared constructor field only controls session fallback, so the flag - // does not gate history collection. - let client = PeekClient::new( - CoordinatorClient::Background { - tx: self.internal_cmd_tx.clone(), - metrics: self.metrics.clone(), - }, - &catalog, - Arc::clone(&self.controller.storage_collections), - Arc::clone(&self.transient_id_gen), - self.optimizer_metrics.clone(), - self.persist_client.clone(), - self.statement_logging.create_frontend(build_version), - Arc::clone(&self.occ_write_semaphore), - FRONTEND_READ_THEN_WRITE.get(self.catalog().system_config().dyncfgs()), - self.group_commit_tx.clone(), - self.controller.read_only(), - ); + let client = self.background_peek_client(&catalog); Sweep { client, object_history_id: catalog.resolve_builtin_table(&MZ_OBJECT_HYDRATION_HISTORY), diff --git a/src/adapter/src/coord/message_handler.rs b/src/adapter/src/coord/message_handler.rs index b17378da5c2d5..c35e8f1140daf 100644 --- a/src/adapter/src/coord/message_handler.rs +++ b/src/adapter/src/coord/message_handler.rs @@ -191,6 +191,9 @@ impl Coordinator { Message::HydrationHistoryRun => { self.run_hydration_history_collection(); } + Message::CaughtUpCheck(request) => { + self.handle_caught_up_check_request(request).await; + } Message::RetireExecute { otel_ctx, data, diff --git a/src/adapter/src/coord/peek.rs b/src/adapter/src/coord/peek.rs index a8d653495210f..57baa8816b8f6 100644 --- a/src/adapter/src/coord/peek.rs +++ b/src/adapter/src/coord/peek.rs @@ -21,6 +21,7 @@ use std::sync::Arc; use differential_dataflow::consolidation::consolidate; use itertools::Itertools; use mz_adapter_types::connection::ConnectionId; +use mz_adapter_types::dyncfgs::FRONTEND_READ_THEN_WRITE; use mz_cluster_client::ReplicaId; use mz_compute_client::controller::PeekNotification; use mz_compute_client::protocol::command::PeekTarget; @@ -56,11 +57,13 @@ use tracing::{Instrument, Span}; use uuid::Uuid; use crate::active_compute_sink::{ActiveComputeSink, ActiveCopyTo}; +use crate::catalog::Catalog; use crate::coord::timestamp_selection::TimestampDetermination; use crate::optimize::OptimizerError; +use crate::peek_client::CoordinatorClient; use crate::statement_logging::WatchSetCreation; use crate::statement_logging::{StatementEndedExecutionReason, StatementExecutionStrategy}; -use crate::{AdapterError, ExecuteContextGuard, ExecuteResponse}; +use crate::{AdapterError, ExecuteContextGuard, ExecuteResponse, PeekClient}; /// A peek is a request to read data from a maintained arrangement. #[derive(Debug)] @@ -1025,6 +1028,34 @@ impl crate::coord::Coordinator { }) } + /// Returns a [`PeekClient`] for coordinator-owned queries, which have to + /// run off the main loop because the client calls back into it. + /// + /// The client holds no session [`Client`](crate::Client), so it does not + /// keep the coordinator alive. + pub(crate) fn background_peek_client(&self, catalog: &Arc) -> PeekClient { + let build_version = catalog.state().config().build_info.human_version(None); + PeekClient::new( + CoordinatorClient::Background { + tx: self.internal_cmd_tx.clone(), + metrics: self.metrics.clone(), + }, + catalog, + Arc::clone(&self.controller.storage_collections), + Arc::clone(&self.transient_id_gen), + self.optimizer_metrics.clone(), + self.persist_client.clone(), + self.statement_logging.create_frontend(build_version), + Arc::clone(&self.occ_write_semaphore), + // Background read-then-write always uses the frontend OCC path. + // This field only controls session fallback, so the flag does not + // gate background work. + FRONTEND_READ_THEN_WRITE.get(catalog.system_config().dyncfgs()), + self.group_commit_tx.clone(), + self.controller.read_only(), + ) + } + /// Creates an async stream that processes peek responses and yields rows. /// /// TODO(peek-seq): Move this out of `coord` once we delete the old peek sequencing. diff --git a/src/adapter/src/frontend_peek.rs b/src/adapter/src/frontend_peek.rs index c8d2e3b89009c..70b9a4b3fe629 100644 --- a/src/adapter/src/frontend_peek.rs +++ b/src/adapter/src/frontend_peek.rs @@ -12,11 +12,12 @@ use std::collections::BTreeSet; use std::sync::Arc; use std::time::Duration; +use futures::StreamExt; use itertools::Itertools; use mz_adapter_types::dyncfgs::ENABLE_FRONTEND_SUBSCRIBES; use mz_compute_types::ComputeInstanceId; use mz_compute_types::dataflows::DataflowDescription; -use mz_controller_types::ClusterId; +use mz_controller_types::{ClusterId, ReplicaId}; use mz_expr::{CollectionPlan, ResultSpec, RowSetFinishing}; use mz_ore::cast::{CastFrom, CastLossy}; use mz_ore::collections::CollectionExt; @@ -24,7 +25,7 @@ use mz_ore::now::EpochMillis; use mz_ore::task::JoinHandle; use mz_ore::{soft_assert_eq_or_log, soft_assert_or_log, soft_panic_or_log}; use mz_repr::optimize::{OptimizerFeatures, OverrideFrom}; -use mz_repr::{Datum, GlobalId, IntoRowIterator, Timestamp}; +use mz_repr::{Datum, GlobalId, IntoRowIterator, Row, RowIterator, Timestamp}; use mz_sql::ast::Raw; use mz_sql::catalog::CatalogCluster; use mz_sql::plan::Params; @@ -34,7 +35,10 @@ use mz_sql::plan::{ }; use mz_sql::rbac; use mz_sql::session::metadata::SessionMetadata; -use mz_sql::session::vars::IsolationLevel; +use mz_sql::session::user::MZ_SYSTEM_ROLE_ID; +use mz_sql::session::vars::{ + CLUSTER, CLUSTER_REPLICA, IsolationLevel, TRANSACTION_ISOLATION, Var, VarInput, +}; use mz_sql_parser::ast::{CopyDirection, ExplainStage, ShowStatement, Statement}; use mz_transform::EmptyStatisticsOracle; use mz_transform::dataflow::DataflowMetainfo; @@ -44,9 +48,9 @@ use tracing::{Span, debug, warn}; use tracing_opentelemetry::OpenTelemetrySpanExt; use crate::catalog::Catalog; -use crate::command::Command; +use crate::command::{CatalogSnapshot, Command}; use crate::coord; -use crate::coord::peek::{FastPathPlan, PeekPlan}; +use crate::coord::peek::{FastPathPlan, PeekPlan, PeekResponseUnary}; use crate::coord::sequencer::{eval_copy_to_uri, statistics_oracle}; use crate::coord::timeline::timedomain_for; use crate::coord::timestamp_selection::TimestampDetermination; @@ -203,6 +207,101 @@ impl PeekClient { .await } + /// Executes a coordinator-owned `SELECT`, pinned to `replica_id`, and + /// returns its rows. + /// + /// The caller must build `sql` itself rather than accept it from a user. + /// It runs as the system role, in a session of its own, and is not + /// statement-logged. This calls into the coordinator, so awaiting it on the + /// coordinator's main loop deadlocks. + /// + /// NOTE: Dropping the returned future does not cancel the peek, since + /// there is no connection to cancel it with. A peek dropped after its issue + /// keeps its read holds until the replica answers or goes away. One dropped + /// between its registration with the coordinator and its issue leaves the + /// registration behind, since only a response, a cancel of its connection, + /// or a drop of its inputs retires it. + pub(crate) async fn background_peek( + &mut self, + sql: &str, + cluster_id: ClusterId, + replica_id: ReplicaId, + ) -> Result, AdapterError> { + // Not `catalog_snapshot`, which expects the coordinator to outlive its + // caller. A background client may race the coordinator's shutdown. + let CatalogSnapshot { catalog } = self + .call_coordinator(|tx| Command::CatalogSnapshot { tx }) + .await?; + let (cluster_name, replica_name) = { + let cluster = catalog + .try_get_cluster(cluster_id) + .ok_or_else(|| AdapterError::Internal(format!("unknown cluster {cluster_id}")))?; + let replica = cluster + .replica(replica_id) + .ok_or_else(|| AdapterError::Internal(format!("unknown replica {replica_id}")))?; + (cluster.name.clone(), replica.name.clone()) + }; + let stmt = mz_sql::parse::parse(sql)?.into_element().ast; + + let mut session = Session::dummy(); + session.initialize_role_metadata(MZ_SYSTEM_ROLE_ID); + let system_vars = catalog.system_config(); + // Serializable reads at the freshest time the inputs can serve. A + // strict-serializable read picks the stalest valid time at or past the + // oracle's read timestamp instead. In a read-only environment whose + // leader is not running, nothing advances the oracle, and such a read + // sees introspection logs as of a time before the replicas started. + for (var, value) in [ + (&CLUSTER, cluster_name.as_str()), + (&CLUSTER_REPLICA, replica_name.as_str()), + (&TRANSACTION_ISOLATION, "serializable"), + ] { + session + .vars_mut() + .set(system_vars, var.name(), VarInput::Flat(value), false)?; + } + session.start_transaction_single_stmt(mz_ore::now::to_datetime((catalog.config().now)())); + + let mut logging = ExecutionLogging::adopt(None, self); + let response = self + .try_frontend_peek_inner( + &mut session, + catalog, + Some(Arc::new(stmt)), + Params::empty(), + &mut logging, + ) + .await?; + + let mut rows = Vec::new(); + let mut collect = |mut iter: Box| { + while let Some(row) = iter.next() { + rows.push(row.to_owned()); + } + }; + match response { + Some(ExecuteResponse::SendingRowsImmediate { rows }) => collect(rows), + Some(ExecuteResponse::SendingRowsStreaming { mut rows, .. }) => { + while let Some(response) = rows.next().await { + match response { + PeekResponseUnary::Rows(iter) => collect(iter), + PeekResponseUnary::Error(error) => return Err(error), + PeekResponseUnary::Canceled => return Err(AdapterError::Canceled), + PeekResponseUnary::DependencyDropped(dependency) => { + return Err(dependency.to_concurrent_dependency_drop()); + } + } + } + } + response => { + return Err(AdapterError::Internal(format!( + "background peek returned an unexpected response: {response:?}" + ))); + } + } + Ok(rows) + } + /// This is encapsulated in an inner function so that the outer function can still do statement /// logging after the `?` returns of the inner function. /// diff --git a/src/adapter/src/peek_client.rs b/src/adapter/src/peek_client.rs index b9b77940783ce..c6afc25393b6c 100644 --- a/src/adapter/src/peek_client.rs +++ b/src/adapter/src/peek_client.rs @@ -56,7 +56,7 @@ pub type StorageCollectionsHandle = Arc; /// Clients needed for peek sequencing in the Adapter Frontend. -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct PeekClient { coordinator_client: CoordinatorClient, /// Cache of the latest catalog snapshot. Serves diff --git a/src/compute-types/src/config.rs b/src/compute-types/src/config.rs index cde3aef4b685d..f7702741af2a7 100644 --- a/src/compute-types/src/config.rs +++ b/src/compute-types/src/config.rs @@ -43,6 +43,11 @@ pub struct ComputeReplicaLogging { /// The interval at which to log. /// /// A `None` value indicates that logging is disabled. + /// + /// NOTE: 0dt cutovers measure their stability period from the replica's + /// hydration log (see `mz_adapter::coord::caught_up`). With logging + /// disabled they still wait for hydration, but not for the period, so a + /// crash-looping replica can pass. pub interval: Option, } diff --git a/test/0dt/mzcompose.py b/test/0dt/mzcompose.py index 57245651f07de..4e47dee8fb9ef 100644 --- a/test/0dt/mzcompose.py +++ b/test/0dt/mzcompose.py @@ -3053,6 +3053,148 @@ def crash_loop() -> None: c.await_mz_deployment_status(DeploymentStatus.IS_LEADER, "mz_new", sleep_time=None) +def _pids(c: Composition, mz_service: str, command: str) -> set[str]: + """Returns the PIDs of processes named *command* in *mz_service*.""" + return set( + c.exec( + mz_service, + "bash", + "-c", + f"ps -eo pid=,comm= | awk '$2 == \"{command}\" {{ print $1 }}'", + capture=True, + silent=True, + ).stdout.split() + ) + + +def workflow_caught_up_stability_survives_restart(c: Composition) -> None: + """Verify a DDL-triggered restart of mz_new keeps its stability progress. + + mz_new's replicas outlive the restart, and the stability gate reads their + hydration times from the replicas, so the period keeps counting from the + original hydration. Restarting the period would make mz_new ready no + earlier than the restart plus a full period. + """ + period = 120 + ddl_after = 60 + + c.down(destroy_volumes=True) + c.up("mz_old") + + # Poll for DDL every second, so the restart comes right after the DDL and + # not at the end of the period. + c.sql( + f""" + ALTER SYSTEM SET with_0dt_caught_up_check_stability_period = '{period}s'; + ALTER SYSTEM SET with_0dt_deployment_ddl_check_interval = '1s'; + """, + service="mz_old", + port=6877, + user="mz_system", + ) + c.sql( + """ + CREATE CLUSTER stable SIZE 'scale=1,workers=1'; + CREATE TABLE t (a int); + CREATE MATERIALIZED VIEW mv IN CLUSTER stable AS SELECT * FROM t; + CREATE INDEX mv_idx IN CLUSTER stable ON mv (a); + INSERT INTO t VALUES (1), (2), (3); + """, + service="mz_old", + ) + + c.up("mz_new") + time.sleep(ddl_after) + environmentd = _pids(c, "mz_new", "environmentd") + replicas = _pids(c, "mz_new", "clusterd") + + # A table has no dataflow, so no replica gets a new export to hydrate. + c.sql("CREATE TABLE unrelated (a int)", service="mz_old") + + deadline = time.time() + 60 + while True: + pids = _pids(c, "mz_new", "environmentd") + if pids and pids != environmentd: + break + assert time.time() < deadline, "the DDL did not restart mz_new" + time.sleep(0.5) + restarted = time.time() + + c.await_mz_deployment_status(DeploymentStatus.READY_TO_PROMOTE, "mz_new") + ready_after = time.time() - restarted + + assert ( + _pids(c, "mz_new", "clusterd") == replicas + ), "mz_new's replicas did not survive its restart" + assert ( + ready_after < period + ), f"mz_new became ready {ready_after:.0f}s after its restart, so the restart reset the stability period" + + c.promote_mz("mz_new", retire_mz_service="mz_old") + c.await_mz_deployment_status(DeploymentStatus.IS_LEADER, "mz_new", sleep_time=None) + + +def workflow_caught_up_stability_without_dataflows(c: Composition) -> None: + """Verify the stability gate does not wait on hydration no replica reports. + + A materialized view past its last refresh has a sealed shard, so mz_new + never runs a dataflow for it. A materialized view pinned to one replica + runs only there. Neither may keep any replica of the cluster from counting + as hydrated. + """ + c.down(destroy_volumes=True) + c.up("mz_old") + + c.sql( + """ + ALTER SYSTEM SET with_0dt_caught_up_check_stability_period = '0s'; + ALTER SYSTEM SET enable_replica_targeted_materialized_views = true; + """, + service="mz_old", + port=6877, + user="mz_system", + ) + c.sql( + """ + CREATE CLUSTER pinned REPLICAS ( + r1 (SIZE 'scale=1,workers=1'), r2 (SIZE 'scale=1,workers=1') + ); + CREATE TABLE t (a int); + INSERT INTO t VALUES (1), (2), (3); + CREATE MATERIALIZED VIEW sealed IN CLUSTER pinned + WITH (REFRESH AT mz_now()::string::int8) AS SELECT * FROM t; + CREATE MATERIALIZED VIEW on_r1 IN CLUSTER pinned REPLICA r1 + AS SELECT count(*) FROM t; + """, + service="mz_old", + ) + + # Wait for the refresh to seal `sealed`, so mz_new starts with an empty + # as_of for it. + for _ in range(120): + sealed = c.sql_query( + """ + SELECT f.write_frontier IS NULL + FROM mz_internal.mz_frontiers f + JOIN mz_materialized_views mv ON f.object_id = mv.id + WHERE mv.name = 'sealed' + """, + service="mz_old", + ) + if sealed and sealed[0][0]: + break + time.sleep(1) + else: + raise RuntimeError("materialized view `sealed` never sealed its shard") + + c.up("mz_new") + c.await_mz_deployment_status( + DeploymentStatus.READY_TO_PROMOTE, "mz_new", timeout=300 + ) + c.promote_mz("mz_new", retire_mz_service="mz_old") + c.await_mz_deployment_status(DeploymentStatus.IS_LEADER, "mz_new", sleep_time=None) + + def workflow_ddl_detection_with_id_pool(c: Composition) -> None: """Verify that DDL detection works correctly with batch-allocated user IDs.