use v2 cluster_resources_new

Author: adwk67

rust/operator-binary/src/airflow_controller.rs

@@ -13,7 +13,7 @@ use stackable_operator::{
     crd::git_sync,
     k8s_openapi::api::core::v1::EnvVar,
     kube::{
-        Resource, ResourceExt,
+        ResourceExt,
         core::{DeserializeGuard, error_boundary},
         runtime::controller::Action,
     },
@@ -24,7 +24,10 @@ use stackable_operator::{
         compute_conditions, operations::ClusterOperationsConditionBuilder,
         statefulset::StatefulSetConditionBuilder,
     },
-    v2::types::operator::{RoleGroupName, RoleName},
+    v2::{
+        cluster_resources::cluster_resources_new,
+        types::operator::{RoleGroupName, RoleName},
+    },
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
@@ -42,6 +45,7 @@ use crate::{
             },
             volumes::LOG_VOLUME_NAME,
         },
+        controller_name, operator_name, product_name,
     },
     crd::{
         APP_NAME, AirflowClusterStatus, AirflowConfigOverrides, Container, OPERATOR_NAME,
@@ -134,11 +138,6 @@ pub enum Error {
     #[snafu(display("invalid git-sync specification"))]
     InvalidGitSyncSpec { source: git_sync::v1alpha2::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,
@@ -239,7 +238,7 @@ pub async fn reconcile_airflow(
         validated_cluster.internal_secret_name().as_ref(),
         INTERNAL_SECRET_SECRET_KEY,
         256,
-        airflow,
+        &validated_cluster,
         client,
     )
     .await
@@ -249,7 +248,7 @@ pub async fn reconcile_airflow(
         validated_cluster.jwt_secret_name().as_ref(),
         JWT_SECRET_SECRET_KEY,
         256,
-        airflow,
+        &validated_cluster,
         client,
     )
     .await
@@ -263,21 +262,22 @@ pub async fn reconcile_airflow(
         validated_cluster.fernet_key_name().as_ref(),
         FERNET_KEY_SECRET_KEY,
         32,
-        airflow,
+        &validated_cluster,
         client,
     )
     .await
     .context(InternalSecretSnafu)?;
 
-    let mut cluster_resources = ClusterResources::new(
-        APP_NAME,
-        OPERATOR_NAME,
-        AIRFLOW_CONTROLLER_NAME,
-        &airflow.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(&airflow.spec.cluster_operation),
         &airflow.spec.object_overrides,
-    )
-    .context(CreateClusterResourcesSnafu)?;
+    );
 
     let required_labels = cluster_resources
         .get_required_labels()

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

@@ -400,17 +400,17 @@ impl ValidatedCluster {
 }
 
 /// The product name (`airflow`) as a type-safe label value.
-fn product_name() -> ProductName {
+pub(crate) fn product_name() -> ProductName {
     ProductName::from_str(APP_NAME).expect("'airflow' 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(AIRFLOW_CONTROLLER_NAME)
         .expect("the controller name is a valid label value")
 }

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

@@ -169,18 +169,18 @@ pub fn validate_cluster(
     } = dereferenced;
 
     // The Celery result-backend and broker only work with configured celeryExecutors. Emit a
-    // warning if they are provided without a Celery executor. The connection details themselves are
+    // warning if either are provided without a Celery executor. The connection details themselves are
     // derived from the CRD later, in the build step (see
     // `ValidatedCluster::celery_database_connection_details`).
-    if airflow.spec.cluster_config.celery_results_backend.is_some()
-        && airflow.spec.cluster_config.celery_broker.is_some()
+    if (airflow.spec.cluster_config.celery_results_backend.is_some()
+        || airflow.spec.cluster_config.celery_broker.is_some())
         && !matches!(
             &airflow.spec.executor,
             AirflowExecutor::CeleryExecutors { .. }
         )
     {
         tracing::warn!(
-            "No `spec.celeryExecutors` configured, but `spec.clusterConfig.celeryResultsBackend` and `spec.clusterConfig.celeryBroker` are provided. This only works in combination with a celery executor!"
+            "No `spec.celeryExecutors` configured, but `spec.clusterConfig.celeryResultsBackend` and/or `spec.clusterConfig.celeryBroker` are provided. This only works in combination with a celery executor!"
         )
     }