refactor: adapt pdb to build instead of mutate

Author: maltesander

rust/operator-binary/src/controller.rs

@@ -16,7 +16,7 @@ use const_format::concatcp;
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     cli::OperatorEnvironmentOptions,
-    cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
+    cluster_resources::ClusterResourceApplyStrategy,
     commons::{
         networking::DomainName, product_image_selection::ResolvedProductImage,
         rbac::build_rbac_resources,
@@ -38,6 +38,7 @@ use stackable_operator::{
     },
     v2::{
         HasName, HasUid, NameIsValidLabelValue,
+        cluster_resources::cluster_resources_new,
         kvp::label::{recommended_labels, role_group_selector},
         role_group_utils::ResourceNames,
         types::{
@@ -63,7 +64,7 @@ use crate::{
             properties::listener::get_kafka_listener_config,
             resource::{
                 listener::build_broker_rolegroup_bootstrap_listener,
-                pdb::add_pdbs,
+                pdb::build_pdb,
                 service::{build_rolegroup_headless_service, build_rolegroup_metrics_service},
                 statefulset::{
                     build_broker_rolegroup_statefulset, build_controller_rolegroup_statefulset,
@@ -314,17 +315,17 @@ impl NameIsValidLabelValue for ValidatedCluster {
 }
 
 /// The product name (`kafka`) as a type-safe label value.
-fn product_name() -> ProductName {
+pub(crate) fn product_name() -> ProductName {
     ProductName::from_str(APP_NAME).expect("'kafka' is a valid product name")
 }
 
 /// The operator name as a type-safe label value.
-fn operator_name() -> OperatorName {
+pub(crate) fn operator_name() -> OperatorName {
     OperatorName::from_str(OPERATOR_NAME).expect("the operator name is a valid label value")
 }
 
 /// The controller name as a type-safe label value.
-fn controller_name() -> ControllerName {
+pub(crate) fn controller_name() -> ControllerName {
     ControllerName::from_str(KAFKA_CONTROLLER_NAME)
         .expect("the controller name is a valid label value")
 }
@@ -461,11 +462,6 @@ pub enum Error {
         source: stackable_operator::cluster_resources::Error,
     },
 
-    #[snafu(display("failed to create cluster resources"))]
-    CreateClusterResources {
-        source: stackable_operator::cluster_resources::Error,
-    },
-
     #[snafu(display("failed to patch service account"))]
     ApplyServiceAccount {
         source: stackable_operator::cluster_resources::Error,
@@ -486,9 +482,9 @@ pub enum Error {
         source: stackable_operator::commons::rbac::Error,
     },
 
-    #[snafu(display("failed to create PodDisruptionBudget"))]
-    FailedToCreatePdb {
-        source: crate::controller::build::resource::pdb::Error,
+    #[snafu(display("failed to apply PodDisruptionBudget"))]
+    ApplyPdb {
+        source: stackable_operator::cluster_resources::Error,
     },
 
     #[snafu(display("failed to get required Labels"))]
@@ -530,12 +526,11 @@ impl ReconcilerError for Error {
             Error::BuildDiscoveryConfig { .. } => None,
             Error::ApplyDiscoveryConfig { .. } => None,
             Error::DeleteOrphans { .. } => None,
-            Error::CreateClusterResources { .. } => None,
             Error::ApplyServiceAccount { .. } => None,
             Error::ApplyRoleBinding { .. } => None,
             Error::ApplyStatus { .. } => None,
             Error::BuildRbacResources { .. } => None,
-            Error::FailedToCreatePdb { .. } => None,
+            Error::ApplyPdb { .. } => None,
             Error::GetRequiredLabels { .. } => None,
             Error::InvalidKafkaCluster { .. } => None,
             Error::BuildStatefulset { .. } => None,
@@ -568,15 +563,16 @@ pub async fn reconcile_kafka(
         validate::validate(kafka, dereferenced_objects, &ctx.operator_environment)
             .context(ValidateClusterSnafu)?;
 
-    let mut cluster_resources = ClusterResources::new(
-        APP_NAME,
-        OPERATOR_NAME,
-        KAFKA_CONTROLLER_NAME,
-        &validated_cluster.object_ref(&()),
+    let mut cluster_resources = cluster_resources_new(
+        &product_name(),
+        &operator_name(),
+        &controller_name(),
+        &validated_cluster.name,
+        &validated_cluster.namespace,
+        &validated_cluster.uid,
         ClusterResourceApplyStrategy::from(&kafka.spec.cluster_operation),
         &kafka.spec.object_overrides,
-    )
-    .context(CreateClusterResourcesSnafu)?;
+    );
 
     tracing::debug!(
         kerberos_enabled = validated_cluster.cluster_config.kafka_security.has_kerberos_enabled(),
@@ -718,10 +714,12 @@ pub async fn reconcile_kafka(
         if let Some(GenericRoleConfig {
             pod_disruption_budget: pdb,
         }) = role_cfg
+            && let Some(pdb) = build_pdb(pdb, &validated_cluster, kafka_role)
         {
-            add_pdbs(pdb, kafka, kafka_role, client, &mut cluster_resources)
+            cluster_resources
+                .add(client, pdb)
                 .await
-                .context(FailedToCreatePdbSnafu)?;
+                .context(ApplyPdbSnafu)?;
         }
     }
 

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

@@ -1,61 +1,37 @@
-use snafu::{ResultExt, Snafu};
 use stackable_operator::{
-    builder::pdb::PodDisruptionBudgetBuilder, client::Client, cluster_resources::ClusterResources,
-    commons::pdb::PdbConfig, kube::ResourceExt,
+    commons::pdb::PdbConfig, k8s_openapi::api::policy::v1::PodDisruptionBudget,
+    v2::builder::pdb::pod_disruption_budget_builder_with_role,
 };
 
 use crate::{
-    controller::KAFKA_CONTROLLER_NAME,
-    crd::{APP_NAME, OPERATOR_NAME, role::KafkaRole, v1alpha1},
+    controller::{ValidatedCluster, controller_name, operator_name, product_name},
+    crd::role::KafkaRole,
 };
 
-#[derive(Snafu, Debug)]
-pub enum Error {
-    #[snafu(display("Cannot create PodDisruptionBudget for role [{role}]"))]
-    CreatePdb {
-        source: stackable_operator::builder::pdb::Error,
-        role: String,
-    },
-    #[snafu(display("Cannot apply PodDisruptionBudget [{name}]"))]
-    ApplyPdb {
-        source: stackable_operator::cluster_resources::Error,
-        name: String,
-    },
-}
-
-pub async fn add_pdbs(
+/// Builds the [`PodDisruptionBudget`] for the given `role`, or `None` if PDBs are disabled.
+pub fn build_pdb(
     pdb: &PdbConfig,
-    kafka: &v1alpha1::KafkaCluster,
+    validated_cluster: &ValidatedCluster,
     role: &KafkaRole,
-    client: &Client,
-    cluster_resources: &mut ClusterResources<'_>,
-) -> Result<(), Error> {
+) -> Option<PodDisruptionBudget> {
     if !pdb.enabled {
-        return Ok(());
+        return None;
     }
     let max_unavailable = pdb.max_unavailable.unwrap_or(match role {
         KafkaRole::Broker => max_unavailable_brokers(),
         KafkaRole::Controller => max_unavailable_controllers(),
     });
-    let pdb = PodDisruptionBudgetBuilder::new_with_role(
-        kafka,
-        APP_NAME,
-        &role.to_string(),
-        OPERATOR_NAME,
-        KAFKA_CONTROLLER_NAME,
+    let pdb = pod_disruption_budget_builder_with_role(
+        validated_cluster,
+        &product_name(),
+        &ValidatedCluster::role_name(role),
+        &operator_name(),
+        &controller_name(),
     )
-    .with_context(|_| CreatePdbSnafu {
-        role: role.to_string(),
-    })?
     .with_max_unavailable(max_unavailable)
     .build();
-    let pdb_name = pdb.name_any();
-    cluster_resources
-        .add(client, pdb)
-        .await
-        .with_context(|_| ApplyPdbSnafu { name: pdb_name })?;
 
-    Ok(())
+    Some(pdb)
 }
 
 fn max_unavailable_brokers() -> u16 {

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

@@ -1,15 +1,12 @@
-use std::ops::Deref;
+use std::{ops::Deref, str::FromStr};
 
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
     builder::{
         meta::ObjectMetaBuilder,
         pod::{
-            PodBuilder,
-            container::ContainerBuilder,
-            resources::ResourceRequirementsBuilder,
-            security::PodSecurityContextBuilder,
-            volume::{ListenerOperatorVolumeSourceBuilder, ListenerReference, VolumeBuilder},
+            PodBuilder, container::ContainerBuilder, resources::ResourceRequirementsBuilder,
+            security::PodSecurityContextBuilder, volume::VolumeBuilder,
         },
     },
     commons::product_image_selection::ResolvedProductImage,
@@ -35,8 +32,13 @@ use stackable_operator::{
         },
     },
     v2::{
-        builder::meta::ownerreference_from_resource, jvm_argument_overrides::JvmArgumentOverrides,
+        builder::{
+            meta::ownerreference_from_resource,
+            pod::volume::{ListenerReference, listener_operator_volume_source_builder_build_pvc},
+        },
+        jvm_argument_overrides::JvmArgumentOverrides,
         role_group_utils::ResourceNames,
+        types::kubernetes::{ListenerName, PersistentVolumeClaimName},
     },
 };
 
@@ -97,11 +99,6 @@ pub enum Error {
         source: stackable_operator::builder::pod::Error,
     },
 
-    #[snafu(display("failed to build bootstrap listener pvc"))]
-    BuildBootstrapListenerPvc {
-        source: stackable_operator::builder::pod::volume::ListenerOperatorVolumeSourceBuilderError,
-    },
-
     #[snafu(display("failed to build pod descriptors"))]
     BuildPodDescriptors {
         source: crate::controller::PodDescriptorsError,
@@ -191,16 +188,17 @@ pub fn build_broker_rolegroup_statefulset(
 
     // bootstrap listener should be persistent,
     // main broker listener is an ephemeral PVC instead
-    pvcs.push(
-        ListenerOperatorVolumeSourceBuilder::new(
-            &ListenerReference::ListenerName(
-                validated_cluster.bootstrap_listener_name(kafka_role, role_group_name),
-            ),
-            &unversioned_recommended_labels,
-        )
-        .build_pvc(LISTENER_BOOTSTRAP_VOLUME_NAME)
-        .context(BuildBootstrapListenerPvcSnafu)?,
-    );
+    let bootstrap_listener_name = ListenerName::from_str(
+        &validated_cluster.bootstrap_listener_name(kafka_role, role_group_name),
+    )
+    .expect("the bootstrap listener name is a valid Listener name");
+    let bootstrap_pvc_name = PersistentVolumeClaimName::from_str(LISTENER_BOOTSTRAP_VOLUME_NAME)
+        .expect("the bootstrap listener volume name is a valid PVC name");
+    pvcs.push(listener_operator_volume_source_builder_build_pvc(
+        &ListenerReference::Listener(bootstrap_listener_name),
+        &unversioned_recommended_labels,
+        &bootstrap_pvc_name,
+    ));
 
     if kafka_security.has_kerberos_enabled() {
         add_kerberos_pod_config(