Merge remote-tracking branch 'origin/main' into op-rs-0.79.0

Author: sbernauer

CHANGELOG.md

@@ -18,6 +18,7 @@
 
 - Implement `envOverrides` for HbaseCluster ([#550]).
 - Omid test: use 1.1.2, update default port number and raise test timeout ([#556]).
+- An invalid `HBaseCluster` doesn't cause the operator to stop functioning (#[575]).
 
 ### Removed
 

rust/operator-binary/src/hbase_controller.rs

@@ -39,6 +39,7 @@ use stackable_operator::{
         apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString},
         DeepMerge,
     },
+    kube::core::{error_boundary, DeserializeGuard},
     kube::{runtime::controller::Action, Resource},
     kvp::{Label, LabelError, Labels, ObjectLabels},
     logging::controller::ReconcilerError,
@@ -305,6 +306,11 @@ pub enum Error {
     AddVolumeMount {
         source: builder::pod::container::Error,
     },
+
+    #[snafu(display("HBaseCluster object is invalid"))]
+    InvalidHBaseCluster {
+        source: error_boundary::InvalidObject,
+    },
 }
 
 type Result<T, E = Error> = std::result::Result<T, E>;
@@ -315,31 +321,39 @@ impl ReconcilerError for Error {
     }
 }
 
-pub async fn reconcile_hbase(hbase: Arc<HbaseCluster>, ctx: Arc<Ctx>) -> Result<Action> {
+pub async fn reconcile_hbase(
+    hbase: Arc<DeserializeGuard<HbaseCluster>>,
+    ctx: Arc<Ctx>,
+) -> Result<Action> {
     tracing::info!("Starting reconcile");
 
+    let hbase = hbase
+        .0
+        .as_ref()
+        .map_err(error_boundary::InvalidObject::clone)
+        .context(InvalidHBaseClusterSnafu)?;
+
     let client = &ctx.client;
 
-    validate_cr(&hbase)?;
+    validate_cr(hbase)?;
 
     let resolved_product_image = hbase
         .spec
         .image
         .resolve(DOCKER_IMAGE_BASE_NAME, crate::built_info::PKG_VERSION);
-    let zookeeper_connection_information = ZookeeperConnectionInformation::retrieve(&hbase, client)
+    let zookeeper_connection_information = ZookeeperConnectionInformation::retrieve(hbase, client)
         .await
         .context(RetrieveZookeeperConnectionInformationSnafu)?;
 
-    let vector_aggregator_address = resolve_vector_aggregator_address(&hbase, client)
+    let vector_aggregator_address = resolve_vector_aggregator_address(hbase, client)
         .await
         .context(ResolveVectorAggregatorAddressSnafu)?;
 
-    let roles = build_roles(&hbase)?;
+    let roles = build_roles(hbase)?;
 
     let validated_config = validate_all_roles_and_groups_config(
         &resolved_product_image.app_version_label,
-        &transform_all_roles_to_config(hbase.as_ref(), roles)
-            .context(GenerateProductConfigSnafu)?,
+        &transform_all_roles_to_config(hbase, roles).context(GenerateProductConfigSnafu)?,
         &ctx.product_config,
         false,
         false,
@@ -348,7 +362,7 @@ pub async fn reconcile_hbase(hbase: Arc<HbaseCluster>, ctx: Arc<Ctx>) -> Result<
 
     let hbase_opa_config = match &hbase.spec.cluster_config.authorization {
         Some(opa_config) => Some(
-            HbaseOpaConfig::from_opa_config(client, &hbase, opa_config)
+            HbaseOpaConfig::from_opa_config(client, hbase, opa_config)
                 .await
                 .context(InvalidOpaConfigSnafu)?,
         ),
@@ -365,15 +379,15 @@ pub async fn reconcile_hbase(hbase: Arc<HbaseCluster>, ctx: Arc<Ctx>) -> Result<
     .context(CreateClusterResourcesSnafu)?;
 
     let region_server_role_service =
-        build_region_server_role_service(&hbase, &resolved_product_image)?;
+        build_region_server_role_service(hbase, &resolved_product_image)?;
     cluster_resources
         .add(client, region_server_role_service)
         .await
         .context(ApplyRoleServiceSnafu)?;
 
     // discovery config map
     let discovery_cm = build_discovery_configmap(
-        &hbase,
+        hbase,
         &client.kubernetes_cluster_info,
         &zookeeper_connection_information,
         &resolved_product_image,
@@ -385,7 +399,7 @@ pub async fn reconcile_hbase(hbase: Arc<HbaseCluster>, ctx: Arc<Ctx>) -> Result<
         .context(ApplyDiscoveryConfigMapSnafu)?;
 
     let (rbac_sa, rbac_rolebinding) = build_rbac_resources(
-        hbase.as_ref(),
+        hbase,
         APP_NAME,
         cluster_resources
             .get_required_labels()
@@ -419,9 +433,9 @@ pub async fn reconcile_hbase(hbase: Arc<HbaseCluster>, ctx: Arc<Ctx>) -> Result<
                 .context(FailedToResolveConfigSnafu)?;
 
             let rg_service =
-                build_rolegroup_service(&hbase, &hbase_role, &rolegroup, &resolved_product_image)?;
+                build_rolegroup_service(hbase, &hbase_role, &rolegroup, &resolved_product_image)?;
             let rg_configmap = build_rolegroup_config_map(
-                &hbase,
+                hbase,
                 &client.kubernetes_cluster_info,
                 &rolegroup,
                 rolegroup_config,
@@ -432,7 +446,7 @@ pub async fn reconcile_hbase(hbase: Arc<HbaseCluster>, ctx: Arc<Ctx>) -> Result<
                 vector_aggregator_address.as_deref(),
             )?;
             let rg_statefulset = build_rolegroup_statefulset(
-                &hbase,
+                hbase,
                 &hbase_role,
                 &rolegroup,
                 rolegroup_config,
@@ -466,7 +480,7 @@ pub async fn reconcile_hbase(hbase: Arc<HbaseCluster>, ctx: Arc<Ctx>) -> Result<
             pod_disruption_budget: pdb,
         }) = role_config
         {
-            add_pdbs(pdb, &hbase, &hbase_role, client, &mut cluster_resources)
+            add_pdbs(pdb, hbase, &hbase_role, client, &mut cluster_resources)
                 .await
                 .context(FailedToCreatePdbSnafu)?;
         }
@@ -476,18 +490,15 @@ pub async fn reconcile_hbase(hbase: Arc<HbaseCluster>, ctx: Arc<Ctx>) -> Result<
         ClusterOperationsConditionBuilder::new(&hbase.spec.cluster_operation);
 
     let status = HbaseClusterStatus {
-        conditions: compute_conditions(
-            hbase.as_ref(),
-            &[&ss_cond_builder, &cluster_operation_cond_builder],
-        ),
+        conditions: compute_conditions(hbase, &[&ss_cond_builder, &cluster_operation_cond_builder]),
     };
 
     cluster_resources
         .delete_orphaned_resources(client)
         .await
         .context(DeleteOrphanedResourcesSnafu)?;
     client
-        .apply_patch_status(OPERATOR_NAME, hbase.as_ref(), &status)
+        .apply_patch_status(OPERATOR_NAME, hbase, &status)
         .await
         .context(ApplyStatusSnafu)?;
 
@@ -1104,8 +1115,16 @@ where
     })
 }
 
-pub fn error_policy(_obj: Arc<HbaseCluster>, _error: &Error, _ctx: Arc<Ctx>) -> Action {
-    Action::requeue(*Duration::from_secs(5))
+pub fn error_policy(
+    _obj: Arc<DeserializeGuard<HbaseCluster>>,
+    error: &Error,
+    _ctx: Arc<Ctx>,
+) -> Action {
+    match error {
+        // root object is invalid, will be requed when modified
+        Error::InvalidHBaseCluster { .. } => Action::await_change(),
+        _ => Action::requeue(*Duration::from_secs(5)),
+    }
 }
 
 pub fn build_recommended_labels<'a>(

rust/operator-binary/src/main.rs

@@ -14,6 +14,7 @@ use stackable_hbase_crd::{HbaseCluster, APP_NAME};
 use stackable_operator::{
     cli::{Command, ProductOperatorRun},
     k8s_openapi::api::{apps::v1::StatefulSet, core::v1::Service},
+    kube::core::DeserializeGuard,
     kube::runtime::{controller::Controller, watcher},
     logging::controller::report_controller_reconciled,
     CustomResourceExt,
@@ -70,7 +71,7 @@ async fn main() -> anyhow::Result<()> {
             .await?;
 
             Controller::new(
-                watch_namespace.get_api::<HbaseCluster>(&client),
+                watch_namespace.get_api::<DeserializeGuard<HbaseCluster>>(&client),
                 watcher::Config::default(),
             )
             .owns(