mz_orchestratord/controller/materialize/

generation.rs

// 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 std::{
    collections::BTreeMap,
    net::{IpAddr, Ipv4Addr, SocketAddr},
    sync::LazyLock,
    time::Duration,
};

use k8s_openapi::{
    api::{
        apps::v1::{StatefulSet, StatefulSetSpec},
        core::v1::{
            Capabilities, ConfigMap, ConfigMapVolumeSource, Container, ContainerPort, EnvVar,
            EnvVarSource, KeyToPath, PodSecurityContext, PodSpec, PodTemplateSpec, Probe,
            SeccompProfile, Secret, SecretKeySelector, SecretVolumeSource, SecurityContext,
            Service, ServicePort, ServiceSpec, TCPSocketAction, Toleration, Volume, VolumeMount,
        },
    },
    apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString},
};
use kube::{Api, Client, ResourceExt, api::ObjectMeta, runtime::controller::Action};
use maplit::btreemap;
use mz_server_core::listeners::{
    AllowedRoles, AuthenticatorKind, BaseListenerConfig, RouteGroup, SqlListenerConfig,
    VersionedListenersConfig, v0_147_0, v26_32_0,
};
use reqwest::{Client as HttpClient, StatusCode};
use semver::{BuildMetadata, Prerelease, Version};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use tracing::{error, trace};

use super::Error;
use super::matching_image_from_environmentd_image_ref;
use crate::k8s::{apply_resource, delete_resource, get_resource};
use crate::tls::issuer_ref_defined;
use mz_cloud_provider::CloudProvider;
use mz_cloud_resources::crd::materialize::v1alpha1::Materialize;
use mz_cloud_resources::crd::{ManagedResource, recommended_k8s_labels};
use mz_ore::instrument;

static V140_DEV0: LazyLock<Version> = LazyLock::new(|| Version {
    major: 0,
    minor: 140,
    patch: 0,
    pre: Prerelease::new("dev.0").expect("dev.0 is valid prerelease"),
    build: BuildMetadata::new("").expect("empty string is valid buildmetadata"),
});
const V143: Version = Version::new(0, 143, 0);
const V144: Version = Version::new(0, 144, 0);
static V147_DEV0: LazyLock<Version> = LazyLock::new(|| Version {
    major: 0,
    minor: 147,
    patch: 0,
    pre: Prerelease::new("dev.0").expect("dev.0 is valid prerelease"),
    build: BuildMetadata::new("").expect("empty string is valid buildmetadata"),
});
const V153: Version = Version::new(0, 153, 0);
static V154_DEV0: LazyLock<Version> = LazyLock::new(|| Version {
    major: 0,
    minor: 154,
    patch: 0,
    pre: Prerelease::new("").expect("dev.0 is valid prerelease"),
    build: BuildMetadata::new("").expect("empty string is valid buildmetadata"),
});
pub const V161: Version = Version::new(0, 161, 0);

/// Version at which `environmentd` learned
/// `--orchestrator-kubernetes-priority-class-name`. Older images reject the
/// unknown argument and fail to start, so the flag is only forwarded at or
/// above this version.
static V26_42_0: LazyLock<Version> = LazyLock::new(|| Version {
    major: 26,
    minor: 42,
    patch: 0,
    pre: Prerelease::new("dev.0").expect("dev.0 is valid prerelease"),
    build: BuildMetadata::new("").expect("empty string is valid buildmetadata"),
});

static V26_1_0: LazyLock<Version> = LazyLock::new(|| Version {
    major: 26,
    minor: 1,
    patch: 0,
    pre: Prerelease::new("dev.0").expect("dev.0 is valid prerelease"),
    build: BuildMetadata::new("").expect("empty string is valid buildmetadata"),
});

/// Version at which HTTP listeners moved `allowed_roles` to be per route group
/// (the `v26_32_0::ListenersConfig` schema). Older `environmentd` parses the
/// legacy `v0_147_0::ListenersConfig` schema, so we serve that to them.
static PER_ROUTE_GROUP_ROLES_VERSION: LazyLock<Version> = LazyLock::new(|| Version {
    major: 26,
    minor: 32,
    patch: 0,
    pre: Prerelease::new("dev.0").expect("dev.0 is valid prerelease"),
    build: BuildMetadata::new("").expect("empty string is valid buildmetadata"),
});

/// Describes the status of a deployment.
///
/// This is a simplified representation of `DeploymentState`, suitable for
/// announcement to the external orchestrator.
#[derive(
    Debug,
    Serialize,
    Deserialize,
    Clone,
    Copy,
    PartialEq,
    Eq,
    PartialOrd,
    Ord
)]
pub enum DeploymentStatus {
    /// This deployment is not the leader. It is initializing and is not yet
    /// ready to become the leader.
    Initializing,
    /// This deployment is not the leader, but it is ready to become the leader.
    ReadyToPromote,
    /// This deployment is in the process of becoming the leader.
    Promoting,
    /// This deployment is the leader.
    IsLeader,
}

#[derive(
    Debug,
    Serialize,
    Deserialize,
    Clone,
    Copy,
    PartialEq,
    Eq,
    PartialOrd,
    Ord
)]
pub struct GetLeaderStatusResponse {
    status: DeploymentStatus,
}

#[derive(Deserialize, Serialize)]
pub struct LoginCredentials {
    username: String,
    // TODO Password type?
    password: String,
}

#[derive(Debug, Serialize)]
pub struct ConnectionInfo {
    pub environmentd_url: String,
    pub mz_system_secret_name: Option<String>,
    pub listeners_configmap: ConfigMap,
}

#[derive(Debug, Serialize)]
pub struct Resources {
    pub generation: u64,
    pub public_service: Box<Service>,
    pub generation_service: Box<Service>,
    pub persist_pubsub_service: Box<Service>,
    pub environmentd_statefulset: Box<StatefulSet>,
    pub connection_info: Box<ConnectionInfo>,
}

impl Resources {
    pub fn new(config: &super::Config, mz: &Materialize, generation: u64) -> Self {
        let public_service = Box::new(create_public_service_object(config, mz, generation));
        let generation_service = Box::new(create_generation_service_object(config, mz, generation));
        let persist_pubsub_service =
            Box::new(create_persist_pubsub_service(config, mz, generation));
        let environmentd_statefulset = Box::new(create_environmentd_statefulset_object(
            config, mz, generation,
        ));
        let connection_info = Box::new(create_connection_info(config, mz, generation));

        Self {
            generation,
            public_service,
            generation_service,
            persist_pubsub_service,
            environmentd_statefulset,
            connection_info,
        }
    }

