refactor: Introduce build aggregator

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

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

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<KubernetesResources, Error> {
+    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"]
+        );
+    }
+}

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(

rust/operator-binary/src/controller/build/resource/statefulset.rs

@@ -40,14 +40,16 @@ 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::{
                 LISTENER_VOLUME_DIR, LISTENER_VOLUME_NAME, build_group_listener_pvc,
                 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");

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<StatefulSet>,
+    pub services: Vec<Service>,
+    pub listeners: Vec<Listener>,
+    pub config_maps: Vec<ConfigMap>,
+    pub pod_disruption_budgets: Vec<PodDisruptionBudget>,
+}
 
 #[derive(Clone, Debug)]
 pub struct ValidatedTls {