refactor: move to v2 pdb and clusteresources

Author: maltesander

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

@@ -1,54 +1,55 @@
-use std::cmp::max;
+use std::{cmp::max, str::FromStr};
 
-use snafu::{ResultExt, Snafu};
 use stackable_operator::{
-    builder::pdb::PodDisruptionBudgetBuilder, commons::pdb::PdbConfig,
+    commons::pdb::PdbConfig,
     k8s_openapi::api::policy::v1::PodDisruptionBudget,
+    v2::{builder::pdb::pod_disruption_budget_builder_with_role, types::operator::RoleName},
 };
 
 use crate::{
-    crd::{APP_NAME, TrinoRole, v1alpha1},
-    trino_controller::{CONTROLLER_NAME, OPERATOR_NAME},
+    controller::{ValidatedCluster, controller_name, operator_name, product_name},
+    crd::TrinoRole,
 };
 
-#[derive(Snafu, Debug)]
-pub enum Error {
-    #[snafu(display("cannot create PodDisruptionBudget for role [{role}]"))]
-    CreatePdb {
-        source: stackable_operator::builder::pdb::Error,
-        role: String,
-    },
-}
-
 /// Builds the [`PodDisruptionBudget`] for the given `role`, or `None` if PDBs are disabled.
 ///
 /// The reconciler applies the returned object; this function does not touch the cluster.
 pub fn build_pdb(
     pdb: &PdbConfig,
-    trino: &v1alpha1::TrinoCluster,
+    cluster: &ValidatedCluster,
     role: &TrinoRole,
-) -> Result<Option<PodDisruptionBudget>, Error> {
+) -> Option<PodDisruptionBudget> {
     if !pdb.enabled {
-        return Ok(None);
+        return None;
     }
     let max_unavailable = pdb.max_unavailable.unwrap_or(match role {
         TrinoRole::Coordinator => max_unavailable_coordinators(),
-        TrinoRole::Worker => max_unavailable_workers(trino.num_workers()),
+        TrinoRole::Worker => max_unavailable_workers(worker_count(cluster)),
     });
-    let pdb = PodDisruptionBudgetBuilder::new_with_role(
-        trino,
-        APP_NAME,
-        &role.to_string(),
-        OPERATOR_NAME,
-        CONTROLLER_NAME,
+    let role_name =
+        RoleName::from_str(&role.to_string()).expect("a TrinoRole is a valid RFC 1123 role name");
+    let pdb = pod_disruption_budget_builder_with_role(
+        cluster,
+        &product_name(),
+        &role_name,
+        &operator_name(),
+        &controller_name(),
     )
-    .with_context(|_| CreatePdbSnafu {
-        role: role.to_string(),
-    })?
     .with_max_unavailable(max_unavailable)
     .build();
 
-    Ok(Some(pdb))
+    Some(pdb)
+}
+
+/// Total number of worker replicas across all worker role groups.
+fn worker_count(cluster: &ValidatedCluster) -> u16 {
+    cluster
+        .role_group_configs
+        .get(&TrinoRole::Worker)
+        .into_iter()
+        .flat_map(|groups| groups.values())
+        .map(|rg| rg.replicas)
+        .sum()
 }
 
 fn max_unavailable_coordinators() -> u16 {

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

@@ -12,10 +12,14 @@ use stackable_operator::{
     kube::{Resource, api::ObjectMeta},
     shared::time::Duration,
     v2::{
+        HasName, HasUid, NameIsValidLabelValue,
         role_group_utils::ResourceNames,
         types::{
             kubernetes::{NamespaceName, Uid},
-            operator::{ClusterName, RoleGroupName as RoleGroupNameV2, RoleName},
+            operator::{
+                ClusterName, ControllerName, OperatorName, ProductName,
+                RoleGroupName as RoleGroupNameV2, RoleName,
+            },
         },
     },
 };
@@ -29,9 +33,10 @@ use crate::{
         fault_tolerant_execution::ResolvedFaultTolerantExecutionConfig,
     },
     crd::{
-        HTTP_PORT, HTTP_PORT_NAME, HTTPS_PORT, HTTPS_PORT_NAME, TrinoRole, discovery::TrinoPodRef,
-        v1alpha1,
+        APP_NAME, HTTP_PORT, HTTP_PORT_NAME, HTTPS_PORT, HTTPS_PORT_NAME, TrinoRole,
+        discovery::TrinoPodRef, v1alpha1,
     },
+    trino_controller::{CONTROLLER_NAME, OPERATOR_NAME},
 };
 
 pub(crate) mod build;
@@ -204,6 +209,39 @@ impl Resource for ValidatedCluster {
     }
 }
 
