Check for ownerReferences in ClusterResources (#862)

* Check for `ownerReferences` in `ClusterResources`

Fixes #861

* Changelog

Author: nightkr

crates/stackable-operator/CHANGELOG.md

@@ -17,11 +17,13 @@ All notable changes to this project will be documented in this file.
 
 - Fix the CRD description of `ClientAuthenticationDetails` to not contain internal Rust doc, but a public CRD description ([#846]).
 - `StackableAffinity` fields are no longer erroneously marked as required ([#855]).
+- BREAKING: `ClusterResources` will now only consider deleting objects that are marked as directly owned (via `.metadata.ownerReferences`) ([#862]).
 
 [#846]: https://github.com/stackabletech/operator-rs/pull/846
 [#851]: https://github.com/stackabletech/operator-rs/pull/851
 [#855]: https://github.com/stackabletech/operator-rs/pull/855
 [#858]: https://github.com/stackabletech/operator-rs/pull/858
+[#862]: https://github.com/stackabletech/operator-rs/pull/862
 
 ## [0.74.0] - 2024-08-22
 

crates/stackable-operator/src/cluster_resources.rs

@@ -388,6 +388,9 @@ pub struct ClusterResources {
     /// The name of the application
     app_name: String,
 
+    /// The uid of the cluster object
+    cluster_uid: String,
+
     // TODO (Techassi): Add doc comments
     operator_name: String,
 
@@ -434,17 +437,22 @@ impl ClusterResources {
     ) -> Result<Self> {
         let namespace = cluster
             .namespace
-            .to_owned()
+            .clone()
             .context(MissingObjectKeySnafu { key: "namespace" })?;
         let app_instance = cluster
             .name
-            .to_owned()
+            .clone()
             .context(MissingObjectKeySnafu { key: "name" })?;
+        let cluster_uid = cluster
+            .uid
+            .clone()
+            .context(MissingObjectKeySnafu { key: "uid" })?;
 
         Ok(ClusterResources {
             namespace,
             app_instance,
             app_name: app_name.into(),
+            cluster_uid,
             operator_name: operator_name.into(),
             controller_name: controller_name.into(),
             manager: format_full_controller_name(operator_name, controller_name),
@@ -741,10 +749,22 @@ impl ClusterResources {
             ..Default::default()
         };
 
-        let resources = client
+        let mut resources = client
             .list_with_label_selector::<T>(&self.namespace, &label_selector)
             .await?;
 
+        // filter out objects without a direct ownership relationship, for example:
+        // - indirect ownership where the labels are still propagated
+        // - objects owned by versions of the cluster recreated before/after the current snapshot
+        resources.retain(|resource| {
+            resource
+                .meta()
+                .owner_references
+                .iter()
+                .flatten()
+                .any(|reference| reference.uid == self.cluster_uid)
+        });
+
         Ok(resources)
     }
 }