refactor: Introduce build aggregator

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

@@ -4,8 +4,23 @@
 
 use std::str::FromStr;
 
+use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::v2::types::{common::Port, operator::RoleGroupName};
 
+use crate::{
+    controller::{
+        KubernetesResources, ValidatedCluster,
+        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_node_rolegroup_statefulset,
+        },
+    },
+    crd::NifiRole,
+};
+
 pub mod git_sync;
 pub mod graceful_shutdown;
 pub mod jvm;
@@ -27,3 +42,131 @@ pub const BALANCE_PORT: Port = Port(6243);
 // Filesystem paths shared by multiple builders. Single-consumer paths live in their builder.
 pub const NIFI_CONFIG_DIRECTORY: &str = "/stackable/nifi/conf";
 pub const NIFI_PYTHON_WORKING_DIRECTORY: &str = "/nifi-python-working-directory";
+
+#[derive(Snafu, Debug)]
+pub enum Error {
+    #[snafu(display("NifiCluster has no nodes role defined"))]
+    NoNodesDefined,
+
+    #[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 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,
+    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![];
+
+    // NiFi has a single role (`node`), which must always be present.
+    let nifi_role = NifiRole::Node;
+    let node_role_group_configs = cluster
+        .role_group_configs
+        .get(&nifi_role)
+        .context(NoNodesDefinedSnafu)?;
+
+    // Role-level resources (one per role): the PodDisruptionBudget and the group Listener.
+    let role_config = &cluster.role_config;
+    if let Some(pdb) = build_pdb(&role_config.pdb, cluster, &nifi_role) {
+        pod_disruption_budgets.push(pdb);
+    }
+    listeners.push(build_group_listener(
+        cluster,
+        role_config.listener_class.clone(),
+        group_listener_name(cluster, &nifi_role.to_string()),
+    ));
+
+    for (role_group_name, rg) in node_role_group_configs {
+        services.push(build_rolegroup_headless_service(cluster, role_group_name));
+        services.push(build_rolegroup_metrics_service(cluster, role_group_name));
+
+        config_maps.push(
+            build_rolegroup_config_map(cluster, role_group_name, rg).context(ConfigMapSnafu {
+                role_group: role_group_name.clone(),
+            })?,
+        );
+
+        let effective_replicas = rg.replicas.map(i32::from);
+        stateful_sets.push(
+            build_node_rolegroup_statefulset(
+                cluster,
+                role_group_name,
+                rg,
+                effective_replicas,
+                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, properties::test_support::minimal_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_resources() {
+        let cluster = minimal_validated_cluster();
+        let resources = build(&cluster, "simple-nifi-serviceaccount").expect("build succeeds");
+
+        // The minimal fixture has a single `default` role group for the `node` role.
+        assert_eq!(
+            sorted_names(&resources.stateful_sets),
+            ["simple-nifi-node-default"]
+        );
+        assert_eq!(
+            sorted_names(&resources.config_maps),
+            ["simple-nifi-node-default"]
+        );
+        // One headless and one metrics Service per role group.
+        assert_eq!(resources.services.len(), 2);
+        // One group Listener and one PDB for the single `node` role.
+        assert_eq!(sorted_names(&resources.listeners), ["simple-nifi-node"]);
+        assert_eq!(
+            sorted_names(&resources.pod_disruption_budgets),
+            ["simple-nifi-node"]
+        );
+    }
+}

rust/operator-binary/src/controller/build/properties.rs

@@ -81,7 +81,7 @@ pub(crate) mod test_support {
     use std::str::FromStr as _;
 
     use stackable_operator::{
-        commons::product_image_selection::ResolvedProductImage,
+        commons::{networking::DomainName, product_image_selection::ResolvedProductImage},
         crd::authentication::r#static::v1alpha1::{
             AuthenticationProvider as StaticAuthProvider, UserCredentialsSecretRef,
         },
@@ -168,6 +168,7 @@ pub(crate) mod test_support {
         ValidatedCluster::new(
             name,
             namespace,
+            DomainName::from_str("cluster.local").expect("valid cluster domain"),
             uid,
             image,
             product_version,

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

@@ -32,7 +32,7 @@ use stackable_operator::{
         self,
         framework::{create_vector_shutdown_file_command, remove_vector_shutdown_file_command},
     },
-    utils::{COMMON_BASH_TRAP_FUNCTIONS, cluster_info::KubernetesClusterInfo},
+    utils::COMMON_BASH_TRAP_FUNCTIONS,
     v2::{
         builder::pod::container::{EnvVarSet, new_container_builder},
         product_logging::framework::{
@@ -164,10 +164,8 @@ pub(crate) const LISTENER_DEFAULT_PORT_HTTPS_ENV: &str = "LISTENER_DEFAULT_PORT_
 ///
 /// The [`Pod`](`stackable_operator::k8s_openapi::api::core::v1::Pod`)s are accessible through the
 /// corresponding [`stackable_operator::k8s_openapi::api::core::v1::Service`] (from `build_rolegroup_headless_service`).
-#[allow(clippy::too_many_arguments)]
-pub(crate) async fn build_node_rolegroup_statefulset(
+pub(crate) fn build_node_rolegroup_statefulset(
     cluster: &ValidatedCluster,
-    cluster_info: &KubernetesClusterInfo,
     role_group_name: &RoleGroupName,
     rg: &NifiRoleGroupConfig,
     effective_replicas: Option<i32>,
@@ -251,7 +249,7 @@ pub(crate) async fn build_node_rolegroup_statefulset(
         "$POD_NAME.{service_name}.{namespace}.svc.{cluster_domain}",
         service_name = resource_names.headless_service_name(),
         namespace = cluster.namespace,
-        cluster_domain = cluster_info.cluster_domain,
+        cluster_domain = cluster.cluster_domain,
     );
 
     let sensitive_key_secret = &cluster.cluster_config.sensitive_properties.key_secret;

rust/operator-binary/src/controller/dereference.rs

@@ -6,6 +6,7 @@
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     client::Client,
+    commons::networking::DomainName,
     v2::{
         controller_utils::{self, get_namespace},
         types::kubernetes::NamespaceName,
@@ -38,6 +39,9 @@ type Result<T, E = Error> = std::result::Result<T, E>;
 pub struct DereferencedObjects {
     /// The namespace of the [`v1alpha1::NifiCluster`], parsed once here and reused everywhere.
     pub namespace: NamespaceName,
+    /// The Kubernetes cluster domain, captured from the client here so the build step needs no
+    /// client to assemble in-cluster DNS names.
+    pub cluster_domain: DomainName,
     pub authentication_classes: DereferencedAuthenticationClasses,
     pub authorization: DereferencedAuthorization,
 }
@@ -63,6 +67,7 @@ pub async fn dereference(
 
     Ok(DereferencedObjects {
         namespace,
+        cluster_domain: client.kubernetes_cluster_info.cluster_domain.clone(),
         authentication_classes,
         authorization,
     })

rust/operator-binary/src/controller/mod.rs

@@ -7,11 +7,19 @@ use std::{collections::BTreeMap, str::FromStr as _};
 use stackable_operator::{
     commons::{
         affinity::StackableAffinity,
+        networking::DomainName,
         product_image_selection::ResolvedProductImage,
         resources::{NoRuntimeLimits, Resources},
     },
-    crd::git_sync,
-    k8s_openapi::{api::core::v1::Volume, apimachinery::pkg::apis::meta::v1::ObjectMeta},
+    crd::{git_sync, listener},
+    k8s_openapi::{
+        api::{
+            apps::v1::StatefulSet,
+            core::v1::{ConfigMap, Service, Volume},
+            policy::v1::PodDisruptionBudget,
+        },
+        apimachinery::pkg::apis::meta::v1::ObjectMeta,
+    },
     kube::Resource,
     kvp::Labels,
     shared::time::Duration,
@@ -47,6 +55,15 @@ pub(crate) mod build;
 pub(crate) mod dereference;
 pub(crate) mod validate;
 
+/// Every Kubernetes resource produced by the [`build`] step.
+pub struct KubernetesResources {
+    pub stateful_sets: Vec<StatefulSet>,
+    pub services: Vec<Service>,
+    pub listeners: Vec<listener::v1alpha1::Listener>,
+    pub config_maps: Vec<ConfigMap>,
+    pub pod_disruption_budgets: Vec<PodDisruptionBudget>,
+}
+
 /// A validated, merged (default <- role <- role-group) NiFi rolegroup config.
 pub type NifiRoleGroupConfig =
     RoleGroupConfig<ValidatedNifiConfig, JavaCommonConfig, v1alpha1::NifiConfigOverrides>;
@@ -140,6 +157,9 @@ pub struct ValidatedCluster {
     pub name: ClusterName,
     /// The namespace of the NifiCluster, parsed once in the dereference step and reused everywhere.
     pub namespace: NamespaceName,
+    /// The Kubernetes cluster domain, captured from the client in the dereference step so the
+    /// build step needs no client to assemble in-cluster DNS names.
+    pub cluster_domain: DomainName,
     /// The UID of the NifiCluster, used to build OwnerReferences downstream.
     pub uid: Uid,
     /// The product image.
@@ -198,6 +218,7 @@ impl ValidatedCluster {
     pub fn new(
         name: ClusterName,
         namespace: NamespaceName,
+        cluster_domain: DomainName,
         uid: Uid,
         image: ResolvedProductImage,
         product_version: ProductVersion,
@@ -216,6 +237,7 @@ impl ValidatedCluster {
             metadata,
             name,
             namespace,
+            cluster_domain,
             uid,
             image,
             product_version,

rust/operator-binary/src/controller/validate.rs

@@ -156,6 +156,7 @@ pub fn validate(
         .context(NoNodesDefinedSnafu)?;
 
     let namespace = dereferenced_objects.namespace.clone();
+    let cluster_domain = dereferenced_objects.cluster_domain.clone();
     let uid = get_uid(nifi).context(GetUidSnafu)?;
 
     // `app_version_label_value` is constructed to be a valid label value, so it is always a valid
@@ -166,6 +167,7 @@ pub fn validate(
     Ok(ValidatedCluster::new(
         name,
         namespace,
+        cluster_domain,
         uid,
         image,
         product_version,