diff --git a/CHANGELOG.md b/CHANGELOG.md index 72ad617a..4267c8c8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,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 @@ -44,6 +45,7 @@ All notable changes to this project will be documented in this file. [#748]: https://github.com/stackabletech/hive-operator/pull/748 [#754]: https://github.com/stackabletech/hive-operator/pull/754 [#759]: https://github.com/stackabletech/hive-operator/pull/759 +[#765]: https://github.com/stackabletech/hive-operator/pull/765 ## [26.7.0] - 2026-07-21 diff --git a/deploy/helm/hive-operator/templates/clusterrole-operator.yaml b/deploy/helm/hive-operator/templates/clusterrole-operator.yaml index b0868771..68bd3e29 100644 --- a/deploy/helm/hive-operator/templates/clusterrole-operator.yaml +++ b/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: diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs index 56b68b56..be5c0c4a 100644 --- a/rust/operator-binary/src/main.rs +++ b/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, @@ -125,6 +125,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::>(&client), @@ -169,6 +170,18 @@ async fn main() -> anyhow::Result<()> { .map(|hive| ObjectRef::from_obj(&*hive)) }, ) + .watches( + watch_namespace + .get_api::>(&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, @@ -217,8 +230,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, + s3_connection: &DeserializeGuard, +) -> 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 { + 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 { + 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 { + 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") + )); + } +}