    #[instrument]
    pub async fn apply(
        &self,
        client: &Client,
        force_promote: bool,
        namespace: &str,
    ) -> Result<Option<Action>, Error> {
        let service_api: Api<Service> = Api::namespaced(client.clone(), namespace);
        let statefulset_api: Api<StatefulSet> = Api::namespaced(client.clone(), namespace);
        let configmap_api: Api<ConfigMap> = Api::namespaced(client.clone(), namespace);

        trace!("applying environmentd per-generation service");
        apply_resource(&service_api, &*self.generation_service).await?;

        trace!("creating persist pubsub service");
        apply_resource(&service_api, &*self.persist_pubsub_service).await?;

        trace!("applying listeners configmap");
        apply_resource(&configmap_api, &self.connection_info.listeners_configmap).await?;

        trace!("creating new environmentd statefulset");
        apply_resource(&statefulset_api, &*self.environmentd_statefulset).await?;

        let retry_action = Action::requeue(Duration::from_secs(rand::random_range(5..10)));

        let statefulset = get_resource(
            &statefulset_api,
            &self.environmentd_statefulset.name_unchecked(),
        )
        .await?;
        if statefulset
            .and_then(|statefulset| statefulset.status)
            .and_then(|status| status.ready_replicas)
            .unwrap_or(0)
            == 0
        {
            trace!("environmentd statefulset is not ready yet...");
            return Ok(Some(retry_action));
        }

        let Some(http_client) = self.get_http_client(client.clone(), namespace).await else {
            return Ok(Some(retry_action));
        };
        let status_url = reqwest::Url::parse(&format!(
            "{}/api/leader/status",
            self.connection_info.environmentd_url,
        ))
        .unwrap();

        match http_client.get(status_url.clone()).send().await {
            Ok(response) => {
                let response: GetLeaderStatusResponse = match response.error_for_status() {
                    Ok(response) => response.json().await?,
                    Err(e) => {
                        trace!("failed to get status of environmentd, retrying... ({e})");
                        return Ok(Some(retry_action));
                    }
                };
                if force_promote {
                    trace!("skipping cluster catchup");
                    let skip_catchup_url = reqwest::Url::parse(&format!(
                        "{}/api/leader/skip-catchup",
                        self.connection_info.environmentd_url,
                    ))
                    .unwrap();
                    let response = http_client.post(skip_catchup_url).send().await?;
                    if response.status() == StatusCode::BAD_REQUEST {
                        let err: SkipCatchupError = response.json().await?;
                        return Err(
                            anyhow::anyhow!("failed to skip catchup: {}", err.message).into()
                        );
                    }
                } else {
                    match response.status {
                        DeploymentStatus::Initializing => {
                            trace!("environmentd is still initializing, retrying...");
                            return Ok(Some(retry_action));
                        }
                        DeploymentStatus::ReadyToPromote
                        | DeploymentStatus::Promoting
                        | DeploymentStatus::IsLeader => trace!("environmentd is ready"),
                    }
                }
            }
            Err(e) => {
                trace!("failed to connect to environmentd, retrying... ({e})");
                return Ok(Some(retry_action));
            }
        }

        Ok(None)
    }

    #[instrument]
    async fn get_http_client(&self, client: Client, namespace: &str) -> Option<HttpClient> {
        let secret_api: Api<Secret> = Api::namespaced(client.clone(), namespace);
        Some(match &self.connection_info.mz_system_secret_name {
            Some(mz_system_secret_name) => {
                let http_client = reqwest::Client::builder()
                    .timeout(std::time::Duration::from_secs(10))
                    .cookie_store(true)
                    // TODO add_root_certificate instead
                    .danger_accept_invalid_certs(true)
                    .build()
                    .unwrap();

                let secret = secret_api
                    .get(mz_system_secret_name)
                    .await
                    .map_err(|e| {
                        error!("Failed to get backend secret: {:?}", e);
                        e
                    })
                    .ok()?;
                if let Some(data) = secret.data {
                    if let Some(password) = data.get("external_login_password_mz_system").cloned() {
                        let password = String::from_utf8_lossy(&password.0).to_string();
                        let login_url = reqwest::Url::parse(&format!(
                            "{}/api/login",
                            self.connection_info.environmentd_url,
                        ))
                        .unwrap();
                        match http_client
                            .post(login_url)
                            .body(
                                serde_json::to_string(&LoginCredentials {
                                    username: "mz_system".to_owned(),
                                    password,
                                })
                                .expect(
                                    "Serializing a simple struct with utf8 strings doesn't fail.",
                                ),
                            )
                            .header("Content-Type", "application/json")
                            .send()
                            .await
                        {
                            Ok(response) => {
                                if let Err(e) = response.error_for_status() {
                                    trace!("failed to login to environmentd, retrying... ({e})");
                                    return None;
                                }
                            }
                            Err(e) => {
                                trace!("failed to connect to environmentd, retrying... ({e})");
                                return None;
                            }
                        };
                    }
                };
                http_client
            }
            None => reqwest::Client::builder()
                .timeout(std::time::Duration::from_secs(10))
                .build()
                .unwrap(),
        })
    }

    #[instrument]
    pub async fn promote_services(
        &self,
        client: &Client,
        namespace: &str,
    ) -> Result<Option<Action>, Error> {
        let service_api: Api<Service> = Api::namespaced(client.clone(), namespace);
        let retry_action = Action::requeue(Duration::from_secs(rand::random_range(5..10)));

        let promote_url = reqwest::Url::parse(&format!(
            "{}/api/leader/promote",
            self.connection_info.environmentd_url,
        ))
        .unwrap();

        let Some(http_client) = self.get_http_client(client.clone(), namespace).await else {
            return Ok(Some(retry_action));
        };

        trace!("promoting new environmentd to leader");
        let response = http_client.post(promote_url).send().await?;
        let response: BecomeLeaderResponse = response.error_for_status()?.json().await?;
        if let BecomeLeaderResult::Failure { message } = response.result {
            return Err(Error::Anyhow(anyhow::anyhow!(
                "failed to promote new environmentd: {message}"
            )));
        }

        // A successful POST to the promotion endpoint only indicates
        // that the promotion process was kicked off. It does not
        // guarantee that the environment will be successfully promoted
        // (e.g., if the environment crashes immediately after responding
        // to the request, but before executing the takeover, the
        // promotion will be lost).
        //
        // To guarantee the environment has been promoted successfully,
        // we must wait to see at least one `IsLeader` status returned
        // from the environment.

        let status_url = reqwest::Url::parse(&format!(
            "{}/api/leader/status",
            self.connection_info.environmentd_url,
        ))
        .unwrap();
        match http_client.get(status_url.clone()).send().await {
            Ok(response) => {
                let response: GetLeaderStatusResponse = response.json().await?;
                if response.status != DeploymentStatus::IsLeader {
                    trace!(
                        "environmentd is still promoting (status: {:?}), retrying...",
                        response.status
                    );
                    return Ok(Some(retry_action));
                } else {
                    trace!("environmentd is ready");
                }
            }
            Err(e) => {
                trace!("failed to connect to environmentd, retrying... ({e})");
                return Ok(Some(retry_action));
            }
        }

        trace!("applying environmentd public service");
        apply_resource(&service_api, &*self.public_service).await?;

        Ok(None)
    }

    #[instrument]
    pub async fn teardown_generation(
        &self,
        client: &Client,
        mz: &Materialize,
        generation: u64,
    ) -> Result<(), anyhow::Error> {
        let configmap_api: Api<ConfigMap> = Api::namespaced(client.clone(), &mz.namespace());
        let service_api: Api<Service> = Api::namespaced(client.clone(), &mz.namespace());
        let statefulset_api: Api<StatefulSet> = Api::namespaced(client.clone(), &mz.namespace());

        trace!("deleting environmentd statefulset for generation {generation}");
        delete_resource(
            &statefulset_api,
            &mz.environmentd_statefulset_name(generation),
        )
        .await?;

        trace!("deleting persist pubsub service for generation {generation}");
        delete_resource(&service_api, &mz.persist_pubsub_service_name(generation)).await?;

        trace!("deleting environmentd per-generation service for generation {generation}");
        delete_resource(
            &service_api,
            &mz.environmentd_generation_service_name(generation),
        )
        .await?;

        trace!("deleting listeners configmap for generation {generation}");
        delete_resource(&configmap_api, &mz.listeners_configmap_name(generation)).await?;

        Ok(())
    }

