Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
184 changes: 183 additions & 1 deletion rust/operator-binary/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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::<DeserializeGuard<ConfigMap>>(&client),
Expand Down Expand Up @@ -169,6 +170,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,
Expand Down Expand Up @@ -217,8 +230,177 @@ fn references_config_map(
return false;
};

if hive.namespace() != config_map.namespace() {

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

drive-by - I assume this should be namespace-scoped as well

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

might be worth a tracing::warn! here, as I believe this would be very unexpected.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I looked into this a bit. As far as I can tell, this function is used as a filter and lots of mismatches (all HiveCluster x ConfigMap events) will occur here and are expected. We would have to be very specific about when we want to warn here. I am not sure whether this is the right place for it. @NickLarsenNZ any suggestions how to approach this correctly?

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;
}
Comment on lines +251 to +253

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same thing here with tracing::warn!


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")
));
}
}