fix: watch s3 connection for changes

CHANGELOG.md

@@ -45,6 +45,7 @@ All notable changes to this project will be documented in this file.
   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 ([#741]).
 - The operator now watches all resources that it creates and early-exits the reconcile action when the
   cluster is marked for deletion ([#754]).
+- The operator now watches the `S3Connection` referenced by `spec.clusterConfig.s3.reference` ([#765]).
 
 [#726]: https://github.com/stackabletech/hive-operator/pull/726
 [#731]: https://github.com/stackabletech/hive-operator/pull/731
@@ -56,6 +57,7 @@ All notable changes to this project will be documented in this file.
 [#754]: https://github.com/stackabletech/hive-operator/pull/754
 [#759]: https://github.com/stackabletech/hive-operator/pull/759
 [#764]: https://github.com/stackabletech/hive-operator/pull/764
+[#765]: https://github.com/stackabletech/hive-operator/pull/765
 [#767]: https://github.com/stackabletech/hive-operator/pull/767
 [#770]: https://github.com/stackabletech/hive-operator/pull/770
 

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

@@ -118,7 +118,8 @@ rules:
       - {{ include "hive-operator.name" . }}clusters/status
     verbs:
       - patch
-  # Read S3Connection configuration referenced in the HiveCluster spec.
+  # Read S3Connection configuration referenced in the HiveCluster spec. Watched by the controller,
+  # so that changing it triggers a reconciliation.
   - apiGroups:
       - s3.stackable.tech
     resources:

rust/operator-binary/src/main.rs

@@ -10,7 +10,7 @@ use futures::{FutureExt, StreamExt, TryFutureExt};
 use stackable_operator::{
     YamlSchema,
     cli::{Command, RunArguments},
-    crd::listener::v1alpha1::Listener,
+    crd::{listener::v1alpha1::Listener, s3},
     eos::EndOfSupportChecker,
     k8s_openapi::api::{
         apps::v1::StatefulSet,
@@ -140,6 +140,7 @@ async fn main() -> anyhow::Result<()> {
                 watcher::Config::default(),
             );
             let config_map_store = hive_controller.store();
+            let s3_connection_store = hive_controller.store();
             let hive_controller = hive_controller
                 .owns(
                     watch_namespace.get_api::<DeserializeGuard<ConfigMap>>(&client),
@@ -184,6 +185,18 @@ async fn main() -> anyhow::Result<()> {
                             .map(|hive| ObjectRef::from_obj(&*hive))
                     },
                 )
+                .watches(
+                    watch_namespace
+                        .get_api::<DeserializeGuard<s3::v1alpha1::S3Connection>>(&client),
+                    watcher::Config::default(),
+                    move |s3_connection| {
+                        s3_connection_store
+                            .state()
+                            .into_iter()
+                            .filter(move |hive| references_s3_connection(hive, &s3_connection))
+                            .map(|hive| ObjectRef::from_obj(&*hive))
+                    },
+                )
                 .graceful_shutdown_on(sigterm_watcher.handle())
                 .run(
                     controller::reconcile_hive,
@@ -233,8 +246,177 @@ fn references_config_map(
         return false;
     };
 
+    if hive.namespace() != config_map.namespace() {
+        return false;
+    }
+
     match &hive.spec.cluster_config.hdfs {
         Some(hdfs_connection) => hdfs_connection.config_map.as_ref() == config_map.name_any(),
         None => false,
     }
 }
+
+fn references_s3_connection(
+    hive: &DeserializeGuard<v1alpha1::HiveCluster>,
+    s3_connection: &DeserializeGuard<s3::v1alpha1::S3Connection>,
+) -> bool {
+    let Ok(hive) = &hive.0 else {
+        return false;
+    };
+
+    if hive.namespace() != s3_connection.namespace() {
+        return false;
+    }
+
+    match &hive.spec.cluster_config.s3 {
+        Some(s3::v1alpha1::InlineConnectionOrReference::Reference(s3_connection_name)) => {
+            s3_connection_name == &s3_connection.name_any()
+        }
+        Some(s3::v1alpha1::InlineConnectionOrReference::Inline(_)) | None => false,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use indoc::indoc;
+    use rstest::rstest;
+
+    use super::*;
+
+    fn hive_cluster(cluster_config: &str) -> DeserializeGuard<v1alpha1::HiveCluster> {
+        let input = format!(
+            indoc! {r#"
+                apiVersion: hive.stackable.tech/v1alpha1
+                kind: HiveCluster
+                metadata:
+                  name: hive
+                  namespace: default
+                spec:
+                  image:
+                    productVersion: 4.2.0
+                  clusterConfig:
+                    metadataDatabase:
+                      derby: {{}}
+                    {cluster_config}
+                  metastore:
+                    roleGroups:
+                      default:
+                        replicas: 1
+            "#},
+            cluster_config = cluster_config
+        );
+
+        let hive = crate::controller::test_support::minimal_hive(&input);
+
+        DeserializeGuard(Ok(hive))
+    }
+
+    fn config_map(namespace: &str, name: &str) -> DeserializeGuard<ConfigMap> {
+        serde_yaml::from_str(&format!(
+            indoc! {r#"
+                apiVersion: v1
+                kind: ConfigMap
+                metadata:
+                  name: {name}
+                  namespace: {namespace}
+            "#},
+            name = name,
+            namespace = namespace
+        ))
+        .expect("ConfigMap YAML parses")
+    }
+
+    fn s3_connection(namespace: &str, name: &str) -> DeserializeGuard<s3::v1alpha1::S3Connection> {
+        serde_yaml::from_str(&format!(
+            indoc! {r#"
+                apiVersion: s3.stackable.tech/v1alpha1
+                kind: S3Connection
+                metadata:
+                  name: {name}
+                  namespace: {namespace}
+                spec:
+                  host: minio
+            "#},
+            name = name,
+            namespace = namespace
+        ))
+        .expect("S3Connection YAML parses")
+    }
+
+    #[rstest]
+    #[case::referenced("s3:\n      reference: minio", "default", "minio", true)]
+    #[case::other_connection("s3:\n      reference: minio", "default", "other", false)]
+    #[case::other_namespace("s3:\n      reference: minio", "elsewhere", "minio", false)]
+    #[case::inline("s3:\n      inline:\n        host: minio", "default", "minio", false)]
+    #[case::no_s3("", "default", "minio", false)]
+    fn references_s3_connection_matches_only_the_referenced_connection(
+        #[case] spec_s3: &str,
+        #[case] connection_namespace: &str,
+        #[case] connection_name: &str,
+        #[case] expected: bool,
+    ) {
+        assert_eq!(
+            references_s3_connection(
+                &hive_cluster(spec_s3),
+                &s3_connection(connection_namespace, connection_name)
+            ),
+            expected
+        );
+    }
+
+    #[test]
+    fn references_s3_connection_ignores_undeserializable_clusters() {
+        let hive = serde_yaml::from_str(indoc! {r#"
+            apiVersion: hive.stackable.tech/v1alpha1
+            kind: HiveCluster
+            metadata:
+              name: hive
+              namespace: default
+            spec: {}
+        "#})
+        .expect("YAML parses; the invalid spec is captured inside the DeserializeGuard");
+
+        assert!(!references_s3_connection(
+            &hive,
+            &s3_connection("default", "minio")
+        ));
+    }
+
+    #[rstest]
+    #[case::referenced("hdfs:\n      configMap: hdfs", "default", "hdfs", true)]
+    #[case::other_config_map("hdfs:\n      configMap: hdfs", "default", "other", false)]
+    #[case::other_namespace("hdfs:\n      configMap: hdfs", "elsewhere", "hdfs", false)]
+    #[case::no_hdfs("", "default", "hdfs", false)]
+    fn references_config_map_matches_only_the_referenced_config_map(
+        #[case] spec_hdfs: &str,
+        #[case] config_map_namespace: &str,
+        #[case] config_map_name: &str,
+        #[case] expected: bool,
+    ) {
+        assert_eq!(
+            references_config_map(
+                &hive_cluster(spec_hdfs),
+                &config_map(config_map_namespace, config_map_name)
+            ),
+            expected
+        );
+    }
+
+    #[test]
+    fn references_config_map_ignores_undeserializable_clusters() {
+        let hive = serde_yaml::from_str(indoc! {r#"
+            apiVersion: hive.stackable.tech/v1alpha1
+            kind: HiveCluster
+            metadata:
+              name: hive
+              namespace: default
+            spec: {}
+        "#})
+        .expect("YAML parses; the invalid spec is captured inside the DeserializeGuard");
+
+        assert!(!references_config_map(
+            &hive,
+            &config_map("default", "hdfs")
+        ));
+    }
+}