    // ideally we would just be able to hash the objects directly, but the
    // generated kubernetes objects don't implement the Hash trait
    pub fn generate_hash(&self) -> String {
        let mut hasher = Sha256::new();
        hasher.update(&serde_json::to_string(self).unwrap());
        hex::encode(hasher.finalize())
    }
}

fn create_public_service_object(
    config: &super::Config,
    mz: &Materialize,
    generation: u64,
) -> Service {
    let mut service = create_base_service_object(
        config,
        mz,
        generation,
        &mz.environmentd_service_name(),
        true,
    );
    // Only the stable, public Service is annotated for Teleport discovery,
    // not the per-generation Service: the agent must not see the name churn
    // on every rollout.
    super::teleport::apply_teleport_registration(config, mz, &mut service);
    service
}

fn create_generation_service_object(
    config: &super::Config,
    mz: &Materialize,
    generation: u64,
) -> Service {
    create_base_service_object(
        config,
        mz,
        generation,
        &mz.environmentd_generation_service_name(generation),
        false,
    )
}

fn create_base_service_object(
    config: &super::Config,
    mz: &Materialize,
    generation: u64,
    service_name: &str,
    headless: bool,
) -> Service {
    let ports = vec![
        ServicePort {
            port: config.environmentd_sql_port.into(),
            protocol: Some("TCP".to_string()),
            name: Some("sql".to_string()),
            ..Default::default()
        },
        ServicePort {
            port: config.environmentd_http_port.into(),
            protocol: Some("TCP".to_string()),
            name: Some("https".to_string()),
            ..Default::default()
        },
        ServicePort {
            port: config.environmentd_internal_sql_port.into(),
            protocol: Some("TCP".to_string()),
            name: Some("internal-sql".to_string()),
            ..Default::default()
        },
        ServicePort {
            port: config.environmentd_internal_http_port.into(),
            protocol: Some("TCP".to_string()),
            name: Some("internal-http".to_string()),
            ..Default::default()
        },
    ];

    let selector = btreemap! {"materialize.cloud/name".to_string() => mz.environmentd_statefulset_name(generation)};

    let spec = ServiceSpec {
        type_: Some("ClusterIP".to_string()),
        cluster_ip: if headless {
            Some("None".to_string())
        } else {
            None
        },
        selector: Some(selector),
        ports: Some(ports),
        ..Default::default()
    };

    Service {
        metadata: mz.managed_resource_meta(service_name.to_string()),
        spec: Some(spec),
        status: None,
    }
}

fn create_persist_pubsub_service(
    config: &super::Config,
    mz: &Materialize,
    generation: u64,
) -> Service {
    Service {
        metadata: mz.managed_resource_meta(mz.persist_pubsub_service_name(generation)),
        spec: Some(ServiceSpec {
            type_: Some("ClusterIP".to_string()),
            cluster_ip: Some("None".to_string()),
            selector: Some(btreemap! {
                "materialize.cloud/name".to_string() => mz.environmentd_statefulset_name(generation),
            }),
            ports: Some(vec![ServicePort {
                name: Some("grpc".to_string()),
                protocol: Some("TCP".to_string()),
                port: config.environmentd_internal_persist_pubsub_port.into(),
                ..Default::default()
            }]),
            ..Default::default()
        }),
        status: None,
    }
}

