conditionally create mz_system and mz_probe cluster replicas (#31452)

See commit messages for details
<!--
Describe the contents of the PR briefly but completely.

If you write detailed commit messages, it is acceptable to copy/paste
them
here, or write "see commit messages for details." If there is only one
commit
in the PR, GitHub will have already added its commit message above.
-->

### Motivation
  * This PR adds a known-desirable feature.
MaterializeInc/database-issues#8954
<!--
Which of the following best describes the motivation behind this PR?

  * This PR fixes a recognized bug.

    [Ensure issue is linked somewhere.]


    [Ensure issue is linked somewhere.]

  * This PR fixes a previously unreported bug.

    [Describe the bug in detail, as if you were filing a bug report.]

  * This PR adds a feature that has not yet been specified.

[Write a brief specification for the feature, including justification
for its inclusion in Materialize, as if you were writing the original
     feature specification.]

   * This PR refactors existing code.

[Describe what was wrong with the existing code, if it is not obvious.]
-->

### Tips for reviewer

<!--
Leave some tips for your reviewer, like:

    * The diff is much smaller if viewed with whitespace hidden.
    * [Some function/module/file] deserves extra attention.
* [Some function/module/file] is pure code movement and only needs a
skim.

Delete this section if no tips.
-->

### Checklist

- [ ] This PR has adequate test coverage / QA involvement has been duly
considered. ([trigger-ci for additional test/nightly
runs](https://trigger-ci.dev.materialize.com/))
- [ ] This PR has an associated up-to-date [design
doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md),
is a design doc
([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)),
or is sufficiently small to not require a design.
  <!-- Reference the design in the description. -->
- [ ] If this PR evolves [an existing `$T ⇔ Proto$T`
mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md)
(possibly in a backwards-incompatible way), then it is tagged with a
`T-proto` label.
- [ ] If this PR will require changes to cloud orchestration or tests,
there is a companion cloud PR to account for those changes that is
tagged with the release-blocker label
([example](MaterializeInc/cloud#5021)).
<!-- Ask in #team-cloud on Slack if you need help preparing the cloud
PR. -->
- [ ] If this PR includes major [user-facing behavior
changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note),
I have pinged the relevant PM to schedule a changelog post.

Author: SangJunBak

Cargo.lock

@@ -125,6 +125,10 @@ The following table lists the configurable parameters of the Materialize operato
 | `operator.cloudProvider.providers.gcp` | GCP Configuration (placeholder for future use) | ``{"enabled":false}`` |
 | `operator.cloudProvider.region` | Common cloud provider settings | ``"kind"`` |
 | `operator.cloudProvider.type` | Specifies cloud provider. Valid values are 'aws', 'gcp', 'azure' , 'generic', or 'local' | ``"local"`` |
+| `operator.clusters.defaultReplicationFactor.analytics` |  | ``0`` |
+| `operator.clusters.defaultReplicationFactor.probe` |  | ``0`` |
+| `operator.clusters.defaultReplicationFactor.support` |  | ``0`` |
+| `operator.clusters.defaultReplicationFactor.system` |  | ``0`` |
 | `operator.clusters.defaultSizes.analytics` |  | ``"25cc"`` |
 | `operator.clusters.defaultSizes.catalogServer` |  | ``"50cc"`` |
 | `operator.clusters.defaultSizes.default` |  | ``"25cc"`` |

misc/helm-charts/operator/README.md

@@ -103,6 +103,18 @@ spec:
         {{ if .Values.operator.clusters.defaultSizes.analytics }}
         - "--bootstrap-builtin-analytics-cluster-replica-size={{ .Values.operator.clusters.defaultSizes.analytics }}"
         {{- end }}
+        {{ if ne .Values.operator.clusters.defaultReplicationFactor.system nil }}
+        - "--bootstrap-builtin-system-cluster-replication-factor={{ .Values.operator.clusters.defaultReplicationFactor.system }}"
+        {{- end }}
+        {{ if ne .Values.operator.clusters.defaultReplicationFactor.probe nil }}
+        - "--bootstrap-builtin-probe-cluster-replication-factor={{ .Values.operator.clusters.defaultReplicationFactor.probe }}"
+        {{- end }}
+        {{ if ne .Values.operator.clusters.defaultReplicationFactor.support nil }}
+        - "--bootstrap-builtin-support-cluster-replication-factor={{ .Values.operator.clusters.defaultReplicationFactor.support }}"
+        {{- end }}
+        {{ if ne .Values.operator.clusters.defaultReplicationFactor.analytics nil }}
+        - "--bootstrap-builtin-analytics-cluster-replication-factor={{ .Values.operator.clusters.defaultReplicationFactor.analytics }}"
+        {{- end }}
         {{- end }}
         - "--image-pull-policy={{ kebabcase .Values.operator.image.pullPolicy }}"
         {{- range $key, $value := .Values.environmentd.nodeSelector }}

misc/helm-charts/operator/templates/deployment.yaml

@@ -175,6 +175,11 @@ operator:
       support: 25cc
       catalogServer: 50cc
       analytics: 25cc
+    defaultReplicationFactor:
+      system: 0
+      probe: 0
+      support: 0
+      analytics: 0
 
   # Node selector to use for the operator pod
   nodeSelector: {}

misc/helm-charts/operator/values.yaml

@@ -0,0 +1,24 @@
+// 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.
+
+//! Types for bootstrap builtin cluster configuration.
+
+#[derive(Debug, Clone)]
+pub struct BootstrapBuiltinClusterConfig {
+    pub size: String,
+    pub replication_factor: u32,
+}
+
+pub const SYSTEM_CLUSTER_DEFAULT_REPLICATION_FACTOR: u32 = 1;
+pub const CATALOG_SERVER_CLUSTER_DEFAULT_REPLICATION_FACTOR: u32 = 1;
+pub const PROBE_CLUSTER_DEFAULT_REPLICATION_FACTOR: u32 = 1;
+// Support and analytics clusters are ephemeral - they are only spun up temporarily when needed.
+// Since they are short-lived, they don't need replication by default.
+pub const SUPPORT_CLUSTER_DEFAULT_REPLICATION_FACTOR: u32 = 0;
+pub const ANALYTICS_CLUSTER_DEFAULT_REPLICATION_FACTOR: u32 = 0;

src/adapter-types/src/bootstrap_builtin_cluster_config.rs

@@ -9,6 +9,7 @@
 
 //! Types for the adapter.
 
+pub mod bootstrap_builtin_cluster_config;
 pub mod compaction;
 pub mod connection;
 pub mod dyncfgs;

src/adapter-types/src/lib.rs

@@ -19,6 +19,11 @@ use std::sync::Arc;
 use futures::future::BoxFuture;
 use futures::{Future, FutureExt};
 use itertools::Itertools;
+use mz_adapter_types::bootstrap_builtin_cluster_config::{
+    BootstrapBuiltinClusterConfig, ANALYTICS_CLUSTER_DEFAULT_REPLICATION_FACTOR,
+    CATALOG_SERVER_CLUSTER_DEFAULT_REPLICATION_FACTOR, PROBE_CLUSTER_DEFAULT_REPLICATION_FACTOR,
+    SUPPORT_CLUSTER_DEFAULT_REPLICATION_FACTOR, SYSTEM_CLUSTER_DEFAULT_REPLICATION_FACTOR,
+};
 use mz_adapter_types::connection::ConnectionId;
 use mz_audit_log::{EventType, FullNameV1, ObjectType, VersionedStorageUsage};
 use mz_build_info::DUMMY_BUILD_INFO;
@@ -690,11 +695,26 @@ impl Catalog {
                 boot_ts: previous_ts,
                 skip_migrations: true,
                 cluster_replica_sizes: bootstrap_args.cluster_replica_size_map.clone(),
-                builtin_system_cluster_replica_size: replica_size.clone(),
-                builtin_catalog_server_cluster_replica_size: replica_size.clone(),
-                builtin_probe_cluster_replica_size: replica_size.clone(),
-                builtin_support_cluster_replica_size: replica_size.clone(),
-                builtin_analytics_cluster_replica_size: replica_size.clone(),
+                builtin_system_cluster_config: BootstrapBuiltinClusterConfig {
+                    size: replica_size.clone(),
+                    replication_factor: SYSTEM_CLUSTER_DEFAULT_REPLICATION_FACTOR,
+                },
+                builtin_catalog_server_cluster_config: BootstrapBuiltinClusterConfig {
+                    size: replica_size.clone(),
+                    replication_factor: CATALOG_SERVER_CLUSTER_DEFAULT_REPLICATION_FACTOR,
+                },
+                builtin_probe_cluster_config: BootstrapBuiltinClusterConfig {
+                    size: replica_size.clone(),
+                    replication_factor: PROBE_CLUSTER_DEFAULT_REPLICATION_FACTOR,
+                },
+                builtin_support_cluster_config: BootstrapBuiltinClusterConfig {
+                    size: replica_size.clone(),
+                    replication_factor: SUPPORT_CLUSTER_DEFAULT_REPLICATION_FACTOR,
+                },
+                builtin_analytics_cluster_config: BootstrapBuiltinClusterConfig {
+                    size: replica_size.clone(),
+                    replication_factor: ANALYTICS_CLUSTER_DEFAULT_REPLICATION_FACTOR,
+                },
                 system_parameter_defaults,
                 remote_system_parameters: None,
                 availability_zones: vec![],

src/adapter/src/catalog.rs

@@ -17,6 +17,7 @@ use std::time::{Duration, Instant};
 
 use futures::future::{BoxFuture, FutureExt};
 use itertools::{Either, Itertools};
+use mz_adapter_types::bootstrap_builtin_cluster_config::BootstrapBuiltinClusterConfig;
 use mz_adapter_types::dyncfgs::{ENABLE_CONTINUAL_TASK_BUILTINS, ENABLE_EXPRESSION_CACHE};
 use mz_catalog::builtin::{
     Builtin, Fingerprint, BUILTINS, BUILTIN_CLUSTERS, BUILTIN_CLUSTER_REPLICAS, BUILTIN_PREFIXES,
@@ -200,22 +201,22 @@ impl Catalog {
             // Add any new builtin objects and remove old ones.
             let (migrated_builtins, new_builtin_collections) =
                 add_new_remove_old_builtin_items_migration(&state.config().builtins_cfg, &mut txn)?;
-            let cluster_sizes = BuiltinBootstrapClusterSizes {
-                system_cluster: config.builtin_system_cluster_replica_size,
-                catalog_server_cluster: config.builtin_catalog_server_cluster_replica_size,
-                probe_cluster: config.builtin_probe_cluster_replica_size,
-                support_cluster: config.builtin_support_cluster_replica_size,
-                analytics_cluster: config.builtin_analytics_cluster_replica_size,
+            let builtin_bootstrap_cluster_config_map = BuiltinBootstrapClusterConfigMap {
+                system_cluster: config.builtin_system_cluster_config,
+                catalog_server_cluster: config.builtin_catalog_server_cluster_config,
+                probe_cluster: config.builtin_probe_cluster_config,
+                support_cluster: config.builtin_support_cluster_config,
+                analytics_cluster: config.builtin_analytics_cluster_config,
             };
             add_new_remove_old_builtin_clusters_migration(
                 &mut txn,
-                &cluster_sizes,
+                &builtin_bootstrap_cluster_config_map,
                 &state.cluster_replica_sizes,
             )?;
             add_new_remove_old_builtin_introspection_source_migration(&mut txn)?;
             add_new_remove_old_builtin_cluster_replicas_migration(
                 &mut txn,
-                &cluster_sizes,
+                &builtin_bootstrap_cluster_config_map,
                 &state.cluster_replica_sizes,
             )?;
             add_new_remove_old_builtin_roles_migration(&mut txn)?;
@@ -855,7 +856,7 @@ fn add_new_remove_old_builtin_items_migration(
 
 fn add_new_remove_old_builtin_clusters_migration(
     txn: &mut mz_catalog::durable::Transaction<'_>,
-    builtin_cluster_sizes: &BuiltinBootstrapClusterSizes,
+    builtin_cluster_config_map: &BuiltinBootstrapClusterConfigMap,
     cluster_sizes: &ClusterReplicaSizeMap,
 ) -> Result<(), mz_catalog::durable::CatalogError> {
     let mut durable_clusters: BTreeMap<_, _> = txn
@@ -867,18 +868,19 @@ fn add_new_remove_old_builtin_clusters_migration(
     // Add new clusters.
     for builtin_cluster in BUILTIN_CLUSTERS {
         if durable_clusters.remove(builtin_cluster.name).is_none() {
-            let cluster_size = builtin_cluster_sizes.get_size(builtin_cluster.name)?;
-            let cluster_allocation = cluster_sizes.get_allocation_by_name(&cluster_size)?;
+            let cluster_config = builtin_cluster_config_map.get_config(builtin_cluster.name)?;
+            let cluster_allocation = cluster_sizes.get_allocation_by_name(&cluster_config.size)?;
+
             txn.insert_system_cluster(
                 builtin_cluster.name,
                 vec![],
                 builtin_cluster.privileges.to_vec(),
                 builtin_cluster.owner_id.to_owned(),
                 mz_catalog::durable::ClusterConfig {
                     variant: mz_catalog::durable::ClusterVariant::Managed(ClusterVariantManaged {
-                        size: cluster_size,
+                        size: cluster_config.size,
                         availability_zones: vec![],
-                        replication_factor: builtin_cluster.replication_factor,
+                        replication_factor: cluster_config.replication_factor,
                         disk: cluster_allocation.is_cc,
                         logging: default_logging_config(),
                         optimizer_feature_overrides: Default::default(),
@@ -968,7 +970,7 @@ fn add_new_remove_old_builtin_roles_migration(
 
 fn add_new_remove_old_builtin_cluster_replicas_migration(
     txn: &mut Transaction<'_>,
-    builtin_cluster_sizes: &BuiltinBootstrapClusterSizes,
+    builtin_cluster_config_map: &BuiltinBootstrapClusterConfigMap,
     cluster_sizes: &ClusterReplicaSizeMap,
 ) -> Result<(), AdapterError> {
     let cluster_lookup: BTreeMap<_, _> = txn
@@ -996,12 +998,18 @@ fn add_new_remove_old_builtin_cluster_replicas_migration(
         let replica_names = durable_replicas
             .get_mut(&cluster.id)
             .unwrap_or(&mut empty_map);
-        if replica_names.remove(builtin_replica.name).is_none() {
+
+        let builtin_cluster_boostrap_config =
+            builtin_cluster_config_map.get_config(builtin_replica.cluster_name)?;
+        if replica_names.remove(builtin_replica.name).is_none()
+            // NOTE(SangJunBak): We need to explicitly check the replication factor because
+            // BUILT_IN_CLUSTER_REPLICAS is constant throughout all deployments but the replication
+            // factor is configurable on bootstrap.
+            && builtin_cluster_boostrap_config.replication_factor > 0
+        {
             let replica_size = match cluster.config.variant {
                 ClusterVariant::Managed(ClusterVariantManaged { ref size, .. }) => size.clone(),
-                ClusterVariant::Unmanaged => {
-                    builtin_cluster_sizes.get_size(builtin_replica.cluster_name)?
-                }
+                ClusterVariant::Unmanaged => builtin_cluster_boostrap_config.size,
             };
             let replica_allocation = cluster_sizes.get_allocation_by_name(&replica_size)?;
 
@@ -1115,37 +1123,43 @@ fn default_logging_config() -> ReplicaLogging {
         interval: Some(Duration::from_secs(1)),
     }
 }
-pub struct BuiltinBootstrapClusterSizes {
-    /// Size to default system_cluster on bootstrap
-    pub system_cluster: String,
-    /// Size to default catalog_server_cluster on bootstrap
-    pub catalog_server_cluster: String,
-    /// Size to default probe_cluster on bootstrap
-    pub probe_cluster: String,
-    /// Size to default support_cluster on bootstrap
-    pub support_cluster: String,
+
+#[derive(Debug)]
+pub struct BuiltinBootstrapClusterConfigMap {
+    /// Size and replication factor to default system_cluster on bootstrap
+    pub system_cluster: BootstrapBuiltinClusterConfig,
+    /// Size and replication factor to default catalog_server_cluster on bootstrap
+    pub catalog_server_cluster: BootstrapBuiltinClusterConfig,
+    /// Size and replication factor to default probe_cluster on bootstrap
+    pub probe_cluster: BootstrapBuiltinClusterConfig,
+    /// Size and replication factor to default support_cluster on bootstrap
+    pub support_cluster: BootstrapBuiltinClusterConfig,
     /// Size to default analytics_cluster on bootstrap
-    pub analytics_cluster: String,
+    pub analytics_cluster: BootstrapBuiltinClusterConfig,
 }
 
-impl BuiltinBootstrapClusterSizes {
+impl BuiltinBootstrapClusterConfigMap {
     /// Gets the size of the builtin cluster based on the provided name
-    fn get_size(&self, cluster_name: &str) -> Result<String, mz_catalog::durable::CatalogError> {
-        if cluster_name == mz_catalog::builtin::MZ_SYSTEM_CLUSTER.name {
-            Ok(self.system_cluster.clone())
+    fn get_config(
+        &self,
+        cluster_name: &str,
+    ) -> Result<BootstrapBuiltinClusterConfig, mz_catalog::durable::CatalogError> {
+        let cluster_config = if cluster_name == mz_catalog::builtin::MZ_SYSTEM_CLUSTER.name {
+            &self.system_cluster
         } else if cluster_name == mz_catalog::builtin::MZ_CATALOG_SERVER_CLUSTER.name {
-            Ok(self.catalog_server_cluster.clone())
+            &self.catalog_server_cluster
         } else if cluster_name == mz_catalog::builtin::MZ_PROBE_CLUSTER.name {
-            Ok(self.probe_cluster.clone())
+            &self.probe_cluster
         } else if cluster_name == mz_catalog::builtin::MZ_SUPPORT_CLUSTER.name {
-            Ok(self.support_cluster.clone())
+            &self.support_cluster
         } else if cluster_name == mz_catalog::builtin::MZ_ANALYTICS_CLUSTER.name {
-            Ok(self.analytics_cluster.clone())
+            &self.analytics_cluster
         } else {
-            Err(mz_catalog::durable::CatalogError::Catalog(
+            return Err(mz_catalog::durable::CatalogError::Catalog(
                 SqlCatalogError::UnexpectedBuiltinCluster(cluster_name.to_owned()),
-            ))
-        }
+            ));
+        };
+        Ok(cluster_config.clone())
     }
 }
 

src/adapter/src/catalog/open.rs

@@ -88,6 +88,7 @@ use futures::StreamExt;
 use http::Uri;
 use ipnet::IpNet;
 use itertools::{Either, Itertools};
+use mz_adapter_types::bootstrap_builtin_cluster_config::BootstrapBuiltinClusterConfig;
 use mz_adapter_types::compaction::CompactionWindow;
 use mz_adapter_types::connection::ConnectionId;
 use mz_adapter_types::dyncfgs::WITH_0DT_DEPLOYMENT_CAUGHT_UP_CHECK_INTERVAL;
@@ -999,11 +1000,11 @@ pub struct Config {
     pub cloud_resource_controller: Option<Arc<dyn CloudResourceController>>,
     pub availability_zones: Vec<String>,
     pub cluster_replica_sizes: ClusterReplicaSizeMap,
-    pub builtin_system_cluster_replica_size: String,
-    pub builtin_catalog_server_cluster_replica_size: String,
-    pub builtin_probe_cluster_replica_size: String,
-    pub builtin_support_cluster_replica_size: String,
-    pub builtin_analytics_cluster_replica_size: String,
+    pub builtin_system_cluster_config: BootstrapBuiltinClusterConfig,
+    pub builtin_catalog_server_cluster_config: BootstrapBuiltinClusterConfig,
+    pub builtin_probe_cluster_config: BootstrapBuiltinClusterConfig,
+    pub builtin_support_cluster_config: BootstrapBuiltinClusterConfig,
+    pub builtin_analytics_cluster_config: BootstrapBuiltinClusterConfig,
     pub system_parameter_defaults: BTreeMap<String, String>,
     pub storage_usage_client: StorageUsageClient,
     pub storage_usage_collection_interval: Duration,
@@ -3888,11 +3889,11 @@ pub fn serve(
         secrets_controller,
         cloud_resource_controller,
         cluster_replica_sizes,
-        builtin_system_cluster_replica_size,
-        builtin_catalog_server_cluster_replica_size,
-        builtin_probe_cluster_replica_size,
-        builtin_support_cluster_replica_size,
-        builtin_analytics_cluster_replica_size,
+        builtin_system_cluster_config,
+        builtin_catalog_server_cluster_config,
+        builtin_probe_cluster_config,
+        builtin_support_cluster_config,
+        builtin_analytics_cluster_config,
         system_parameter_defaults,
         availability_zones,
         storage_usage_client,
@@ -4041,11 +4042,11 @@ pub fn serve(
                 boot_ts: boot_ts.clone(),
                 skip_migrations: false,
                 cluster_replica_sizes,
-                builtin_system_cluster_replica_size,
-                builtin_catalog_server_cluster_replica_size,
-                builtin_probe_cluster_replica_size,
-                builtin_support_cluster_replica_size,
-                builtin_analytics_cluster_replica_size,
+                builtin_system_cluster_config,
+                builtin_catalog_server_cluster_config,
+                builtin_probe_cluster_config,
+                builtin_support_cluster_config,
+                builtin_analytics_cluster_config,
                 system_parameter_defaults,
                 remote_system_parameters,
                 availability_zones,

src/adapter/src/coord.rs

@@ -32,6 +32,7 @@ rust_binary(
     version = "0.133.0-dev.0",
     deps = [
         "//src/adapter:mz_adapter",
+        "//src/adapter-types:mz_adapter_types",
         "//src/build-info:mz_build_info",
         "//src/catalog:mz_catalog",
         "//src/cloud-resources:mz_cloud_resources",