refactor: introduce KubernetesResource struct

maltesander · maltesander · commit 3528930ba95b · 2026-07-10T11:41:09.000+02:00
diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs
@@ -9,6 +9,12 @@ use stackable_operator::{
     cli::OperatorEnvironmentOptions,
     cluster_resources::ClusterResourceApplyStrategy,
     commons::rbac::build_rbac_resources,
+    crd::listener::v1alpha1::Listener,
+    k8s_openapi::api::{
+        apps::v1::StatefulSet,
+        core::v1::{ConfigMap, Service},
+        policy::v1::PodDisruptionBudget,
+    },
     kube::{
         ResourceExt,
         core::{DeserializeGuard, error_boundary},
@@ -29,11 +35,7 @@ use stackable_operator::{
 use strum::{EnumDiscriminants, IntoStaticStr};
 
 use crate::{
-    controller::build::resource::{
-        listener::{build_group_listener, group_listener_name},
-        pdb::build_pdb,
-        service::{build_rolegroup_headless_service, build_rolegroup_metrics_service},
-    },
+    controller::build::resource::listener::{build_group_listener, group_listener_name},
     crd::{APP_NAME, DruidClusterStatus, DruidRole, OPERATOR_NAME, v1alpha1},
     internal_secret::create_shared_internal_secret,
 };
@@ -70,9 +72,30 @@ pub struct Ctx {
     pub operator_environment: OperatorEnvironmentOptions,
 }
 
+/// Every Kubernetes resource produced by the client-free [`build`](build::build) step.
+///
+/// The Router group `Listener` and the discovery `ConfigMap`s are intentionally *not* yet part of this
+/// bundle: the discovery `ConfigMap` derives from the *applied* Router listener's ingress address,
+/// so both are built and applied in the reconcile step instead.
+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(Snafu, Debug, EnumDiscriminants)]
 #[strum_discriminants(derive(IntoStaticStr))]
 pub enum Error {
+    #[snafu(display("failed to build the Kubernetes resources"))]
+    BuildResources { source: build::Error },
+
+    #[snafu(display("failed to apply Kubernetes resource"))]
+    ApplyResource {
+        source: stackable_operator::cluster_resources::Error,
+    },
+
     #[snafu(display("failed to apply Service for role group {role_group}"))]
     ApplyRoleGroupService {
         source: stackable_operator::cluster_resources::Error,
@@ -234,99 +257,76 @@ pub async fn reconcile_druid(
         .await
         .context(FailedInternalSecretCreationSnafu)?;
 
-    let mut ss_cond_builder = StatefulSetConditionBuilder::default();
+    let resources =
+        build::build(&validated_cluster, &service_account_name).context(BuildResourcesSnafu)?;
 
-    for (druid_role, groups) in validated_cluster.role_group_configs.iter() {
-        for (rolegroup_name, rg) in groups.iter() {
-            let rg_headless_service =
-                build_rolegroup_headless_service(&validated_cluster, druid_role, rolegroup_name);
-            let rg_metrics_service =
-                build_rolegroup_metrics_service(&validated_cluster, druid_role, rolegroup_name);
-
-            let rg_configmap = build::resource::config_map::build_rolegroup_config_map(
-                &validated_cluster,
-                druid_role,
-                rolegroup_name,
-                rg,
-            )
-            .context(BuildConfigMapSnafu)?;
-            let rg_statefulset = build::resource::statefulset::build_rolegroup_statefulset(
-                &validated_cluster,
-                druid_role,
-                rolegroup_name,
-                rg,
-                &service_account_name,
-            )
-            .context(BuildRoleGroupStatefulSetSnafu)?;
+    let mut ss_cond_builder = StatefulSetConditionBuilder::default();
 
+    // Apply order: everything a Pod mounts (ConfigMaps) must exist before the StatefulSets, so the
+    // StatefulSets are applied last to prevent unnecessary Pod restarts.
+    // See https://github.com/stackabletech/commons-operator/issues/111 for details.
+    for service in resources.services {
+        cluster_resources
+            .add(client, service)
+            .await
+            .context(ApplyResourceSnafu)?;
+    }
+    for listener in resources.listeners {
+        cluster_resources
+            .add(client, listener)
+            .await
+            .context(ApplyResourceSnafu)?;
+    }
+    for config_map in resources.config_maps {
+        cluster_resources
+            .add(client, config_map)
+            .await
+            .context(ApplyResourceSnafu)?;
+    }
+    for pdb in resources.pod_disruption_budgets {
+        cluster_resources
+            .add(client, pdb)
+            .await
+            .context(ApplyResourceSnafu)?;
+    }
+    for stateful_set in resources.stateful_sets {
+        ss_cond_builder.add(
             cluster_resources
-                .add(client, rg_headless_service)
-                .await
-                .with_context(|_| ApplyRoleGroupServiceSnafu {
-                    role_group: rolegroup_name.to_string(),
-                })?;
-            cluster_resources
-                .add(client, rg_metrics_service)
-                .await
-                .with_context(|_| ApplyRoleGroupServiceSnafu {
-                    role_group: rolegroup_name.to_string(),
-                })?;
-            cluster_resources
-                .add(client, rg_configmap)
+                .add(client, stateful_set)
                 .await
-                .with_context(|_| ApplyRoleGroupConfigSnafu {
-                    role_group: rolegroup_name.to_string(),
-                })?;
-
-            // 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.
-            ss_cond_builder.add(
-                cluster_resources
-                    .add(client, rg_statefulset)
-                    .await
-                    .with_context(|_| ApplyRoleGroupStatefulSetSnafu {
-                        role_group: rolegroup_name.to_string(),
-                    })?,
-            );
-        }
+                .context(ApplyResourceSnafu)?,
+        );
+    }
 
-        if let Some(listener_class) = &validated_cluster.role_config(druid_role).listener_class
-            && let Some(listener_group_name) = group_listener_name(&validated_cluster, druid_role)
-        {
-            let role_group_listener = build_group_listener(
-                &validated_cluster,
-                listener_class,
-                listener_group_name,
-                druid_role,
-            );
-
-            let listener = cluster_resources
-                .add(client, role_group_listener)
-                .await
-                .context(ApplyGroupListenerSnafu)?;
-
-            if *druid_role == DruidRole::Router {
-                // discovery
-                for discovery_cm in build_discovery_configmaps(&validated_cluster, listener)
-                    .await
-                    .context(BuildDiscoveryConfigSnafu)?
-                {
-                    cluster_resources
-                        .add(client, discovery_cm)
-                        .await
-                        .context(ApplyDiscoveryConfigSnafu)?;
-                }
-            }
-        }
+    // The Router group Listener and its discovery ConfigMaps are applied here rather than in the
+    // build step: the discovery ConfigMap derives from the *applied* Router listener's ingress
+    // address, which is only known after the Listener has been applied.
+    if let Some(listener_class) = &validated_cluster
+        .role_config(&DruidRole::Router)
+        .listener_class
+        && let Some(listener_group_name) =
+            group_listener_name(&validated_cluster, &DruidRole::Router)
+    {
+        let router_listener = build_group_listener(
+            &validated_cluster,
+            listener_class,
+            listener_group_name,
+            &DruidRole::Router,
+        );
 
-        let role_config = validated_cluster.role_config(druid_role);
+        let listener = cluster_resources
+            .add(client, router_listener)
+            .await
+            .context(ApplyGroupListenerSnafu)?;
 
-        if let Some(pdb) = build_pdb(&role_config.pdb, &validated_cluster, druid_role) {
+        for discovery_cm in build_discovery_configmaps(&validated_cluster, listener)
+            .await
+            .context(BuildDiscoveryConfigSnafu)?
+        {
             cluster_resources
-                .add(client, pdb)
+                .add(client, discovery_cm)
                 .await
-                .context(ApplyPdbSnafu)?;
+                .context(ApplyDiscoveryConfigSnafu)?;
         }
     }
 
diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs
@@ -2,8 +2,24 @@
 
 use std::str::FromStr;
 
+use snafu::{ResultExt, Snafu};
 use stackable_operator::v2::types::operator::{ProductVersion, RoleGroupName};
 
+use crate::{
+    controller::{
+        KubernetesResources,
+        build::resource::{
+            config_map::build_rolegroup_config_map,
+            listener::{build_group_listener, group_listener_name},
+            pdb::build_pdb,
+            service::{build_rolegroup_headless_service, build_rolegroup_metrics_service},
+            statefulset::build_rolegroup_statefulset,
+        },
+        validate::ValidatedCluster,
+    },
+    crd::DruidRole,
+};
+
 // Placeholder role-group name used for the recommended labels of the role-level discovery
 // `ConfigMap` (which is not tied to a single role group).
 stackable_operator::constant!(pub(crate) PLACEHOLDER_DISCOVERY_ROLE_GROUP: RoleGroupName = "discovery");
@@ -22,3 +38,154 @@ pub mod jvm;
 pub mod properties;
 pub mod resource;
 pub mod security;
+
+#[derive(Snafu, Debug)]
+pub enum Error {
+    #[snafu(display("failed to build ConfigMap for role group {role_group}"))]
+    ConfigMap {
+        source: resource::config_map::Error,
+        role_group: RoleGroupName,
+    },
+
+    #[snafu(display("failed to build StatefulSet for role group {role_group}"))]
+    StatefulSet {
+        source: resource::statefulset::Error,
+        role_group: RoleGroupName,
+    },
+}
+
+/// Builds the Kubernetes resources 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. The remaining errors 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).
+///
+/// The Router group `Listener` and the discovery `ConfigMap`s are not built here: the discovery
+/// `ConfigMap` derives from the *applied* Router listener's ingress address, so both are built and
+/// applied in the reconcile step instead.
+pub fn build(
+    cluster: &ValidatedCluster,
+    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 (druid_role, role_group_configs) in &cluster.role_group_configs {
+        let role_config = cluster.role_config(druid_role);
+
+        if let Some(pdb) = build_pdb(&role_config.pdb, cluster, druid_role) {
+            pod_disruption_budgets.push(pdb);
+        }
+
+        // The Router group Listener is built and applied in the reconcile step instead (see the
+        // module docs and [`KubernetesResources`]), so it is skipped here.
+        if *druid_role != DruidRole::Router
+            && let Some(listener_class) = &role_config.listener_class
+            && let Some(listener_group_name) = group_listener_name(cluster, druid_role)
+        {
+            listeners.push(build_group_listener(
+                cluster,
+                listener_class,
+                listener_group_name,
+                druid_role,
+            ));
+        }
+
+        for (role_group_name, rg) in role_group_configs {
+            services.push(build_rolegroup_headless_service(
+                cluster,
+                druid_role,
+                role_group_name,
+            ));
+            services.push(build_rolegroup_metrics_service(
+                cluster,
+                druid_role,
+                role_group_name,
+            ));
+            config_maps.push(
+                build_rolegroup_config_map(cluster, druid_role, role_group_name, rg).context(
+                    ConfigMapSnafu {
+                        role_group: role_group_name.clone(),
+                    },
+                )?,
+            );
+            stateful_sets.push(
+                build_rolegroup_statefulset(
+                    cluster,
+                    druid_role,
+                    role_group_name,
+                    rg,
+                    service_account_name,
+                )
+                .context(StatefulSetSnafu {
+                    role_group: role_group_name.clone(),
+                })?,
+            );
+        }
+    }
+
+    Ok(KubernetesResources {
+        stateful_sets,
+        services,
+        listeners,
+        config_maps,
+        pod_disruption_budgets,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use stackable_operator::kube::Resource;
+
+    use super::build;
+    use crate::controller::validate::test_support::{
+        MINIMAL_DRUID_YAML, druid_from_yaml, validated_cluster,
+    };
+
+    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 druid = druid_from_yaml(MINIMAL_DRUID_YAML);
+        let cluster = validated_cluster(&druid);
+        let resources = build(&cluster, "simple-druid-serviceaccount").expect("build succeeds");
+
+        // One StatefulSet and one ConfigMap per role group (one role group per role).
+        let expected_role_group_names = [
+            "simple-druid-broker-default",
+            "simple-druid-coordinator-default",
+            "simple-druid-historical-default",
+            "simple-druid-middlemanager-default",
+            "simple-druid-router-default",
+        ];
+        assert_eq!(
+            sorted_names(&resources.stateful_sets),
+            expected_role_group_names
+        );
+        assert_eq!(
+            sorted_names(&resources.config_maps),
+            expected_role_group_names
+        );
+
+        // Group Listeners are built for the externally reachable roles except the Router (whose
+        // Listener is applied in the reconcile step): Broker and Coordinator.
+        // TODO: add router listener here once properly build in the built step.
+        assert_eq!(
+            sorted_names(&resources.listeners),
+            ["simple-druid-broker", "simple-druid-coordinator"]
+        );
+    }
+}