fn create_environmentd_statefulset_object(
    config: &super::Config,
    mz: &Materialize,
    generation: u64,
) -> StatefulSet {
    // IMPORTANT: Only pass secrets via environment variables. All other
    // parameters should be passed as command line arguments, possibly gated
    // with a `meets_minimum_version` call. This ensures typos cause
    // `environmentd` to fail to start, rather than silently ignoring the
    // configuration.
    //
    // When passing a secret, use a `SecretKeySelector` to forward a secret into
    // the pod. Do *not* hardcode a secret as the value directly, as doing so
    // will leak the secret to anyone with permission to describe the pod.
    let mut env = vec![
        EnvVar {
            name: "MZ_METADATA_BACKEND_URL".to_string(),
            value_from: Some(EnvVarSource {
                secret_key_ref: Some(SecretKeySelector {
                    name: mz.backend_secret_name(),
                    key: "metadata_backend_url".to_string(),
                    optional: Some(false),
                }),
                ..Default::default()
            }),
            ..Default::default()
        },
        EnvVar {
            name: "MZ_PERSIST_BLOB_URL".to_string(),
            value_from: Some(EnvVarSource {
                secret_key_ref: Some(SecretKeySelector {
                    name: mz.backend_secret_name(),
                    key: "persist_backend_url".to_string(),
                    optional: Some(false),
                }),
                ..Default::default()
            }),
            ..Default::default()
        },
    ];

    env.push(EnvVar {
        name: "AWS_REGION".to_string(),
        value: Some(config.region.clone()),
        ..Default::default()
    });

    env.extend(mz.spec.environmentd_extra_env.iter().flatten().cloned());

    let mut args = vec![];

    if let Some(helm_chart_version) = &config.helm_chart_version {
        args.push(format!("--helm-chart-version={helm_chart_version}"));
    }

    // Add environment ID argument.
    args.push(format!(
        "--environment-id={}",
        mz.environment_id(&config.cloud_provider.to_string(), &config.region)
    ));

    // Add clusterd image argument based on environmentd tag.
    args.push(format!(
        "--clusterd-image={}",
        matching_image_from_environmentd_image_ref(
            &mz.spec.environmentd_image_ref,
            "clusterd",
            None
        )
    ));

    // Add cluster and storage host size arguments.
    args.extend(
        [
            config
                .environmentd_cluster_replica_sizes
                .as_ref()
                .map(|sizes| format!("--cluster-replica-sizes={sizes}")),
            config
                .bootstrap_default_cluster_replica_size
                .as_ref()
                .map(|size| format!("--bootstrap-default-cluster-replica-size={size}")),
            config
                .bootstrap_builtin_system_cluster_replica_size
                .as_ref()
                .map(|size| format!("--bootstrap-builtin-system-cluster-replica-size={size}")),
            config
                .bootstrap_builtin_probe_cluster_replica_size
                .as_ref()
                .map(|size| format!("--bootstrap-builtin-probe-cluster-replica-size={size}")),
            config
                .bootstrap_builtin_support_cluster_replica_size
                .as_ref()
                .map(|size| format!("--bootstrap-builtin-support-cluster-replica-size={size}")),
            config
                .bootstrap_builtin_catalog_server_cluster_replica_size
                .as_ref()
                .map(|size| format!("--bootstrap-builtin-catalog-server-cluster-replica-size={size}")),
            config
                .bootstrap_builtin_analytics_cluster_replica_size
                .as_ref()
                .map(|size| format!("--bootstrap-builtin-analytics-cluster-replica-size={size}")),
            config
                .bootstrap_builtin_system_cluster_replication_factor
                .as_ref()
                .map(|replication_factor| {
                    format!("--bootstrap-builtin-system-cluster-replication-factor={replication_factor}")
                }),
            config
                .bootstrap_builtin_probe_cluster_replication_factor
                .as_ref()
                .map(|replication_factor| format!("--bootstrap-builtin-probe-cluster-replication-factor={replication_factor}")),
            config
                .bootstrap_builtin_support_cluster_replication_factor
                .as_ref()
                .map(|replication_factor| format!("--bootstrap-builtin-support-cluster-replication-factor={replication_factor}")),
            config
                .bootstrap_builtin_analytics_cluster_replication_factor
                .as_ref()
                .map(|replication_factor| format!("--bootstrap-builtin-analytics-cluster-replication-factor={replication_factor}")),
        ]
        .into_iter()
        .flatten(),
    );

    args.extend(
        config
            .environmentd_allowed_origins
            .iter()
            .map(|origin| format!("--cors-allowed-origin={}", origin.to_str().unwrap())),
    );

    args.push(format!(
        "--secrets-controller={}",
        config.secrets_controller
    ));

    if let Some(cluster_replica_sizes) = &config.environmentd_cluster_replica_sizes {
        if let Ok(cluster_replica_sizes) =
            serde_json::from_str::<BTreeMap<String, serde_json::Value>>(cluster_replica_sizes)
        {
            let cluster_replica_sizes: Vec<_> =
                cluster_replica_sizes.keys().map(|s| s.as_str()).collect();
            args.push(format!(
                "--system-parameter-default=allowed_cluster_replica_sizes='{}'",
                cluster_replica_sizes.join("', '")
            ));
        }
    }
    if !config.cloud_provider.is_cloud() {
        args.push("--system-parameter-default=cluster_enable_topology_spread=false".into());
    }

    if config.enable_internal_statement_logging {
        args.push("--system-parameter-default=enable_internal_statement_logging=true".into());
    }

    if let Some(rate) = config.statement_logging_max_sample_rate {
        args.push(format!(
            "--system-parameter-default=statement_logging_max_sample_rate={rate}"
        ));
    }

    if let Some(rate) = config.statement_logging_target_data_rate {
        args.push(format!(
            "--system-parameter-default=statement_logging_target_data_rate={rate}"
        ));
    }

    if !mz.spec.enable_rbac {
        args.push("--system-parameter-default=enable_rbac_checks=false".into());
    }

    if config.disable_database_network_policies {
        args.push("--system-parameter-default=enable_network_policies=false".into());
    }

    // Add persist arguments.

    // Configure the Persist Isolated Runtime to use one less thread than the total available.
    args.push("--persist-isolated-runtime-threads=-1".to_string());

    // Add AWS arguments.
    if config.cloud_provider == CloudProvider::Aws {
        if let Some(azs) = config.environmentd_availability_zones.as_ref() {
            for az in azs {
                args.push(format!("--availability-zone={az}"));
            }
        }

        if let Some(environmentd_connection_role_arn) = mz
            .spec
            .environmentd_connection_role_arn
            .as_deref()
            .or(config.environmentd_connection_role_arn.as_deref())
        {
            args.push(format!(
                "--aws-connection-role-arn={}",
                environmentd_connection_role_arn
            ));
        }
        if let Some(account_id) = &config.aws_account_id {
            args.push(format!("--aws-account-id={account_id}"));
        }

        args.extend([format!(
            "--aws-secrets-controller-tags=Environment={}",
            mz.name_unchecked()
        )]);
        args.extend_from_slice(&config.aws_secrets_controller_tags);
    }

    // Add Kubernetes arguments.
    args.extend([
        "--orchestrator=kubernetes".into(),
        format!(
            "--orchestrator-kubernetes-service-account={}",
            mz.service_account_name()
        ),
        format!(
            "--orchestrator-kubernetes-image-pull-policy={}",
            config.image_pull_policy.as_kebab_case_str(),
        ),
    ]);
    for selector in &config.clusterd_node_selector {
        args.push(format!(
            "--orchestrator-kubernetes-service-node-selector={}={}",
            selector.key, selector.value,
        ));
    }
    if mz.meets_minimum_version(&V144) {
        if let Some(affinity) = &config.clusterd_affinity {
            let affinity = serde_json::to_string(affinity).unwrap();
            args.push(format!(
                "--orchestrator-kubernetes-service-affinity={affinity}"
            ))
        }
        if let Some(tolerations) = &config.clusterd_tolerations {
            let tolerations = serde_json::to_string(tolerations).unwrap();
            args.push(format!(
                "--orchestrator-kubernetes-service-tolerations={tolerations}"
            ))
        }
    }
    if let Some(scheduler_name) = &config.scheduler_name {
        args.push(format!(
            "--orchestrator-kubernetes-scheduler-name={}",
            scheduler_name
        ));
    }
    if mz.meets_minimum_version(&V26_42_0) {
        if let Some(priority_class_name) = &config.clusterd_priority_class_name {
            args.push(format!(
                "--orchestrator-kubernetes-priority-class-name={priority_class_name}"
            ));
        }
    }
    if mz.meets_minimum_version(&V154_DEV0) {
        args.extend(
            mz.spec
                .pod_annotations
                .as_ref()
                .map(|annotations| annotations.iter())
                .unwrap_or_default()
                .map(|(key, val)| {
                    format!("--orchestrator-kubernetes-service-annotation={key}={val}")
                }),
        );
    }
    args.extend(
        mz.default_labels()
            .iter()
            .chain(
                mz.spec
                    .pod_labels
                    .as_ref()
                    .map(|labels| labels.iter())
                    .unwrap_or_default(),
            )
            .map(|(key, val)| format!("--orchestrator-kubernetes-service-label={key}={val}")),
    );
    if let Some(status) = &mz.status {
        args.push(format!(
            "--orchestrator-kubernetes-name-prefix=mz{}-",
            status.resource_id
        ));
    }

    // Add logging and tracing arguments.
    args.extend(["--log-format=json".into()]);
    if let Some(endpoint) = &config.tracing.opentelemetry_endpoint {
        args.push(format!("--opentelemetry-endpoint={}", endpoint));
    }
    // --opentelemetry-resource also configures sentry tags
    args.extend([
        format!(
            "--opentelemetry-resource=organization_id={}",
            mz.spec.environment_id
        ),
        format!(
            "--opentelemetry-resource=environment_name={}",
            mz.name_unchecked()
        ),
    ]);

    if let Some(segment_api_key) = &config.segment_api_key {
        args.push(format!("--segment-api-key={}", segment_api_key));
        if config.segment_client_side {
            args.push("--segment-client-side".into());
        }
    }

    let mut volumes = Vec::new();
    let mut volume_mounts = Vec::new();
    if issuer_ref_defined(
        &config.default_certificate_specs.internal,
        &mz.spec.internal_certificate_spec,
    ) {
        volumes.push(Volume {
            name: "certificate".to_owned(),
            secret: Some(SecretVolumeSource {
                default_mode: Some(0o400),
                secret_name: Some(mz.environmentd_certificate_secret_name()),
                items: None,
                optional: Some(false),
            }),
            ..Default::default()
        });
        volume_mounts.push(VolumeMount {
            name: "certificate".to_owned(),
            mount_path: "/etc/materialized".to_owned(),
            read_only: Some(true),
            ..Default::default()
        });
        args.extend([
            "--tls-mode=require".into(),
            "--tls-cert=/etc/materialized/tls.crt".into(),
            "--tls-key=/etc/materialized/tls.key".into(),
        ]);
    } else {
        args.push("--tls-mode=disable".to_string());
    }
    if let Some(ephemeral_volume_class) = &config.ephemeral_volume_class {
        args.push(format!(
            "--orchestrator-kubernetes-ephemeral-volume-class={}",
            ephemeral_volume_class
        ));
    }
    // The `materialize` user used by clusterd always has gid 999.
    args.push("--orchestrator-kubernetes-service-fs-group=999".to_string());

    // Add system_param configmap
    // This feature was enabled in 0.163 but did not have testing until after 0.164.
    // 0.165 should work with anything greater than 0.164 including v26 and v25.
    if mz.meets_minimum_version(&V26_1_0) {
        if let Some(ref name) = mz.spec.system_parameter_configmap_name {
            volumes.push(Volume {
                name: "system-params".to_string(),
                config_map: Some(ConfigMapVolumeSource {
                    default_mode: Some(0o400),
                    name: name.to_owned(),
                    items: None,
                    optional: Some(true),
                }),
                ..Default::default()
            });
            volume_mounts.push(VolumeMount {
                name: "system-params".to_string(),
                // The user must write to the `system-params.json` entry in the config map
                mount_path: "/system-params".to_owned(),
                read_only: Some(true),
                ..Default::default()
            });
            args.push("--config-sync-file-path=/system-params/system-params.json".to_string());
            args.push("--config-sync-loop-interval=1s".to_string());
        }
    }

    // Add Sentry arguments.
    if let Some(sentry_dsn) = &config.tracing.sentry_dsn {
        args.push(format!("--sentry-dsn={}", sentry_dsn));
        if let Some(sentry_environment) = &config.tracing.sentry_environment {
            args.push(format!("--sentry-environment={}", sentry_environment));
        }
        args.push(format!("--sentry-tag=region={}", config.region));
    }

    // Add Persist PubSub arguments
    args.push(format!(
        "--persist-pubsub-url=http://{}:{}",
        mz.persist_pubsub_service_name(generation),
        config.environmentd_internal_persist_pubsub_port,
    ));
    args.push(format!(
        "--internal-persist-pubsub-listen-addr=0.0.0.0:{}",
        config.environmentd_internal_persist_pubsub_port
    ));

    args.push(format!("--deploy-generation={}", generation));

    // Add URL for internal user impersonation endpoint
    args.push(format!(
        "--internal-console-redirect-url={}",
        config.internal_console_proxy_url,
    ));

    if !config.collect_pod_metrics {
        args.push("--orchestrator-kubernetes-disable-pod-metrics-collection".into());
    }
    if config.enable_prometheus_scrape_annotations {
        args.push("--orchestrator-kubernetes-enable-prometheus-scrape-annotations".into());
    }

    // the --disable-license-key-checks environmentd flag only existed
    // between these versions
    if config.disable_license_key_checks {
        if mz.meets_minimum_version(&V143) && !mz.meets_minimum_version(&V153) {
            args.push("--disable-license-key-checks".into());
        }
    }

    // as of version 0.153, the ability to disable license key checks was
    // removed, so we should always set up license keys in that case
    if (mz.meets_minimum_version(&V140_DEV0) && !config.disable_license_key_checks)
        || mz.meets_minimum_version(&V153)
    {
        volume_mounts.push(VolumeMount {
            name: "license-key".to_string(),
            mount_path: "/license_key".to_string(),
            ..Default::default()
        });
        volumes.push(Volume {
            name: "license-key".to_string(),
            secret: Some(SecretVolumeSource {
                default_mode: Some(256),
                optional: Some(false),
                secret_name: Some(mz.backend_secret_name()),
                items: Some(vec![KeyToPath {
                    key: "license_key".to_string(),
                    path: "license_key".to_string(),
                    ..Default::default()
                }]),
                ..Default::default()
            }),
            ..Default::default()
        });
        env.push(EnvVar {
            name: "MZ_LICENSE_KEY".to_string(),
            value: Some("/license_key/license_key".to_string()),
            ..Default::default()
        });
    }

    // Add user-specified extra arguments.
    if let Some(extra_args) = &mz.spec.environmentd_extra_args {
        args.extend(extra_args.iter().cloned());
    }

    // NOTE: environmentd binds its SQL listeners before booting but only
    // accepts once boot completes, so connections queue in the kernel until
    // then. balancerd and clients use the external SQL port and can fill its
    // queue, after which TCP probes against it time out and liveness kills the
    // pod mid-boot. Probing the internal port keeps client traffic out of the
    // probes' queue. Some authenticator kinds have no internal SQL listener,
    // and probing an unbound port fails every check, so fall back to the
    // external one.
    let listeners_config = create_v0_147_0_listeners_config(config, mz);
    let probe_port = listeners_config
        .sql
        .get("internal")
        .or_else(|| listeners_config.sql.get("external"))
        .expect("external SQL listener is always configured")
        .addr
        .port();
    let base_probe = Probe {
        initial_delay_seconds: Some(1),
        failure_threshold: Some(12),
        tcp_socket: Some(TCPSocketAction {
            host: None,
            port: IntOrString::Int(probe_port.into()),
        }),
        ..Default::default()
    };
    let readiness_probe = environmentd_probe(
        base_probe.clone(),
        config.environmentd_default_readiness_probe.as_ref(),
        mz.spec.environmentd_readiness_probe.as_ref(),
    );
    let liveness_probe = environmentd_probe(
        base_probe.clone(),
        config.environmentd_default_liveness_probe.as_ref(),
        mz.spec.environmentd_liveness_probe.as_ref(),
    );
    // Unlike readiness and liveness, environmentd has no startup probe unless
    // one is configured.
    let startup_probe = (config.environmentd_default_startup_probe.is_some()
        || mz.spec.environmentd_startup_probe.is_some())
    .then(|| {
        environmentd_probe(
            base_probe,
            config.environmentd_default_startup_probe.as_ref(),
            mz.spec.environmentd_startup_probe.as_ref(),
        )
    });

    let security_context = if config.enable_security_context {
        // Since we want to adhere to the most restrictive security context, all
        // of these fields have to be set how they are.
        // See https://kubernetes.io/docs/concepts/security/pod-security-standards/#restricted
        Some(SecurityContext {
            run_as_non_root: Some(true),
            capabilities: Some(Capabilities {
                drop: Some(vec!["ALL".to_string()]),
                ..Default::default()
            }),
            seccomp_profile: Some(SeccompProfile {
                type_: "RuntimeDefault".to_string(),
                ..Default::default()
            }),
            allow_privilege_escalation: Some(false),
            ..Default::default()
        })
    } else {
        None
    };

    let ports = vec![
        ContainerPort {
            container_port: config.environmentd_sql_port.into(),
            name: Some("sql".to_owned()),
            ..Default::default()
        },
        ContainerPort {
            container_port: config.environmentd_internal_sql_port.into(),
            name: Some("internal-sql".to_owned()),
            ..Default::default()
        },
        ContainerPort {
            container_port: config.environmentd_http_port.into(),
            name: Some("http".to_owned()),
            ..Default::default()
        },
        ContainerPort {
            container_port: config.environmentd_internal_http_port.into(),
            name: Some("internal-http".to_owned()),
            ..Default::default()
        },
        ContainerPort {
            container_port: config.environmentd_internal_persist_pubsub_port.into(),
            name: Some("persist-pubsub".to_owned()),
            ..Default::default()
        },
    ];

    // Add networking arguments.
    if mz.meets_minimum_version(&V147_DEV0) {
        volume_mounts.push(VolumeMount {
            name: "listeners-configmap".to_string(),
            mount_path: "/listeners".to_string(),
            ..Default::default()
        });
        volumes.push(Volume {
            name: "listeners-configmap".to_string(),
            config_map: Some(ConfigMapVolumeSource {
                name: mz.listeners_configmap_name(generation),
                default_mode: Some(256),
                optional: Some(false),
                items: Some(vec![KeyToPath {
                    key: "listeners.json".to_string(),
                    path: "listeners.json".to_string(),
                    ..Default::default()
                }]),
            }),
            ..Default::default()
        });
        args.push("--listeners-config-path=/listeners/listeners.json".to_owned());
        if matches!(
            mz.spec.authenticator_kind,
            AuthenticatorKind::Password | AuthenticatorKind::Sasl | AuthenticatorKind::Oidc
        ) {
            args.push("--system-parameter-default=enable_password_auth=true".into());
            env.push(EnvVar {
                name: "MZ_EXTERNAL_LOGIN_PASSWORD_MZ_SYSTEM".to_string(),
                value_from: Some(EnvVarSource {
                    secret_key_ref: Some(SecretKeySelector {
                        name: mz.backend_secret_name(),
                        key: "external_login_password_mz_system".to_string(),
                        optional: Some(false),
                    }),
                    ..Default::default()
                }),
                ..Default::default()
            })
        }
    } else {
        args.extend([
            format!("--sql-listen-addr=0.0.0.0:{}", config.environmentd_sql_port),
            format!(
                "--http-listen-addr=0.0.0.0:{}",
                config.environmentd_http_port
            ),
            format!(
                "--internal-sql-listen-addr=0.0.0.0:{}",
                config.environmentd_internal_sql_port
            ),
            format!(
                "--internal-http-listen-addr=0.0.0.0:{}",
                config.environmentd_internal_http_port
            ),
        ]);
    }

    let container = Container {
        name: "environmentd".to_owned(),
        image: Some(mz.spec.environmentd_image_ref.to_owned()),
        image_pull_policy: Some(config.image_pull_policy.to_string()),
        ports: Some(ports),
        args: Some(args),
        env: Some(env),
        volume_mounts: Some(volume_mounts),
        startup_probe,
        liveness_probe: Some(liveness_probe),
        readiness_probe: Some(readiness_probe),
        resources: mz
            .spec
            .environmentd_resource_requirements
            .clone()
            .or_else(|| config.environmentd_default_resources.clone()),
        security_context: security_context.clone(),
        ..Default::default()
    };

    let mut pod_template_labels = mz.default_labels();
    pod_template_labels.insert(
        "materialize.cloud/name".to_owned(),
        mz.environmentd_statefulset_name(generation),
    );
    pod_template_labels.insert(
        "materialize.cloud/app".to_owned(),
        mz.environmentd_app_name(),
    );
    pod_template_labels.extend(recommended_k8s_labels("environmentd".into()));
    pod_template_labels.extend(
        mz.spec
            .pod_labels
            .as_ref()
            .map(|labels| labels.iter())
            .unwrap_or_default()
            .map(|(key, value)| (key.clone(), value.clone())),
    );

    let mut pod_template_annotations = btreemap! {
        // We can re-enable eviction once we have HA
        "cluster-autoscaler.kubernetes.io/safe-to-evict".to_owned() => "false".to_string(),

        // Prevents old (< 0.30) and new versions of karpenter from evicting database pods
        "karpenter.sh/do-not-evict".to_owned() => "true".to_string(),
        "karpenter.sh/do-not-disrupt".to_owned() => "true".to_string(),
        "materialize.cloud/generation".to_owned() => generation.to_string(),
    };
    if config.enable_prometheus_scrape_annotations {
        pod_template_annotations.insert("prometheus.io/scrape".to_owned(), "true".to_string());
        pod_template_annotations.insert(
            "prometheus.io/port".to_owned(),
            config.environmentd_internal_http_port.to_string(),
        );
        pod_template_annotations.insert("prometheus.io/path".to_owned(), "/metrics".to_string());
        pod_template_annotations.insert("prometheus.io/scheme".to_owned(), "http".to_string());
        pod_template_annotations.insert(
            "materialize.prometheus.io/mz_usage_path".to_owned(),
            "/metrics/mz_usage".to_string(),
        );
        pod_template_annotations.insert(
            "materialize.prometheus.io/mz_frontier_path".to_owned(),
            "/metrics/mz_frontier".to_string(),
        );
        pod_template_annotations.insert(
            "materialize.prometheus.io/mz_compute_path".to_owned(),
            "/metrics/mz_compute".to_string(),
        );
        pod_template_annotations.insert(
            "materialize.prometheus.io/mz_storage_path".to_owned(),
            "/metrics/mz_storage".to_string(),
        );
    }
    pod_template_annotations.extend(
        mz.spec
            .pod_annotations
            .as_ref()
            .map(|annotations| annotations.iter())
            .unwrap_or_default()
            .map(|(key, value)| (key.clone(), value.clone())),
    );

    let mut tolerations = vec![
        // When the node becomes `NotReady` it indicates there is a problem with the node,
        // By default kubernetes waits 300s (5 minutes) before doing anything in this case,
        // But we want to limit this to 30s for faster recovery
        Toleration {
            effect: Some("NoExecute".into()),
            key: Some("node.kubernetes.io/not-ready".into()),
            operator: Some("Exists".into()),
            toleration_seconds: Some(30),
            value: None,
        },
        Toleration {
            effect: Some("NoExecute".into()),
            key: Some("node.kubernetes.io/unreachable".into()),
            operator: Some("Exists".into()),
            toleration_seconds: Some(30),
            value: None,
        },
    ];
    if let Some(user_tolerations) = &config.environmentd_tolerations {
        tolerations.extend(user_tolerations.iter().cloned());
    }
    let tolerations = Some(tolerations);

    let pod_template_spec = PodTemplateSpec {
        // not using managed_resource_meta because the pod should be owned
        // by the statefulset, not the materialize instance
        metadata: Some(ObjectMeta {
            labels: Some(pod_template_labels),
            annotations: Some(pod_template_annotations), // This is inserted into later, do not delete.
            ..Default::default()
        }),
        spec: Some(PodSpec {
            containers: vec![container],
            node_selector: Some(
                config
                    .environmentd_node_selector
                    .iter()
                    .map(|selector| (selector.key.clone(), selector.value.clone()))
                    .collect(),
            ),
            affinity: config.environmentd_affinity.clone(),
            scheduler_name: config.scheduler_name.clone(),
            priority_class_name: config.environmentd_priority_class_name.clone(),
            service_account_name: Some(mz.service_account_name()),
            volumes: Some(volumes),
            security_context: Some(PodSecurityContext {
                fs_group: Some(999),
                run_as_user: Some(999),
                run_as_group: Some(999),
                ..Default::default()
            }),
            tolerations,
            // This (apparently) has the side effect of automatically starting a new pod
            // when the previous pod is currently terminating. This side steps the statefulset fencing
            // but allows for quicker recovery from a failed node
            // The Kubernetes documentation strongly advises against this
            // setting, as StatefulSets attempt to provide "at most once"
            // semantics [0]-- that is, the guarantee that for a given pod in a
            // StatefulSet there is *at most* one pod with that identity running
            // in the cluster
            //
            // Materialize, however, has been carefully designed to *not* rely
            // on this guarantee. (In fact, we do not believe that correct
            // distributed systems can meaningfully rely on Kubernetes's
            // guarantee--network packets from a pod can be arbitrarily delayed,
            // long past that pod's termination.) Running two `environmentd`
            // processes is safe: the newer process will safely and correctly
            // fence out the older process via CockroachDB. In the future,
            // we'll support running multiple `environmentd` processes in
            // parallel for high availability.
            //
            // [0]: https://kubernetes.io/docs/tasks/run-application/force-delete-stateful-set-pod/#statefulset-considerations
            termination_grace_period_seconds: Some(0),
            ..Default::default()
        }),
    };

    let mut match_labels = BTreeMap::new();
    match_labels.insert(
        "materialize.cloud/name".to_owned(),
        mz.environmentd_statefulset_name(generation),
    );

    let statefulset_spec = StatefulSetSpec {
        replicas: Some(1),
        template: pod_template_spec,
        service_name: Some(mz.environmentd_service_name()),
        selector: LabelSelector {
            match_expressions: None,
            match_labels: Some(match_labels),
        },
        ..Default::default()
    };

    StatefulSet {
        metadata: ObjectMeta {
            annotations: Some(btreemap! {
                "materialize.cloud/generation".to_owned() => generation.to_string(),
                "materialize.cloud/force".to_owned() => mz.force_rollout_value(),
            }),
            ..mz.managed_resource_meta(mz.environmentd_statefulset_name(generation))
        },
        spec: Some(statefulset_spec),
        status: None,
    }
}

