diff --git a/CHANGELOG.md b/CHANGELOG.md index 998c9b32..2177753f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,13 @@ All notable changes to this project will be documented in this file. ## [Unreleased] +### Changed + +- Internal operator refactoring: introduce a build() step in the reconciler that + assembles all relevant Kubernetes resources before anything is applied ([#909]). + +[#909]: https://github.com/stackabletech/trino-operator/pull/909 + ## [26.7.0] - 2026-07-21 ## [26.7.0-rc1] - 2026-07-16 diff --git a/rust/operator-binary/src/authentication/password/file.rs b/rust/operator-binary/src/authentication/password/file.rs index 650e951b..a8c5e307 100644 --- a/rust/operator-binary/src/authentication/password/file.rs +++ b/rust/operator-binary/src/authentication/password/file.rs @@ -20,7 +20,7 @@ use stackable_operator::{ use crate::{ authentication::password::PASSWORD_AUTHENTICATOR_NAME, - controller::build::resource::statefulset::LOG_VOLUME_NAME, trino_controller::STACKABLE_LOG_DIR, + controller::{STACKABLE_LOG_DIR, build::resource::statefulset::LOG_VOLUME_NAME}, }; // mounts diff --git a/rust/operator-binary/src/controller/build.rs b/rust/operator-binary/src/controller/build.rs index 6363166b..7f4e4109 100644 --- a/rust/operator-binary/src/controller/build.rs +++ b/rust/operator-binary/src/controller/build.rs @@ -2,7 +2,24 @@ use std::str::FromStr; -use stackable_operator::v2::types::operator::RoleGroupName; +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + utils::cluster_info::KubernetesClusterInfo, v2::types::operator::RoleGroupName, +}; + +use crate::controller::{ + KubernetesResources, ValidatedCluster, + build::resource::{ + config_map, + listener::{build_group_listener, group_listener_name}, + pdb::build_pdb, + service::{ + build_rolegroup_headless_service, build_rolegroup_metrics_service, + headless_service_ports, + }, + statefulset, + }, +}; pub mod command; pub mod graceful_shutdown; @@ -13,4 +30,193 @@ pub mod resource; // Placeholder role-group name used for the recommended labels of a role's group listener. // The group listener is owned by the role (not a single role-group), so there is no real // role-group to attribute it to. -stackable_operator::constant!(pub(crate) PLACEHOLDER_LISTENER_ROLE_GROUP: RoleGroupName = "none"); +stackable_operator::constant!(PLACEHOLDER_LISTENER_ROLE_GROUP: RoleGroupName = "none"); + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to build ConfigMap for role group {role_group}"))] + ConfigMap { + source: config_map::Error, + role_group: RoleGroupName, + }, + + #[snafu(display("failed to build StatefulSet for role group {role_group}"))] + StatefulSet { + source: statefulset::Error, + role_group: RoleGroupName, + }, +} + +/// Builds every Kubernetes resource for the given validated cluster. +/// +/// Does not need a Kubernetes client: every reference to another Kubernetes resource is already +/// dereferenced and validated by this point, so the errors returned here are resource-assembly +/// failures only. +/// +/// `service_account_name` is the name of the RBAC `ServiceAccount` the role-group Pods run under +/// (RBAC resources are built and applied separately, in the reconcile step). +pub fn build( + cluster: &ValidatedCluster, + cluster_info: &KubernetesClusterInfo, + service_account_name: &str, +) -> Result { + let mut stateful_sets = vec![]; + let mut services = vec![]; + let mut listeners = vec![]; + let mut config_maps = vec![]; + let mut pod_disruption_budgets = vec![]; + + for (role, role_group_configs) in &cluster.role_group_configs { + for (role_group_name, role_group_config) in role_group_configs { + let recommended_labels = cluster.recommended_labels(role, role_group_name); + let selector = cluster.role_group_selector(role, role_group_name); + + services.push(build_rolegroup_headless_service( + cluster, + role, + role_group_name, + &recommended_labels, + selector.clone().into(), + headless_service_ports(cluster), + )); + services.push(build_rolegroup_metrics_service( + cluster, + role, + role_group_name, + &recommended_labels, + selector.into(), + )); + config_maps.push( + config_map::build_rolegroup_config_map( + cluster, + role, + role_group_name, + cluster_info, + &recommended_labels, + ) + .context(ConfigMapSnafu { + role_group: role_group_name.clone(), + })?, + ); + config_maps.push( + config_map::build_rolegroup_catalog_config_map( + cluster, + role, + role_group_name, + &recommended_labels, + ) + .context(ConfigMapSnafu { + role_group: role_group_name.clone(), + })?, + ); + stateful_sets.push( + statefulset::build_rolegroup_statefulset( + cluster, + role, + role_group_name, + role_group_config, + service_account_name, + ) + .context(StatefulSetSnafu { + role_group: role_group_name.clone(), + })?, + ); + } + + let Some(role_config) = cluster.role_config(role) else { + continue; + }; + + if let Some(listener_class) = &role_config.listener_class + && let Some(listener_group_name) = group_listener_name(cluster, role) + { + listeners.push(build_group_listener( + cluster, + cluster.recommended_labels(role, &PLACEHOLDER_LISTENER_ROLE_GROUP), + listener_class, + listener_group_name, + )); + } + + pod_disruption_budgets.extend(build_pdb(&role_config.pdb, cluster, role)); + } + + Ok(KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + pod_disruption_budgets, + }) +} + +#[cfg(test)] +mod tests { + use stackable_operator::{ + commons::networking::DomainName, kube::Resource, utils::cluster_info::KubernetesClusterInfo, + }; + + use super::build; + use crate::controller::validated_cluster; + + /// Collects the `.metadata.name`s of the given resources, sorted for stable comparison. + fn sorted_names(resources: &[impl Resource]) -> Vec<&str> { + let mut names: Vec<&str> = resources + .iter() + .filter_map(|resource| resource.meta().name.as_deref()) + .collect(); + names.sort(); + names + } + + #[test] + fn build_produces_expected_resource_names() { + let cluster = validated_cluster(); + let cluster_info = KubernetesClusterInfo { + cluster_domain: DomainName::try_from("cluster.local") + .expect("cluster.local is a valid domain name"), + }; + + let resources = + build(&cluster, &cluster_info, "simple-trino-serviceaccount").expect("build succeeds"); + + // One StatefulSet per role group. + assert_eq!( + sorted_names(&resources.stateful_sets), + [ + "simple-trino-coordinator-default", + "simple-trino-worker-default", + ] + ); + // One headless and one metrics Service per role group. + assert_eq!( + sorted_names(&resources.services), + [ + "simple-trino-coordinator-default-headless", + "simple-trino-coordinator-default-metrics", + "simple-trino-worker-default-headless", + "simple-trino-worker-default-metrics", + ] + ); + // A config ConfigMap and a catalog ConfigMap per role group. + assert_eq!( + sorted_names(&resources.config_maps), + [ + "simple-trino-coordinator-default", + "simple-trino-coordinator-default-catalog", + "simple-trino-worker-default", + "simple-trino-worker-default-catalog", + ] + ); + // The coordinator is the only role with a group Listener. + assert_eq!( + sorted_names(&resources.listeners), + ["simple-trino-coordinator"] + ); + // A default PodDisruptionBudget per role. + assert_eq!( + sorted_names(&resources.pod_disruption_budgets), + ["simple-trino-coordinator", "simple-trino-worker"] + ); + } +} diff --git a/rust/operator-binary/src/controller/build/command.rs b/rust/operator-binary/src/controller/build/command.rs index f2121fd9..cb667ae2 100644 --- a/rust/operator-binary/src/controller/build/command.rs +++ b/rust/operator-binary/src/controller/build/command.rs @@ -12,14 +12,16 @@ use crate::{ authentication::TrinoAuthenticationConfig, catalog::config::CatalogConfig, config::{client_protocol, fault_tolerant_execution}, - controller::{ValidatedCluster, ValidatedTrinoConfig, build::properties::ConfigFileName}, + controller::{ + STACKABLE_LOG_CONFIG_DIR, STACKABLE_LOG_DIR, ValidatedCluster, ValidatedTrinoConfig, + build::properties::ConfigFileName, + }, crd::{ CONFIG_DIR_NAME, Container, RW_CONFIG_DIR_NAME, STACKABLE_CLIENT_TLS_DIR, STACKABLE_INTERNAL_TLS_DIR, STACKABLE_MOUNT_INTERNAL_TLS_DIR, STACKABLE_MOUNT_SERVER_TLS_DIR, STACKABLE_SERVER_TLS_DIR, STACKABLE_TLS_STORE_PASSWORD, TrinoRole, catalog::TrinoCatalogName, }, - trino_controller::{STACKABLE_LOG_CONFIG_DIR, STACKABLE_LOG_DIR}, }; pub fn container_prepare_args( diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index a15fc106..20668630 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -40,7 +40,8 @@ use stackable_operator::{ use crate::{ authorization::opa::OPA_TLS_VOLUME_NAME, controller::{ - RoleGroupName, TrinoRoleGroupConfig, ValidatedCluster, build, + MAX_PREPARE_LOG_FILE_SIZE, RoleGroupName, STACKABLE_LOG_CONFIG_DIR, STACKABLE_LOG_DIR, + TrinoRoleGroupConfig, ValidatedCluster, build, build::{ command, resource::listener::{ @@ -48,6 +49,7 @@ use crate::{ group_listener_name, secret_volume_listener_scope, }, }, + shared_internal_secret_name, shared_spooling_secret_name, }, crd::{ CONFIG_DIR_NAME, Container, ENV_INTERNAL_SECRET, ENV_SPOOLING_SECRET, HTTP_PORT, @@ -57,10 +59,6 @@ use crate::{ STACKABLE_MOUNT_SERVER_TLS_DIR, STACKABLE_SERVER_TLS_DIR, STACKABLE_TLS_STORE_PASSWORD, TrinoRole, }, - trino_controller::{ - MAX_PREPARE_LOG_FILE_SIZE, STACKABLE_LOG_CONFIG_DIR, STACKABLE_LOG_DIR, - shared_internal_secret_name, shared_spooling_secret_name, - }, }; stackable_operator::constant!(VECTOR_CONTAINER_NAME: ContainerName = "vector"); diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs index 8f1482bb..7e00d944 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -7,8 +7,15 @@ use stackable_operator::{ product_image_selection::ResolvedProductImage, resources::{NoRuntimeLimits, Resources}, }, + crd::listener::v1alpha1::Listener, + k8s_openapi::api::{ + apps::v1::StatefulSet, + core::v1::{ConfigMap, Service}, + policy::v1::PodDisruptionBudget, + }, kube::{Resource, api::ObjectMeta}, kvp::Labels, + memory::{BinaryMultiple, MemoryQuantity}, shared::time::Duration, v2::{ HasName, HasUid, NameIsValidLabelValue, @@ -40,7 +47,31 @@ pub(crate) mod build; pub(crate) mod dereference; pub(crate) mod validate; +pub use stackable_operator::v2::product_logging::framework::STACKABLE_LOG_DIR; pub use validate::{RoleGroupName, TrinoRoleGroupConfig}; +pub const STACKABLE_LOG_CONFIG_DIR: &str = "/stackable/log_config"; + +pub const MAX_PREPARE_LOG_FILE_SIZE: MemoryQuantity = MemoryQuantity { + value: 1.0, + unit: BinaryMultiple::Mebi, +}; + +pub(crate) fn shared_internal_secret_name(cluster_name: &ClusterName) -> String { + format!("{cluster_name}-internal-secret") +} + +pub(crate) fn shared_spooling_secret_name(cluster_name: &ClusterName) -> String { + format!("{cluster_name}-spooling-secret") +} + +/// Every Kubernetes resource produced by the client-free [`build()`](build::build) step. +pub struct KubernetesResources { + pub stateful_sets: Vec, + pub services: Vec, + pub listeners: Vec, + pub config_maps: Vec, + pub pod_disruption_budgets: Vec, +} #[derive(Clone, Debug)] pub struct ValidatedTls { diff --git a/rust/operator-binary/src/trino_controller.rs b/rust/operator-binary/src/trino_controller.rs index 05ed690f..8c9741eb 100644 --- a/rust/operator-binary/src/trino_controller.rs +++ b/rust/operator-binary/src/trino_controller.rs @@ -13,28 +13,19 @@ use stackable_operator::{ runtime::controller::Action, }, logging::controller::ReconcilerError, - memory::{BinaryMultiple, MemoryQuantity}, shared::time::Duration, status::condition::{ compute_conditions, operations::ClusterOperationsConditionBuilder, statefulset::StatefulSetConditionBuilder, }, - v2::{cluster_resources::cluster_resources_new, types::operator::ClusterName}, + v2::cluster_resources::cluster_resources_new, }; use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ controller::{ - RoleGroupName, build, - build::resource::{ - listener::{build_group_listener, group_listener_name}, - pdb::build_pdb, - service::{ - build_rolegroup_headless_service, build_rolegroup_metrics_service, - headless_service_ports, - }, - }, - controller_name, dereference, operator_name, product_name, validate, + ValidatedCluster, build, controller_name, dereference, operator_name, product_name, + shared_internal_secret_name, shared_spooling_secret_name, validate, }, crd::{APP_NAME, ENV_INTERNAL_SECRET, ENV_SPOOLING_SECRET, v1alpha1}, }; @@ -48,53 +39,22 @@ pub const OPERATOR_NAME: &str = "trino.stackable.tech"; pub const CONTROLLER_NAME: &str = "trinocluster"; pub const FULL_CONTROLLER_NAME: &str = concatcp!(CONTROLLER_NAME, '.', OPERATOR_NAME); -pub use stackable_operator::v2::product_logging::framework::STACKABLE_LOG_DIR; -pub const STACKABLE_LOG_CONFIG_DIR: &str = "/stackable/log_config"; - -pub const MAX_PREPARE_LOG_FILE_SIZE: MemoryQuantity = MemoryQuantity { - value: 1.0, - unit: BinaryMultiple::Mebi, -}; - pub(crate) const CONTAINER_IMAGE_BASE_NAME: &str = "trino"; #[derive(Snafu, Debug, EnumDiscriminants)] #[strum_discriminants(derive(IntoStaticStr))] -#[allow(clippy::enum_variant_names)] pub enum Error { #[snafu(display("failed to delete orphaned resources"))] DeleteOrphanedResources { source: stackable_operator::cluster_resources::Error, }, - #[snafu(display("failed to apply Service for {}", rolegroup))] - ApplyRoleGroupService { - source: stackable_operator::cluster_resources::Error, - rolegroup: RoleGroupName, - }, - - #[snafu(display("failed to build ConfigMap for {}", rolegroup))] - BuildRoleGroupConfigMap { - source: build::resource::config_map::Error, - rolegroup: RoleGroupName, - }, - - #[snafu(display("failed to apply ConfigMap for {}", rolegroup))] - ApplyRoleGroupConfig { - source: stackable_operator::cluster_resources::Error, - rolegroup: RoleGroupName, - }, - - #[snafu(display("failed to build StatefulSet for {}", rolegroup))] - BuildRoleGroupStatefulSet { - source: build::resource::statefulset::Error, - rolegroup: RoleGroupName, - }, + #[snafu(display("failed to build the Kubernetes resources"))] + BuildResources { source: build::Error }, - #[snafu(display("failed to apply StatefulSet for {}", rolegroup))] - ApplyRoleGroupStatefulSet { + #[snafu(display("failed to apply Kubernetes resource"))] + ApplyResource { source: stackable_operator::cluster_resources::Error, - rolegroup: RoleGroupName, }, #[snafu(display("failed to patch service account"))] @@ -117,11 +77,6 @@ pub enum Error { source: stackable_operator::commons::rbac::Error, }, - #[snafu(display("failed to apply PodDisruptionBudget"))] - ApplyPdb { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to get required Labels"))] GetRequiredLabels { source: @@ -144,11 +99,6 @@ pub enum Error { #[snafu(display("failed to validate cluster"))] ValidateCluster { source: validate::Error }, - #[snafu(display("failed to apply group listener"))] - ApplyGroupListener { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to create internal secret"))] CreateInternalSecret { source: random_secret_creation::Error, @@ -212,7 +162,11 @@ pub async fn reconcile_trino( ) .context(BuildRbacResourcesSnafu)?; - let rbac_sa = cluster_resources + // The ServiceAccount name is deterministic on the built object, so the build step does not + // depend on the applied ServiceAccount. + let service_account_name = rbac_sa.name_any(); + + cluster_resources .add(client, rbac_sa) .await .context(ApplyServiceAccountSnafu)?; @@ -222,156 +176,55 @@ pub async fn reconcile_trino( .await .context(ApplyRoleBindingSnafu)?; - random_secret_creation::create_random_secret_if_not_exists( - &shared_internal_secret_name(&validated_cluster.name), - ENV_INTERNAL_SECRET, - 512, - &validated_cluster, - client, - ) - .await - .context(CreateInternalSecretSnafu)?; + ensure_random_secrets(client, &validated_cluster).await?; - // This secret is created even if spooling is not configured. - // Trino currently requires the secret to be exactly 256 bits long. - random_secret_creation::create_random_secret_if_not_exists( - &shared_spooling_secret_name(&validated_cluster.name), - ENV_SPOOLING_SECRET, - 32, + let resources = build::build( &validated_cluster, - client, + &client.kubernetes_cluster_info, + &service_account_name, ) - .await - .context(CreateInternalSecretSnafu)?; + .context(BuildResourcesSnafu)?; let mut sts_cond_builder = StatefulSetConditionBuilder::default(); - for (trino_role, role_group_configs) in &validated_cluster.role_group_configs { - for (role_group_name, rg) in role_group_configs { - let role_group_service_recommended_labels = - validated_cluster.recommended_labels(trino_role, role_group_name); - - let role_group_service_selector = - validated_cluster.role_group_selector(trino_role, role_group_name); - - let rg_headless_service = build_rolegroup_headless_service( - &validated_cluster, - trino_role, - role_group_name, - &role_group_service_recommended_labels, - role_group_service_selector.clone().into(), - headless_service_ports(&validated_cluster), - ); - - let rg_metrics_service = build_rolegroup_metrics_service( - &validated_cluster, - trino_role, - role_group_name, - &role_group_service_recommended_labels, - role_group_service_selector.into(), - ); - - let rg_configmap = build::resource::config_map::build_rolegroup_config_map( - &validated_cluster, - trino_role, - role_group_name, - &client.kubernetes_cluster_info, - &role_group_service_recommended_labels, - ) - .with_context(|_| BuildRoleGroupConfigMapSnafu { - rolegroup: role_group_name.clone(), - })?; - - let rg_catalog_configmap = - build::resource::config_map::build_rolegroup_catalog_config_map( - &validated_cluster, - trino_role, - role_group_name, - &role_group_service_recommended_labels, - ) - .with_context(|_| BuildRoleGroupConfigMapSnafu { - rolegroup: role_group_name.clone(), - })?; - - let rg_stateful_set = build::resource::statefulset::build_rolegroup_statefulset( - &validated_cluster, - trino_role, - role_group_name, - rg, - &rbac_sa.name_any(), - ) - .with_context(|_| BuildRoleGroupStatefulSetSnafu { - rolegroup: role_group_name.clone(), - })?; - - cluster_resources - .add(client, rg_headless_service) - .await - .with_context(|_| ApplyRoleGroupServiceSnafu { - rolegroup: role_group_name.clone(), - })?; - - cluster_resources - .add(client, rg_metrics_service) - .await - .with_context(|_| ApplyRoleGroupServiceSnafu { - rolegroup: role_group_name.clone(), - })?; - - cluster_resources - .add(client, rg_configmap) - .await - .with_context(|_| ApplyRoleGroupConfigSnafu { - rolegroup: role_group_name.clone(), - })?; + for service in resources.services { + cluster_resources + .add(client, service) + .await + .context(ApplyResourceSnafu)?; + } - cluster_resources - .add(client, rg_catalog_configmap) - .await - .with_context(|_| ApplyRoleGroupConfigSnafu { - rolegroup: role_group_name.clone(), - })?; - - // Note: The StatefulSet needs to be applied after all ConfigMaps and Secrets it mounts - // to prevent unnecessary Pod restarts. - // See https://github.com/stackabletech/commons-operator/issues/111 for details. - sts_cond_builder.add( - cluster_resources - .add(client, rg_stateful_set) - .await - .with_context(|_| ApplyRoleGroupStatefulSetSnafu { - rolegroup: role_group_name.clone(), - })?, - ); - } - - let Some(role_config) = validated_cluster.role_config(trino_role) else { - continue; - }; + for listener in resources.listeners { + cluster_resources + .add(client, listener) + .await + .context(ApplyResourceSnafu)?; + } - if let Some(listener_class) = &role_config.listener_class - && let Some(listener_group_name) = group_listener_name(&validated_cluster, trino_role) - { - let role_group_listener = build_group_listener( - &validated_cluster, - validated_cluster - .recommended_labels(trino_role, &build::PLACEHOLDER_LISTENER_ROLE_GROUP), - listener_class, - listener_group_name, - ); + for config_map in resources.config_maps { + cluster_resources + .add(client, config_map) + .await + .context(ApplyResourceSnafu)?; + } - cluster_resources - .add(client, role_group_listener) - .await - .context(ApplyGroupListenerSnafu)?; - } + for pdb in resources.pod_disruption_budgets { + cluster_resources + .add(client, pdb) + .await + .context(ApplyResourceSnafu)?; + } - if let Some(pdb) = build_pdb(&role_config.pdb, &validated_cluster, trino_role) { + // Note: The StatefulSet needs to be applied after all ConfigMaps and Secrets it mounts + // to prevent unnecessary Pod restarts. + // See https://github.com/stackabletech/commons-operator/issues/111 for details. + for stateful_set in resources.stateful_sets { + sts_cond_builder.add( cluster_resources - .add(client, pdb) + .add(client, stateful_set) .await - .context(ApplyPdbSnafu)?; - } + .context(ApplyResourceSnafu)?, + ); } let cluster_operation_cond_builder = @@ -396,6 +249,37 @@ pub async fn reconcile_trino( Ok(Action::await_change()) } +/// Ensures the two shared random Secrets (internal communication and spooling) exist, creating +/// any that are missing. +async fn ensure_random_secrets( + client: &stackable_operator::client::Client, + cluster: &ValidatedCluster, +) -> Result<()> { + random_secret_creation::create_random_secret_if_not_exists( + &shared_internal_secret_name(&cluster.name), + ENV_INTERNAL_SECRET, + 512, + cluster, + client, + ) + .await + .context(CreateInternalSecretSnafu)?; + + // This secret is created even if spooling is not configured. + // Trino currently requires the secret to be exactly 256 bits long. + random_secret_creation::create_random_secret_if_not_exists( + &shared_spooling_secret_name(&cluster.name), + ENV_SPOOLING_SECRET, + 32, + cluster, + client, + ) + .await + .context(CreateInternalSecretSnafu)?; + + Ok(()) +} + pub fn error_policy( _obj: Arc>, error: &Error, @@ -407,14 +291,6 @@ pub fn error_policy( } } -pub(crate) fn shared_internal_secret_name(cluster_name: &ClusterName) -> String { - format!("{cluster_name}-internal-secret") -} - -pub(crate) fn shared_spooling_secret_name(cluster_name: &ClusterName) -> String { - format!("{cluster_name}-spooling-secret") -} - #[cfg(test)] mod tests { use std::str::FromStr; @@ -432,7 +308,7 @@ mod tests { client_protocol::ResolvedClientProtocolConfig, fault_tolerant_execution::ResolvedFaultTolerantExecutionConfig, }, - controller::dereference::DereferencedObjects, + controller::{RoleGroupName, dereference::DereferencedObjects}, crd::{ENV_SPOOLING_SECRET, TrinoRole, v1alpha1}, };