fix: Add missing owns/watches and test early-exit reconcile

CHANGELOG.md

@@ -31,13 +31,16 @@ All notable changes to this project will be documented in this file.
 - Fix a longstanding problem of including empty `categories`, `shortNames` and `additionalPrinterColumns` in the CRDs,
   which could cause problems with GitOps tools (e.g. ArgoCD) reporting a diff in the custom resources.
   See [our internal issue](https://github.com/stackabletech/hdfs-operator/issues/626) and [the fix](https://github.com/kube-rs/kube/pull/2042) for details ([#860]).
+- The operator now watches all resources that it creates and early-exits the reconcile action when the
+  cluster is marked for deletion ([#867]).
 
 [#841]: https://github.com/stackabletech/druid-operator/pull/841
 [#846]: https://github.com/stackabletech/druid-operator/pull/846
 [#855]: https://github.com/stackabletech/druid-operator/pull/855
 [#856]: https://github.com/stackabletech/druid-operator/pull/856
 [#860]: https://github.com/stackabletech/druid-operator/pull/860
 [#865]: https://github.com/stackabletech/druid-operator/pull/865
+[#867]: https://github.com/stackabletech/druid-operator/pull/867
 
 ## [26.7.0] - 2026-07-21
 

Cargo.toml

@@ -18,6 +18,7 @@ clap = "4.6"
 const_format = "0.2"
 fnv = "1.0"
 futures = { version = "0.3", features = ["compat"] }
+http = "1.3"
 indoc = "2.0"
 openssl = "0.10"
 pin-project = "1.1"

deploy/helm/druid-operator/templates/clusterrole-operator.yaml

@@ -29,7 +29,8 @@ rules:
       - patch
       - watch
   # Shared internal authentication secret (cookie passphrase and internal client password).
-  # Orphan cleanup not needed (instead, Kubernetes GC via owner reference).
+  # Orphan cleanup not needed (instead, Kubernetes GC via owner reference). Watched by the
+  # controller, which needs `list` as well: `.owns()` uses lists before it watches.
   - apiGroups:
       - ""
     resources:
@@ -38,9 +39,11 @@ rules:
       - create
       - delete
       - get
+      - list
       - patch
+      - watch
   # ServiceAccount created per DruidCluster for workload pod identity.
-  # Applied via SSA and tracked for orphan cleanup. Not watched by the controller.
+  # Applied via SSA and tracked for orphan cleanup and watched by the controller.
   - apiGroups:
       - ""
     resources:
@@ -51,8 +54,9 @@ rules:
       - get
       - list
       - patch
+      - watch
   # RoleBinding created per DruidCluster to bind the product ClusterRole to the workload
-  # ServiceAccount. Applied via SSA and tracked for orphan cleanup. Not watched by the controller.
+  # ServiceAccount. Applied via SSA and tracked for orphan cleanup and watched by the controller.
   - apiGroups:
       - rbac.authorization.k8s.io
     resources:
@@ -63,6 +67,7 @@ rules:
       - get
       - list
       - patch
+      - watch
   # Required to bind the product ClusterRole to the per-cluster ServiceAccount.
   - apiGroups:
       - rbac.authorization.k8s.io
@@ -85,8 +90,7 @@ rules:
       - list
       - patch
       - watch
-  # PodDisruptionBudget created per role. Applied via SSA and tracked for orphan cleanup.
-  # Not watched by the controller.
+  # PodDisruptionBudget created per role. Applied via SSA and tracked for orphan cleanup and watched by the controller.
   - apiGroups:
       - policy
     resources:
@@ -97,6 +101,7 @@ rules:
       - get
       - list
       - patch
+      - watch
   # Required for maintaining the CRDs within the operator (including the conversion webhook info).
   # Also for the startup condition check before the controller can run.
   - apiGroups:

rust/operator-binary/Cargo.toml

@@ -32,6 +32,7 @@ uuid.workspace = true
 built.workspace = true
 
 [dev-dependencies]
+http.workspace = true
 rstest.workspace = true
 serde_yaml.workspace = true
 stackable-operator = { workspace = true, features = ["test-support"] }

rust/operator-binary/src/controller.rs

@@ -17,6 +17,7 @@ use stackable_operator::{
         rbac::v1::RoleBinding,
     },
     kube::{
+        Resource,
         core::{DeserializeGuard, error_boundary},
         runtime::controller::Action,
     },
@@ -115,6 +116,11 @@ pub async fn reconcile_druid(
     ctx: Arc<Ctx>,
 ) -> Result<Action> {
     tracing::info!("Starting reconcile");
+
+    if druid.meta().deletion_timestamp.is_some() {
+        return Ok(Action::await_change());
+    }
+
     let druid = druid
         .0
         .as_ref()
@@ -173,9 +179,15 @@ mod test {
     use std::str::FromStr;
 
     use rstest::*;
-    use stackable_operator::v2::types::operator::RoleGroupName;
+    use stackable_operator::{
+        client::Client,
+        commons::networking::DomainName,
+        kube::{Client as KubeClient, Config, runtime::controller::Action},
+        utils::cluster_info::KubernetesClusterInfo,
+        v2::types::operator::RoleGroupName,
+    };
 
-    use super::{CONTROLLER_NAME, OPERATOR_NAME, PRODUCT_NAME};
+    use super::{CONTROLLER_NAME, OPERATOR_NAME, PRODUCT_NAME, *};
     use crate::{
         controller::build::{
             properties::ConfigFileName, resource::config_map::build_rolegroup_config_map,
@@ -254,4 +266,142 @@ mod test {
             "role group {tested_rolegroup_name}"
         );
     }
+
+    /// A [`Ctx`] whose client points at a closed port. Any API call made through it fails the
+    /// reconciliation, so an `Ok` result proves the reconciler returned before touching the
+    /// Kubernetes API.
+    fn unreachable_ctx() -> Arc<Ctx> {
+        let config = Config::new(
+            "http://127.0.0.1:1"
+                .parse::<http::Uri>()
+                .expect("valid static URI"),
+        );
+        let kube_client = KubeClient::try_from(config).expect("client from static config");
+
+        Arc::new(Ctx {
+            client: Client::new(
+                kube_client,
+                None,
+                "default".to_owned(),
+                KubernetesClusterInfo {
+                    cluster_domain: DomainName::from_str("cluster.local")
+                        .expect("valid cluster domain"),
+                },
+            ),
+            operator_environment: OperatorEnvironmentOptions {
+                operator_namespace: "stackable-operators".to_owned(),
+                operator_service_name: "druid-operator".to_owned(),
+                image_repository: "oci.stackable.tech/sdp".to_owned(),
+            },
+        })
+    }
+
+    fn reconcile(druid: DeserializeGuard<v1alpha1::DruidCluster>) -> Result<Action> {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread tokio runtime")
+            .block_on(async { reconcile_druid(Arc::new(druid), unreachable_ctx()).await })
+    }
+
+    #[test]
+    fn reconcile_exits_early_for_deleted_cluster() {
+        let druid = serde_yaml::from_str(
+            r#"
+apiVersion: druid.stackable.tech/v1alpha1
+kind: DruidCluster
+metadata:
+  name: druid
+  namespace: default
+  deletionTimestamp: "2026-08-14T12:00:00Z"
+spec:
+  image:
+    productVersion: 3.2.2
+"#,
+        )
+        .expect("valid cluster YAML");
+
+        let action = reconcile(druid).expect("a deleted cluster reconciles without any API call");
+
+        assert_eq!(action, Action::await_change());
+    }
+
+    #[test]
+    fn reconcile_exits_early_for_deleted_cluster_with_invalid_spec() {
+        let druid = serde_yaml::from_str(
+            r#"
+apiVersion: druid.stackable.tech/v1alpha1
+kind: DruidCluster
+metadata:
+  name: druid
+  namespace: default
+  deletionTimestamp: "2026-08-14T12:00:00Z"
+spec: {}
+"#,
+        )
+        .expect("YAML parses; the invalid spec is captured inside the DeserializeGuard");
+
+        let action =
+            reconcile(druid).expect("a deleted cluster reconciles even when its spec is invalid");
+
+        assert_eq!(action, Action::await_change());
+    }
+
+    #[test]
+    fn reconcile_proceeds_for_live_cluster() {
+        // Without a deletion timestamp the reconciler must not exit early.
+        // `validate` resolves the uid, so the fixture needs one. The probe for
+        // "reached the API" is then the random Secret creation rather than the
+        // dereference step: dereference only contacts the API when for
+        // optional objects, whereas the random Secrets are always created.
+        let druid = serde_yaml::from_str(
+            r#"
+apiVersion: druid.stackable.tech/v1alpha1
+kind: DruidCluster
+metadata:
+  name: druid
+  namespace: default
+  uid: 12345678-1234-1234-1234-123456789012
+spec:
+  image:
+    productVersion: 3.2.2
+  clusterConfig:
+    deepStorage:
+      hdfs:
+        configMapName: druid-hdfs
+        directory: /druid
+    metadataDatabase:
+      derby: {}
+    zookeeperConfigMapName: druid-znode
+  brokers:
+    roleGroups:
+      default:
+        replicas: 1
+  coordinators:
+    roleGroups:
+      default:
+        replicas: 1
+  historicals:
+    roleGroups:
+      default:
+        replicas: 1
+  middleManagers:
+    roleGroups:
+      default:
+        replicas: 1
+  routers:
+    roleGroups:
+      default:
+        replicas: 1
+"#,
+        )
+        .expect("valid cluster YAML");
+
+        let result = reconcile(druid);
+
+        assert!(
+            matches!(result, Err(Error::Dereference { .. })),
+            "a live cluster must reach the API but when dereferencing against the unreachable test server: {result:?}"
+        );
+    }
 }

rust/operator-binary/src/main.rs

@@ -15,7 +15,9 @@ use stackable_operator::{
     eos::EndOfSupportChecker,
     k8s_openapi::api::{
         apps::v1::StatefulSet,
-        core::v1::{ConfigMap, Service},
+        core::v1::{ConfigMap, Secret, Service, ServiceAccount},
+        policy::v1::PodDisruptionBudget,
+        rbac::v1::RoleBinding,
     },
     kube::{
         CustomResourceExt, ResourceExt,
@@ -125,19 +127,35 @@ async fn main() -> anyhow::Result<()> {
             let config_map_store = druid_controller.store();
             let druid_controller = druid_controller
                 .owns(
-                    watch_namespace.get_api::<Service>(&client),
+                    watch_namespace.get_api::<ConfigMap>(&client),
                     watcher::Config::default(),
                 )
                 .owns(
-                    watch_namespace.get_api::<StatefulSet>(&client),
+                    watch_namespace.get_api::<Listener>(&client),
                     watcher::Config::default(),
                 )
                 .owns(
-                    watch_namespace.get_api::<ConfigMap>(&client),
+                    watch_namespace.get_api::<PodDisruptionBudget>(&client),
                     watcher::Config::default(),
                 )
                 .owns(
-                    watch_namespace.get_api::<Listener>(&client),
+                    watch_namespace.get_api::<RoleBinding>(&client),
+                    watcher::Config::default(),
+                )
+                .owns(
+                    watch_namespace.get_api::<Secret>(&client),
+                    watcher::Config::default(),
+                )
+                .owns(
+                    watch_namespace.get_api::<Service>(&client),
+                    watcher::Config::default(),
+                )
+                .owns(
+                    watch_namespace.get_api::<ServiceAccount>(&client),
+                    watcher::Config::default(),
+                )
+                .owns(
+                    watch_namespace.get_api::<StatefulSet>(&client),
                     watcher::Config::default(),
                 )
                 .watches(