/// Rejects a probe that sets more than one handler (`exec`, `grpc`, `httpGet`,
/// `tcpSocket`), which Kubernetes would reject when the pod is created.
///
/// `environmentd_probe` only produces a valid probe if every layer passes
/// this check.
pub fn validate_probe_handlers(probe: &Probe) -> Result<(), anyhow::Error> {
    let handlers: Vec<_> = [
        ("exec", probe.exec.is_some()),
        ("grpc", probe.grpc.is_some()),
        ("httpGet", probe.http_get.is_some()),
        ("tcpSocket", probe.tcp_socket.is_some()),
    ]
    .into_iter()
    .filter_map(|(name, set)| set.then_some(name))
    .collect();
    if handlers.len() > 1 {
        anyhow::bail!(
            "probe sets more than one handler ({}), but at most one is allowed",
            handlers.join(", ")
        );
    }
    Ok(())
}

/// Layers the operator-wide default and then the per-environment override onto
/// `base`, so each layer only has to set the fields it changes.
///
/// Fields set in a later layer win. The handler (`exec`, `grpc`, `httpGet`,
/// `tcpSocket`) is replaced as a unit when a later layer sets any of them,
/// because Kubernetes rejects a probe with more than one handler.
fn environmentd_probe(
    base: Probe,
    operator_default: Option<&Probe>,
    environment_override: Option<&Probe>,
) -> Probe {
    [operator_default, environment_override]
        .into_iter()
        .flatten()
        .fold(base, overlay_probe)
}