+impl HasName for ValidatedCluster {
+    fn to_name(&self) -> String {
+        self.name.to_string()
+    }
+}
+
+impl HasUid for ValidatedCluster {
+    fn to_uid(&self) -> Uid {
+        self.uid.clone()
+    }
+}
+
+impl NameIsValidLabelValue for ValidatedCluster {
+    fn to_label_value(&self) -> String {
+        self.name.to_label_value()
+    }
+}
+
+/// The product name (`trino`) as a type-safe label value.
+pub(crate) fn product_name() -> ProductName {
+    ProductName::from_str(APP_NAME).expect("'trino' is a valid product name")
+}
+
+/// The operator name as a type-safe label value.
+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.
+pub(crate) fn controller_name() -> ControllerName {
+    ControllerName::from_str(CONTROLLER_NAME).expect("the controller name is a valid label value")
+}
+
 /// A minimal, valid TrinoCluster spec shared across unit tests.
 #[cfg(test)]
 pub(crate) const MINIMAL_TRINO_YAML: &str = r#"

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

@@ -546,15 +546,6 @@ impl v1alpha1::TrinoCluster {
         }
     }
 
-    pub fn num_workers(&self) -> u16 {
-        self.spec
-            .workers
-            .iter()
-            .flat_map(|w| w.role_groups.values())
-            .map(|rg| rg.replicas.unwrap_or(1))
-            .sum()
-    }
-
     /// List all coordinator pods expected to form the cluster
     ///
     /// We try to predict the pods here rather than looking at the current cluster state in order to

rust/operator-binary/src/trino_controller.rs

@@ -5,10 +5,10 @@ use const_format::concatcp;
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     cli::OperatorEnvironmentOptions,
-    cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
+    cluster_resources::ClusterResourceApplyStrategy,
     commons::{random_secret_creation, rbac::build_rbac_resources},
     kube::{
-        Resource, ResourceExt,
+        ResourceExt,
         core::{DeserializeGuard, error_boundary},
         runtime::controller::Action,
     },
@@ -21,6 +21,7 @@ use stackable_operator::{
         compute_conditions, operations::ClusterOperationsConditionBuilder,
         statefulset::StatefulSetConditionBuilder,
     },
+    v2::cluster_resources::cluster_resources_new,
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
@@ -35,7 +36,7 @@ use crate::{
                 headless_service_ports,
             },
         },
-        dereference, validate,
+        controller_name, dereference, operator_name, product_name, validate,
     },
     crd::{APP_NAME, ENV_INTERNAL_SECRET, ENV_SPOOLING_SECRET, v1alpha1},
 };
@@ -63,11 +64,6 @@ pub(crate) const CONTAINER_IMAGE_BASE_NAME: &str = "trino";
 #[strum_discriminants(derive(IntoStaticStr))]
 #[allow(clippy::enum_variant_names)]
 pub enum Error {
-    #[snafu(display("failed to create cluster resources"))]
-    CreateClusterResources {
-        source: stackable_operator::cluster_resources::Error,
-    },
-
     #[snafu(display("failed to delete orphaned resources"))]
     DeleteOrphanedResources {
         source: stackable_operator::cluster_resources::Error,
@@ -123,9 +119,6 @@ pub enum Error {
         source: stackable_operator::commons::rbac::Error,
     },
 
-    #[snafu(display("failed to create PodDisruptionBudget"))]
-    FailedToCreatePdb { source: build::resource::pdb::Error },
-
     #[snafu(display("failed to apply PodDisruptionBudget"))]
     ApplyPdb {
         source: stackable_operator::cluster_resources::Error,
@@ -211,15 +204,16 @@ pub async fn reconcile_trino(
         "Validated TrinoCluster"
     );
 
-    let mut cluster_resources = ClusterResources::new(
-        APP_NAME,
-        OPERATOR_NAME,
-        CONTROLLER_NAME,
-        &trino.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(&trino.spec.cluster_operation),
         &trino.spec.object_overrides,
-    )
-    .context(CreateClusterResourcesSnafu)?;
+    );
 
     let (rbac_sa, rbac_rolebinding) = build_rbac_resources(
         trino,
@@ -407,7 +401,7 @@ pub async fn reconcile_trino(
         if let Some(GenericRoleConfig {
             pod_disruption_budget: pdb,
         }) = role_config
-            && let Some(pdb) = build_pdb(pdb, trino, trino_role).context(FailedToCreatePdbSnafu)?
+            && let Some(pdb) = build_pdb(pdb, &validated_cluster, trino_role)
         {
             cluster_resources
                 .add(client, pdb)