fn overlay_probe(base: Probe, overlay: &Probe) -> Probe {
    // Exhaustive so that a new `Probe` field fails to compile here instead of
    // being silently dropped.
    let Probe {
        exec,
        failure_threshold,
        grpc,
        http_get,
        initial_delay_seconds,
        period_seconds,
        success_threshold,
        tcp_socket,
        termination_grace_period_seconds,
        timeout_seconds,
    } = overlay.clone();
    let has_handler =
        exec.is_some() || grpc.is_some() || http_get.is_some() || tcp_socket.is_some();
    let (exec, grpc, http_get, tcp_socket) = if has_handler {
        (exec, grpc, http_get, tcp_socket)
    } else {
        (base.exec, base.grpc, base.http_get, base.tcp_socket)
    };
    Probe {
        exec,
        failure_threshold: failure_threshold.or(base.failure_threshold),
        grpc,
        http_get,
        initial_delay_seconds: initial_delay_seconds.or(base.initial_delay_seconds),
        period_seconds: period_seconds.or(base.period_seconds),
        success_threshold: success_threshold.or(base.success_threshold),
        tcp_socket,
        termination_grace_period_seconds: termination_grace_period_seconds
            .or(base.termination_grace_period_seconds),
        timeout_seconds: timeout_seconds.or(base.timeout_seconds),
    }
}

fn create_v0_147_0_listeners_config(
    config: &super::Config,
    mz: &Materialize,
) -> v0_147_0::ListenersConfig {
    let external_enable_tls = issuer_ref_defined(
        &config.default_certificate_specs.internal,
        &mz.spec.internal_certificate_spec,
    );
    let authenticator_kind = mz.spec.authenticator_kind;

    let mut listeners_config = v0_147_0::ListenersConfig {
        sql: btreemap! {
            "external".to_owned() => SqlListenerConfig{
                addr: SocketAddr::new(
                    IpAddr::V4(Ipv4Addr::new(0,0,0,0)),
                    config.environmentd_sql_port,
                ),
                authenticator_kind,
                allowed_roles: AllowedRoles::Normal,
                enable_tls: external_enable_tls,
            },
            "internal".to_owned() => SqlListenerConfig{
                addr: SocketAddr::new(
                    IpAddr::V4(Ipv4Addr::new(0,0,0,0)),
                    config.environmentd_internal_sql_port,
                ),
                authenticator_kind: AuthenticatorKind::None,
                // Should this just be Internal?
                allowed_roles: AllowedRoles::NormalAndInternal,
                enable_tls: false,
            },
        },
        http: btreemap! {
            "external".to_owned() => v0_147_0::HttpListenerConfig{
                base: BaseListenerConfig {
                    addr: SocketAddr::new(
                        IpAddr::V4(Ipv4Addr::new(0,0,0,0)),
                        config.environmentd_http_port,
                    ),
                    // SASL authentication is only supported for SQL (PostgreSQL wire protocol).
                    // HTTP listeners must use Password authentication when SASL is enabled.
                    // This is validated at environmentd startup via ListenerConfig::validate().
                    authenticator_kind: if authenticator_kind == AuthenticatorKind::Sasl {
                        AuthenticatorKind::Password
                    } else {
                        authenticator_kind
                    },
                    allowed_roles: AllowedRoles::Normal,
                    enable_tls: external_enable_tls,
                },
                routes: v0_147_0::HttpRoutes{
                    base: true,
                    webhook: true,
                    internal: false,
                    metrics: false,
                    profiling: false,
                    mcp_agent: true,
                    mcp_developer: true,
                    console_config: true,
                },
            },
            "internal".to_owned() => v0_147_0::HttpListenerConfig{
                base: BaseListenerConfig {
                    addr: SocketAddr::new(
                        IpAddr::V4(Ipv4Addr::new(0,0,0,0)),
                        config.environmentd_internal_http_port,
                    ),
                    authenticator_kind: AuthenticatorKind::None,
                    // Should this just be Internal?
                    allowed_roles: AllowedRoles::NormalAndInternal,
                    enable_tls: false,
                },
                routes: v0_147_0::HttpRoutes{
                    base: true,
                    webhook: true,
                    internal: true,
                    metrics: true,
                    profiling: true,
                    mcp_agent: true,
                    mcp_developer: true,
                    console_config: false,
                },
            },
        },
    };

    // For password/oidc/sasl auth, we want to get rid of the internal port and
    // combine the external and internal listeners into a single listener.
    if matches!(
        authenticator_kind,
        AuthenticatorKind::Password | AuthenticatorKind::Sasl | AuthenticatorKind::Oidc
    ) {
        listeners_config.sql.remove("internal");
        listeners_config.http.remove("internal");

        listeners_config.sql.get_mut("external").map(|listener| {
            listener.allowed_roles = AllowedRoles::NormalAndInternal;
            listener
        });
        listeners_config.http.get_mut("external").map(|listener| {
            listener.base.allowed_roles = AllowedRoles::NormalAndInternal;
            listener.routes.internal = true;
            listener.routes.profiling = true;
            listener
        });

        listeners_config.http.insert(
            "metrics".to_owned(),
            v0_147_0::HttpListenerConfig {
                base: BaseListenerConfig {
                    addr: SocketAddr::new(
                        IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
                        config.environmentd_internal_http_port,
                    ),
                    authenticator_kind: AuthenticatorKind::None,
                    allowed_roles: AllowedRoles::NormalAndInternal,
                    enable_tls: false,
                },
                routes: v0_147_0::HttpRoutes {
                    base: false,
                    webhook: false,
                    internal: false,
                    metrics: true,
                    profiling: false,
                    mcp_agent: false,
                    mcp_developer: false,
                    console_config: false,
                },
            },
        );
    };
    listeners_config
}

fn create_connection_info(
    config: &super::Config,
    mz: &Materialize,
    generation: u64,
) -> ConnectionInfo {
    let external_enable_tls = issuer_ref_defined(
        &config.default_certificate_specs.internal,
        &mz.spec.internal_certificate_spec,
    );
    let authenticator_kind = mz.spec.authenticator_kind;

    let listeners_config = create_v0_147_0_listeners_config(config, mz);

    // Serve each `environmentd` the schema its version understands. Newer
    // versions parse the current (`ListenersConfig`) schema; older versions
    // parse the legacy (`v0_147_0::ListenersConfig`) schema.
    let listeners_json = if mz.meets_minimum_version(&PER_ROUTE_GROUP_ROLES_VERSION) {
        // Convert the legacy `v0_147_0::ListenersConfig` schema to the current `ListenersConfig` schema.
        // The main difference is instead of having a single `allowed_roles` property in the HTTP listener
        // for all route groups, we now have a per-route group `allowed_roles` property. The migration cascades
        // the top level `allowed_roles` into every route group.
        let mut listeners_config: v26_32_0::ListenersConfig = listeners_config.into();
        // The sole reason for the new `ListenersConfig` schema is because for password/oidc/sasl
        // authentication, we were allowing normal roles to access internal and profiling routes
        // since we now have a single listener for both external and internal traffic. However we
        // want to only allow internal roles to access these routes. Thus we set the `allowed_roles`
        // for the internal and profiling route groups to `Internal` and leave the other route groups
        // as `NormalAndInternal`.
        if matches!(
            authenticator_kind,
            AuthenticatorKind::Password | AuthenticatorKind::Sasl | AuthenticatorKind::Oidc
        ) {
            listeners_config.http.get_mut("external").map(|listener| {
                listener.routes.internal = RouteGroup::Enabled(AllowedRoles::Internal);
                listener.routes.profiling = RouteGroup::Enabled(AllowedRoles::Internal);
                listener
            });
        }

        serde_json::to_string(&VersionedListenersConfig::V2(listeners_config)).expect("known valid")
    } else {
        serde_json::to_string(&VersionedListenersConfig::V1(listeners_config)).expect("known valid")
    };
    let listeners_configmap = ConfigMap {
        binary_data: None,
        data: Some(btreemap! {
            "listeners.json".to_owned() => listeners_json,
        }),
        immutable: None,
        metadata: ObjectMeta {
            annotations: Some(btreemap! {
                "materialize.cloud/generation".to_owned() => generation.to_string(),
            }),
            ..mz.managed_resource_meta(mz.listeners_configmap_name(generation))
        },
    };

    let (scheme, leader_api_port, mz_system_secret_name) = match authenticator_kind {
        AuthenticatorKind::Password | AuthenticatorKind::Sasl | AuthenticatorKind::Oidc => {
            let scheme = if external_enable_tls { "https" } else { "http" };
            (
                scheme,
                config.environmentd_http_port,
                Some(mz.spec.backend_secret_name.clone()),
            )
        }
        _ => ("http", config.environmentd_internal_http_port, None),
    };
    let environmentd_url = format!(
        "{}://{}.{}.svc.cluster.local:{}",
        scheme,
        mz.environmentd_generation_service_name(generation),
        mz.namespace(),
        leader_api_port,
    );
    ConnectionInfo {
        environmentd_url,
        listeners_configmap,
        mz_system_secret_name,
    }
}

// see materialize/src/environmentd/src/http.rs
#[derive(Debug, Deserialize, PartialEq, Eq)]
struct BecomeLeaderResponse {
    result: BecomeLeaderResult,
}

#[derive(Debug, Deserialize, PartialEq, Eq)]
enum BecomeLeaderResult {
    Success,
    Failure { message: String },
}

#[derive(Debug, Deserialize, PartialEq, Eq)]
struct SkipCatchupError {
    message: String,
}

#[cfg(test)]
mod tests {
    use k8s_openapi::api::core::v1::{HTTPGetAction, Probe, TCPSocketAction};
    use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString;

    use super::{environmentd_probe, validate_probe_handlers};

    fn base() -> Probe {
        Probe {
            initial_delay_seconds: Some(1),
            failure_threshold: Some(12),
            tcp_socket: Some(TCPSocketAction {
                host: None,
                port: IntOrString::Int(6877),
            }),
            ..Default::default()
        }
    }

    #[mz_ore::test]
    fn environmentd_probe_without_overrides_is_base() {
        assert_eq!(environmentd_probe(base(), None, None), base());
    }

    #[mz_ore::test]
    fn environmentd_probe_layers_fields() {
        let operator_default = Probe {
            failure_threshold: Some(30),
            period_seconds: Some(20),
            ..Default::default()
        };
        let environment_override = Probe {
            period_seconds: Some(5),
            ..Default::default()
        };
        assert_eq!(
            environmentd_probe(base(), Some(&operator_default), Some(&environment_override)),
            Probe {
                failure_threshold: Some(30),
                period_seconds: Some(5),
                ..base()
            }
        );
    }

    #[mz_ore::test]
    fn environmentd_probe_replaces_handler_as_unit() {
        let http_get = HTTPGetAction {
            path: Some("/api/readyz".to_owned()),
            port: IntOrString::Int(6878),
            ..Default::default()
        };
        let environment_override = Probe {
            http_get: Some(http_get.clone()),
            ..Default::default()
        };
        assert_eq!(
            environmentd_probe(base(), None, Some(&environment_override)),
            Probe {
                tcp_socket: None,
                http_get: Some(http_get),
                ..base()
            }
        );
    }

    #[mz_ore::test]
    fn validate_probe_handlers_rejects_multiple_handlers() {
        assert!(validate_probe_handlers(&Probe::default()).is_ok());
        assert!(validate_probe_handlers(&base()).is_ok());
        let both = Probe {
            http_get: Some(HTTPGetAction {
                port: IntOrString::Int(6878),
                ..Default::default()
            }),
            ..base()
        };
        let err = validate_probe_handlers(&both).unwrap_err();
        assert!(
            err.to_string().contains("httpGet, tcpSocket"),
            "unexpected error: {err}"
        );
    }
}