From aee1646bc59f9c74ade4cf6351b917bf9b7e1db0 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Fri, 31 Jul 2026 16:29:14 +0200 Subject: [PATCH 1/3] extract apply and update_status steps --- rust/operator-binary/src/controller.rs | 187 +++++++----------- rust/operator-binary/src/controller/apply.rs | 151 ++++++++++++++ .../src/controller/build/mod.rs | 7 +- .../controller/build/resource/discovery.rs | 2 +- .../src/controller/build/resource/listener.rs | 5 +- .../src/controller/update_status.rs | 61 ++++++ 6 files changed, 289 insertions(+), 124 deletions(-) create mode 100644 rust/operator-binary/src/controller/apply.rs create mode 100644 rust/operator-binary/src/controller/update_status.rs diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index 3cbdb783..98db85af 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -1,10 +1,12 @@ //! Ensures that `Pod`s are configured and running for each [`v1alpha1::HiveCluster`] +mod apply; mod build; mod dereference; +mod update_status; mod validate; -use std::{collections::BTreeMap, hash::Hasher, str::FromStr, sync::Arc}; +use std::{collections::BTreeMap, hash::Hasher, marker::PhantomData, str::FromStr, sync::Arc}; use const_format::concatcp; use fnv::FnvHasher; @@ -35,13 +37,8 @@ use stackable_operator::{ kvp::Labels, logging::controller::ReconcilerError, shared::time::Duration, - status::condition::{ - compute_conditions, operations::ClusterOperationsConditionBuilder, - statefulset::StatefulSetConditionBuilder, - }, v2::{ HasName, HasUid, NameIsValidLabelValue, - cluster_resources::cluster_resources_new, kvp::label::{recommended_labels, role_group_selector}, role_group_utils::ResourceNames, role_utils, @@ -55,8 +52,12 @@ use strum::EnumDiscriminants; use crate::{ OPERATOR_NAME, - controller::build::{UNVERSIONED_PRODUCT_VERSION, resource::discovery}, - crd::{APP_NAME, HdfsConnection, HiveClusterStatus, HiveRole, MetaStoreConfig, v1alpha1}, + controller::{ + apply::Applier, + build::{UNVERSIONED_PRODUCT_VERSION, resource::discovery}, + update_status::update_status, + }, + crd::{APP_NAME, HdfsConnection, HiveRole, MetaStoreConfig, v1alpha1}, }; pub const HIVE_CONTROLLER_NAME: &str = "hivecluster"; @@ -70,31 +71,23 @@ pub struct Ctx { #[derive(Snafu, Debug, EnumDiscriminants)] #[strum_discriminants(derive(strum::IntoStaticStr))] pub enum Error { + #[snafu(display("failed to apply the Kubernetes resources"))] + ApplyResources { source: apply::Error }, + + #[snafu(display("failed to update the cluster status"))] + UpdateStatus { source: update_status::Error }, + #[snafu(display("failed to build the Kubernetes resources"))] BuildResources { source: build::Error }, - #[snafu(display("failed to apply Kubernetes resource"))] - ApplyResource { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to build discovery ConfigMap"))] BuildDiscoveryConfig { source: discovery::Error }, #[snafu(display("failed to apply discovery ConfigMap"))] - ApplyDiscoveryConfig { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to update status"))] - ApplyStatus { - source: stackable_operator::client::Error, - }, + ApplyDiscoveryConfig { source: apply::Error }, #[snafu(display("failed to delete orphaned resources"))] - DeleteOrphanedResources { - source: stackable_operator::cluster_resources::Error, - }, + DeleteOrphanedResources { source: apply::Error }, #[snafu(display("HiveCluster object is invalid"))] InvalidHiveCluster { @@ -416,12 +409,22 @@ pub struct ValidatedRoleConfig { pub listener_class: ListenerClassName, } +/// Marker for prepared Kubernetes resources which are not applied yet. +pub struct Prepared; + +/// Marker for applied Kubernetes resources. +pub struct Applied; + /// Every Kubernetes resource produced by the client-free [`build`](build::build) step. /// /// The role-level discovery `ConfigMap` is deliberately absent: it is built from the *applied* /// role [`Listener`]'s ingress addresses, so it is assembled in the reconcile step after the /// Listener has been applied, not in the build step. -pub struct KubernetesResources { +/// +/// `T` is a marker that indicates if these resources are only [`Prepared`] or already [`Applied`]. +/// The marker is useful e.g. to ensure that the cluster status is updated based on the applied +/// resources. +pub struct KubernetesResources { pub stateful_sets: Vec, pub services: Vec, pub listeners: Vec, @@ -429,6 +432,21 @@ pub struct KubernetesResources { pub pod_disruption_budgets: Vec, pub service_accounts: Vec, pub role_bindings: Vec, + pub status: PhantomData, +} + +impl KubernetesResources { + /// The applied role [`Listener`] of the given role, if it was built and applied. + pub fn role_listener( + &self, + cluster: &ValidatedCluster, + hive_role: &HiveRole, + ) -> Option<&Listener> { + let listener_name = cluster.role_listener_name(hive_role); + self.listeners + .iter() + .find(|listener| listener.metadata.name.as_deref() == Some(listener_name.as_ref())) + } } pub async fn reconcile_hive( @@ -454,122 +472,55 @@ pub async fn reconcile_hive( ) .context(ValidateSnafu)?; - let mut cluster_resources = cluster_resources_new( - &product_name(), - &operator_name(), - &controller_name(), - &validated_cluster.name, - &validated_cluster.namespace, - &validated_cluster.uid, - ClusterResourceApplyStrategy::from(&hive.spec.cluster_operation), - &hive.spec.object_overrides, - ); - let resources = build::build(&validated_cluster, &client.kubernetes_cluster_info) .context(BuildResourcesSnafu)?; - let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - - // Apply order: everything before StatefulSets, StatefulSets last. A StatefulSet must only be - // applied after all ConfigMaps and Secrets it mounts, to prevent unnecessary Pod restarts. - // See https://github.com/stackabletech/commons-operator/issues/111 for details. - for service_account in resources.service_accounts { - cluster_resources - .add(client, service_account) - .await - .context(ApplyResourceSnafu)?; - } - for role_binding in resources.role_bindings { - cluster_resources - .add(client, role_binding) - .await - .context(ApplyResourceSnafu)?; - } - for service in resources.services { - cluster_resources - .add(client, service) - .await - .context(ApplyResourceSnafu)?; - } - - // The role Listener is applied before the discovery ConfigMap, which is built below from the - // applied Listener's ingress addresses. Hive has a single role Listener, so at most one is - // captured here. - let mut applied_role_listener: Option = None; - for listener in resources.listeners { - applied_role_listener = Some( - cluster_resources - .add(client, listener) - .await - .context(ApplyResourceSnafu)?, - ); - } - - for config_map in resources.config_maps { - cluster_resources - .add(client, config_map) - .await - .context(ApplyResourceSnafu)?; - } - - for pdb in resources.pod_disruption_budgets { - cluster_resources - .add(client, pdb) - .await - .context(ApplyResourceSnafu)?; - } + let mut applier = Applier::new( + client, + &validated_cluster, + ClusterResourceApplyStrategy::from(&hive.spec.cluster_operation), + &hive.spec.object_overrides, + ); - for statefulset in resources.stateful_sets { - ss_cond_builder.add( - cluster_resources - .add(client, statefulset) - .await - .context(ApplyResourceSnafu)?, - ); - } + let applied = applier + .apply(resources) + .await + .context(ApplyResourcesSnafu)?; - // The discovery ConfigMap is built from the *applied* role Listener's ingress addresses, so it - // is assembled here rather than in the client-free build step. Its applied resource version - // feeds the status discovery hash. + // Second apply phase: the discovery ConfigMap is built from the *applied* role Listener's + // ingress addresses, which are only known after the Listener has been applied. It goes + // through the same Applier, so that the orphan deletion in `finish` sees it. Its applied + // resource version feeds the status discovery hash. // std's SipHasher is deprecated, and DefaultHasher is unstable across Rust releases. // We don't /need/ stability, but it's still nice to avoid spurious changes where possible. let mut discovery_hash = FnvHasher::with_key(0); - if let Some(role_listener) = applied_role_listener { + if let Some(role_listener) = applied.role_listener(&validated_cluster, &HiveRole::MetaStore) { let discovery_cm = discovery::build_discovery_configmap( &validated_cluster, HiveRole::MetaStore, role_listener, ) .context(BuildDiscoveryConfigSnafu)?; - let discovery_cm = cluster_resources - .add(client, discovery_cm) + let applied_discovery_cms = applier + .apply_config_maps(vec![discovery_cm]) .await .context(ApplyDiscoveryConfigSnafu)?; - if let Some(generation) = discovery_cm.metadata.resource_version { - discovery_hash.write(generation.as_bytes()); + for discovery_cm in &applied_discovery_cms { + if let Some(resource_version) = &discovery_cm.metadata.resource_version { + discovery_hash.write(resource_version.as_bytes()); + } } } - let cluster_operation_cond_builder = - ClusterOperationsConditionBuilder::new(&hive.spec.cluster_operation); - - let status = HiveClusterStatus { - // Serialize as a string to discourage users from trying to parse the value, - // and to keep things flexible if we end up changing the hasher at some point. - discovery_hash: Some(discovery_hash.finish().to_string()), - conditions: compute_conditions(hive, &[&ss_cond_builder, &cluster_operation_cond_builder]), - }; - - client - .apply_patch_status(OPERATOR_NAME, hive, &status) + applier + .finish() .await - .context(ApplyStatusSnafu)?; + .context(DeleteOrphanedResourcesSnafu)?; - cluster_resources - .delete_orphaned_resources(client) + update_status(client, hive, &applied, discovery_hash.finish()) .await - .context(DeleteOrphanedResourcesSnafu)?; + .context(UpdateStatusSnafu)?; Ok(Action::await_change()) } diff --git a/rust/operator-binary/src/controller/apply.rs b/rust/operator-binary/src/controller/apply.rs new file mode 100644 index 00000000..c2712716 --- /dev/null +++ b/rust/operator-binary/src/controller/apply.rs @@ -0,0 +1,151 @@ +//! The apply step in the HiveCluster controller. + +use std::marker::PhantomData; + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + cluster_resources::{ClusterResource, ClusterResourceApplyStrategy, ClusterResources}, + deep_merger::ObjectOverrides, + k8s_openapi::api::core::v1::ConfigMap, + v2::cluster_resources::cluster_resources_new, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::controller::{ + Applied, KubernetesResources, Prepared, ValidatedCluster, controller_name, operator_name, + product_name, +}; + +#[derive(Snafu, Debug, EnumDiscriminants)] +#[strum_discriminants(derive(IntoStaticStr))] +pub enum Error { + #[snafu(display("failed to apply Kubernetes resource"))] + ApplyResource { + source: stackable_operator::cluster_resources::Error, + }, + + #[snafu(display("failed to delete orphaned resources"))] + DeleteOrphanedResources { + source: stackable_operator::cluster_resources::Error, + }, +} + +type Result = std::result::Result; + +/// Applier for the Kubernetes resource specifications produced by this controller. +/// +/// The implementation is not tied to this controller and could theoretically be moved to +/// stackable_operator if [`KubernetesResources`] would contain all possible resource types. +pub struct Applier<'a> { + client: &'a Client, + cluster_resources: ClusterResources<'a>, +} + +impl<'a> Applier<'a> { + pub fn new( + client: &'a Client, + cluster: &ValidatedCluster, + apply_strategy: ClusterResourceApplyStrategy, + object_overrides: &'a ObjectOverrides, + ) -> Applier<'a> { + let cluster_resources = cluster_resources_new( + &product_name(), + &operator_name(), + &controller_name(), + &cluster.name, + &cluster.namespace, + &cluster.uid, + apply_strategy, + object_overrides, + ); + + Applier { + client, + cluster_resources, + } + } + + /// Applies the given Kubernetes resources and marks them as applied. + /// + /// Resources derived from the applied state (the discovery `ConfigMap`) can be applied + /// afterwards via [`Self::apply_config_maps`]; [`Self::finish`] must be called once all + /// resources are applied, so that orphaned resources are deleted exactly once at the end. + pub async fn apply( + &mut self, + resources: KubernetesResources, + ) -> Result> { + // Destructured without `..`, so adding a field to [`KubernetesResources`] fails to + // compile here instead of silently never being applied. + let KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + pod_disruption_budgets, + service_accounts, + role_bindings, + status: _, + } = resources; + + // Apply order is: StatefulSets last (a changed mounted ConfigMap/Secret + // must exist first, else Pods restart -- commons-operator#111). The ServiceAccount comes + // first because the Pods reference it at creation time. + let service_accounts = self.add_resources(service_accounts).await?; + let role_bindings = self.add_resources(role_bindings).await?; + let services = self.add_resources(services).await?; + let listeners = self.add_resources(listeners).await?; + let config_maps = self.add_resources(config_maps).await?; + let pod_disruption_budgets = self.add_resources(pod_disruption_budgets).await?; + let stateful_sets = self.add_resources(stateful_sets).await?; + + Ok(KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + pod_disruption_budgets, + service_accounts, + role_bindings, + status: PhantomData, + }) + } + + /// Applies `ConfigMap`s that are derived from already-applied resources (the discovery + /// `ConfigMap`, which needs the applied role Listener's ingress addresses). + pub async fn apply_config_maps( + &mut self, + config_maps: Vec, + ) -> Result> { + self.add_resources(config_maps).await + } + + /// Deletes resources from earlier reconcile runs that were not applied in this one. + /// + /// Must be called exactly once, after every apply phase: a resource applied after this call + /// would be treated as an orphan and deleted by the next reconcile run. + pub async fn finish(self) -> Result<()> { + self.cluster_resources + .delete_orphaned_resources(self.client) + .await + .context(DeleteOrphanedResourcesSnafu) + } + + async fn add_resources( + &mut self, + resources: Vec, + ) -> Result> { + let mut applied_resources = vec![]; + + for resource in resources { + let applied_resource = self + .cluster_resources + .add(self.client, resource) + .await + .context(ApplyResourceSnafu)?; + applied_resources.push(applied_resource); + } + + Ok(applied_resources) + } +} diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index f6f79545..77191eff 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -1,6 +1,6 @@ //! Builders that turn a `ValidatedCluster` into Kubernetes resources. -use std::str::FromStr; +use std::{marker::PhantomData, str::FromStr}; use snafu::{ResultExt, Snafu}; use stackable_operator::{ @@ -14,7 +14,7 @@ use stackable_operator::{ use crate::{ controller::{ - KubernetesResources, ValidatedCluster, + KubernetesResources, Prepared, ValidatedCluster, build::resource::{ config_map::build_metastore_rolegroup_config_map, listener::build_role_listener, @@ -77,7 +77,7 @@ pub enum Error { pub fn build( cluster: &ValidatedCluster, cluster_info: &KubernetesClusterInfo, -) -> Result { +) -> Result, Error> { let mut stateful_sets = vec![]; let mut services = vec![]; let mut listeners = vec![]; @@ -121,6 +121,7 @@ pub fn build( pod_disruption_budgets, service_accounts: vec![build_service_account(cluster)], role_bindings: vec![build_role_binding(cluster)], + status: PhantomData, }) } diff --git a/rust/operator-binary/src/controller/build/resource/discovery.rs b/rust/operator-binary/src/controller/build/resource/discovery.rs index 1f4a2858..0756daf4 100644 --- a/rust/operator-binary/src/controller/build/resource/discovery.rs +++ b/rust/operator-binary/src/controller/build/resource/discovery.rs @@ -42,7 +42,7 @@ fn cluster_object_ref(cluster: &ValidatedCluster) -> ObjectRef Result { let mut discovery_configmap = ConfigMapBuilder::new(); diff --git a/rust/operator-binary/src/controller/build/resource/listener.rs b/rust/operator-binary/src/controller/build/resource/listener.rs index b6570495..03152881 100644 --- a/rust/operator-binary/src/controller/build/resource/listener.rs +++ b/rust/operator-binary/src/controller/build/resource/listener.rs @@ -22,13 +22,14 @@ pub enum Error { // Builds the connection string with respect to the listener provided objects pub fn build_listener_connection_string( - listener_ref: Listener, + listener_ref: &Listener, role: &str, ) -> Result { // We only need the first address corresponding to the role let listener_address = listener_ref .status - .and_then(|s| s.ingress_addresses?.into_iter().next()) + .as_ref() + .and_then(|status| status.ingress_addresses.as_ref()?.first()) .context(RoleListenerHasNoAddressSnafu { role })?; let conn_str = format!( "thrift://{address}:{port}", diff --git a/rust/operator-binary/src/controller/update_status.rs b/rust/operator-binary/src/controller/update_status.rs new file mode 100644 index 00000000..3e5bdbe1 --- /dev/null +++ b/rust/operator-binary/src/controller/update_status.rs @@ -0,0 +1,61 @@ +//! The update_status step in the HiveCluster controller. + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + status::condition::{ + compute_conditions, operations::ClusterOperationsConditionBuilder, + statefulset::StatefulSetConditionBuilder, + }, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::{ + OPERATOR_NAME, + controller::{Applied, KubernetesResources}, + crd::{HiveClusterStatus, v1alpha1}, +}; + +#[derive(Snafu, Debug, EnumDiscriminants)] +#[strum_discriminants(derive(IntoStaticStr))] +pub enum Error { + #[snafu(display("failed to update status"))] + ApplyStatus { + source: stackable_operator::client::Error, + }, +} + +type Result = std::result::Result; + +/// Computes the cluster status from the applied resources and patches it onto the +/// [`v1alpha1::HiveCluster`]. Takes [`KubernetesResources`] so the type system proves +/// the status derives from applied resources, not merely built ones. `discovery_hash` is +/// derived from the applied discovery `ConfigMap`'s resource version in the reconcile step. +pub async fn update_status( + client: &Client, + hive: &v1alpha1::HiveCluster, + applied: &KubernetesResources, + discovery_hash: u64, +) -> Result<()> { + let mut ss_cond_builder = StatefulSetConditionBuilder::default(); + for stateful_set in &applied.stateful_sets { + ss_cond_builder.add(stateful_set.clone()); + } + + let cluster_operation_cond_builder = + ClusterOperationsConditionBuilder::new(&hive.spec.cluster_operation); + + let status = HiveClusterStatus { + // Serialize as a string to discourage users from trying to parse the value, + // and to keep things flexible if we end up changing the hasher at some point. + discovery_hash: Some(discovery_hash.to_string()), + conditions: compute_conditions(hive, &[&ss_cond_builder, &cluster_operation_cond_builder]), + }; + + client + .apply_patch_status(OPERATOR_NAME, hive, &status) + .await + .context(ApplyStatusSnafu)?; + + Ok(()) +} From 412454aa534f4925cc7e87239047b17494131f53 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Wed, 5 Aug 2026 19:13:12 +0200 Subject: [PATCH 2/3] rework discovery config map so that listeners are actively watched and used once complete --- CHANGELOG.md | 3 + .../templates/clusterrole-operator.yaml | 6 +- rust/operator-binary/src/controller.rs | 90 ++----------- rust/operator-binary/src/controller/apply.rs | 36 ++--- .../src/controller/build/mod.rs | 123 ++++++++++++++++-- .../src/controller/build/resource/listener.rs | 18 ++- .../controller/build/resource/statefulset.rs | 3 +- .../src/controller/dereference.rs | 40 +++++- .../src/controller/update_status.rs | 84 +++++++++++- .../src/controller/validate.rs | 2 + rust/operator-binary/src/main.rs | 9 ++ 11 files changed, 290 insertions(+), 124 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f00304b7..ed5897c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,10 +13,13 @@ All notable changes to this project will be documented in this file. - BREAKING: The `metastore` role is now required by the CRD; a HiveCluster without it was previously accepted by the API server but failed reconciliation ([#731]). - Bump stackable-operator to 0.114.0 ([#735]). +- The reconciler now applies resources and derives the cluster status in discrete + apply and update_status steps ([#737]). [#726]: https://github.com/stackabletech/hive-operator/pull/726 [#731]: https://github.com/stackabletech/hive-operator/pull/731 [#735]: https://github.com/stackabletech/hive-operator/pull/735 +[#737]: https://github.com/stackabletech/hive-operator/pull/737 ## [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 f19df8a1..c78e6be7 100644 --- a/deploy/helm/hive-operator/templates/clusterrole-operator.yaml +++ b/deploy/helm/hive-operator/templates/clusterrole-operator.yaml @@ -134,8 +134,9 @@ rules: - get - list - watch - # Listener created per role group for external access. Applied via SSA and tracked for orphan - # cleanup. + # Listener created per role for external access. Applied via SSA and tracked for orphan + # cleanup. Watched by the controller, so a reconciliation is triggered once the + # listener-operator writes the ingress address (needed for the discovery ConfigMap). - apiGroups: - listeners.stackable.tech resources: @@ -146,3 +147,4 @@ rules: - get - list - patch + - watch diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index 98db85af..1c09085a 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -6,10 +6,9 @@ mod dereference; mod update_status; mod validate; -use std::{collections::BTreeMap, hash::Hasher, marker::PhantomData, str::FromStr, sync::Arc}; +use std::{collections::BTreeMap, marker::PhantomData, str::FromStr, sync::Arc}; use const_format::concatcp; -use fnv::FnvHasher; use snafu::{ResultExt, Snafu}; pub use stackable_operator::v2::types::operator::RoleGroupName; use stackable_operator::{ @@ -43,7 +42,7 @@ use stackable_operator::{ role_group_utils::ResourceNames, role_utils, types::{ - kubernetes::{ListenerClassName, ListenerName, SecretClassName}, + kubernetes::{ListenerClassName, SecretClassName}, operator::{ControllerName, OperatorName, ProductName, ProductVersion, RoleName}, }, }, @@ -53,9 +52,7 @@ use strum::EnumDiscriminants; use crate::{ OPERATOR_NAME, controller::{ - apply::Applier, - build::{UNVERSIONED_PRODUCT_VERSION, resource::discovery}, - update_status::update_status, + apply::Applier, build::UNVERSIONED_PRODUCT_VERSION, update_status::update_status, }, crd::{APP_NAME, HdfsConnection, HiveRole, MetaStoreConfig, v1alpha1}, }; @@ -80,15 +77,6 @@ pub enum Error { #[snafu(display("failed to build the Kubernetes resources"))] BuildResources { source: build::Error }, - #[snafu(display("failed to build discovery ConfigMap"))] - BuildDiscoveryConfig { source: discovery::Error }, - - #[snafu(display("failed to apply discovery ConfigMap"))] - ApplyDiscoveryConfig { source: apply::Error }, - - #[snafu(display("failed to delete orphaned resources"))] - DeleteOrphanedResources { source: apply::Error }, - #[snafu(display("HiveCluster object is invalid"))] InvalidHiveCluster { source: error_boundary::InvalidObject, @@ -186,9 +174,15 @@ pub struct ValidatedCluster { pub role_config: ValidatedRoleConfig, pub cluster_config: ValidatedClusterConfig, pub role_group_configs: BTreeMap>, + /// The metastore role [`Listener`] as currently stored in the cluster (fetched in the + /// dereference step), from which the discovery `ConfigMap` is built. `None` on the first + /// reconcile run, and possibly still address-less until the listener-operator has + /// reconciled it. + pub role_listener: Option, } impl ValidatedCluster { + #[allow(clippy::too_many_arguments)] pub fn new( name: stackable_operator::v2::types::operator::ClusterName, namespace: stackable_operator::v2::types::kubernetes::NamespaceName, @@ -197,6 +191,7 @@ impl ValidatedCluster { role_config: ValidatedRoleConfig, cluster_config: ValidatedClusterConfig, role_group_configs: BTreeMap>, + role_listener: Option, ) -> Self { // `app_version_label_value` is constructed to be a valid label value, so it is also a // valid `ProductVersion`. @@ -217,6 +212,7 @@ impl ValidatedCluster { role_config, cluster_config, role_group_configs, + role_listener, } } @@ -298,16 +294,6 @@ impl ValidatedCluster { pub fn has_kerberos_enabled(&self) -> bool { self.cluster_config.kerberos_secret_class.is_some() } - - /// The name of the per-role [`Listener`] object. - pub fn role_listener_name(&self, hive_role: &HiveRole) -> ListenerName { - ListenerName::from_str(&format!( - "{name}-{role}", - name = self.name, - role = hive_role - )) - .expect("the role listener name is a valid Listener name") - } } /// Lets [`ValidatedCluster`] stand in for the raw [`v1alpha1::HiveCluster`] when building owner @@ -417,10 +403,6 @@ pub struct Applied; /// Every Kubernetes resource produced by the client-free [`build`](build::build) step. /// -/// The role-level discovery `ConfigMap` is deliberately absent: it is built from the *applied* -/// role [`Listener`]'s ingress addresses, so it is assembled in the reconcile step after the -/// Listener has been applied, not in the build step. -/// /// `T` is a marker that indicates if these resources are only [`Prepared`] or already [`Applied`]. /// The marker is useful e.g. to ensure that the cluster status is updated based on the applied /// resources. @@ -435,20 +417,6 @@ pub struct KubernetesResources { pub status: PhantomData, } -impl KubernetesResources { - /// The applied role [`Listener`] of the given role, if it was built and applied. - pub fn role_listener( - &self, - cluster: &ValidatedCluster, - hive_role: &HiveRole, - ) -> Option<&Listener> { - let listener_name = cluster.role_listener_name(hive_role); - self.listeners - .iter() - .find(|listener| listener.metadata.name.as_deref() == Some(listener_name.as_ref())) - } -} - pub async fn reconcile_hive( hive: Arc>, ctx: Arc, @@ -475,7 +443,7 @@ pub async fn reconcile_hive( let resources = build::build(&validated_cluster, &client.kubernetes_cluster_info) .context(BuildResourcesSnafu)?; - let mut applier = Applier::new( + let applier = Applier::new( client, &validated_cluster, ClusterResourceApplyStrategy::from(&hive.spec.cluster_operation), @@ -487,38 +455,7 @@ pub async fn reconcile_hive( .await .context(ApplyResourcesSnafu)?; - // Second apply phase: the discovery ConfigMap is built from the *applied* role Listener's - // ingress addresses, which are only known after the Listener has been applied. It goes - // through the same Applier, so that the orphan deletion in `finish` sees it. Its applied - // resource version feeds the status discovery hash. - // std's SipHasher is deprecated, and DefaultHasher is unstable across Rust releases. - // We don't /need/ stability, but it's still nice to avoid spurious changes where possible. - let mut discovery_hash = FnvHasher::with_key(0); - - if let Some(role_listener) = applied.role_listener(&validated_cluster, &HiveRole::MetaStore) { - let discovery_cm = discovery::build_discovery_configmap( - &validated_cluster, - HiveRole::MetaStore, - role_listener, - ) - .context(BuildDiscoveryConfigSnafu)?; - let applied_discovery_cms = applier - .apply_config_maps(vec![discovery_cm]) - .await - .context(ApplyDiscoveryConfigSnafu)?; - for discovery_cm in &applied_discovery_cms { - if let Some(resource_version) = &discovery_cm.metadata.resource_version { - discovery_hash.write(resource_version.as_bytes()); - } - } - } - - applier - .finish() - .await - .context(DeleteOrphanedResourcesSnafu)?; - - update_status(client, hive, &applied, discovery_hash.finish()) + update_status(client, hive, &applied) .await .context(UpdateStatusSnafu)?; @@ -592,6 +529,7 @@ pub(crate) mod test_support { DereferencedObjects { s3_connection_spec: None, hive_opa_config: None, + role_listener: None, }, ) .expect("validate should succeed for the test fixture") diff --git a/rust/operator-binary/src/controller/apply.rs b/rust/operator-binary/src/controller/apply.rs index c2712716..98cd8a5e 100644 --- a/rust/operator-binary/src/controller/apply.rs +++ b/rust/operator-binary/src/controller/apply.rs @@ -7,7 +7,6 @@ use stackable_operator::{ client::Client, cluster_resources::{ClusterResource, ClusterResourceApplyStrategy, ClusterResources}, deep_merger::ObjectOverrides, - k8s_openapi::api::core::v1::ConfigMap, v2::cluster_resources::cluster_resources_new, }; use strum::{EnumDiscriminants, IntoStaticStr}; @@ -66,13 +65,13 @@ impl<'a> Applier<'a> { } } - /// Applies the given Kubernetes resources and marks them as applied. + /// Applies the given Kubernetes resources, deletes resources from earlier reconcile runs + /// that were not applied in this one, and marks the resources as applied. /// - /// Resources derived from the applied state (the discovery `ConfigMap`) can be applied - /// afterwards via [`Self::apply_config_maps`]; [`Self::finish`] must be called once all - /// resources are applied, so that orphaned resources are deleted exactly once at the end. + /// Consumes the applier: a resource applied after the orphan deletion would itself be + /// treated as an orphan and deleted by the next reconcile run. pub async fn apply( - &mut self, + mut self, resources: KubernetesResources, ) -> Result> { // Destructured without `..`, so adding a field to [`KubernetesResources`] fails to @@ -99,6 +98,11 @@ impl<'a> Applier<'a> { let pod_disruption_budgets = self.add_resources(pod_disruption_budgets).await?; let stateful_sets = self.add_resources(stateful_sets).await?; + self.cluster_resources + .delete_orphaned_resources(self.client) + .await + .context(DeleteOrphanedResourcesSnafu)?; + Ok(KubernetesResources { stateful_sets, services, @@ -111,26 +115,6 @@ impl<'a> Applier<'a> { }) } - /// Applies `ConfigMap`s that are derived from already-applied resources (the discovery - /// `ConfigMap`, which needs the applied role Listener's ingress addresses). - pub async fn apply_config_maps( - &mut self, - config_maps: Vec, - ) -> Result> { - self.add_resources(config_maps).await - } - - /// Deletes resources from earlier reconcile runs that were not applied in this one. - /// - /// Must be called exactly once, after every apply phase: a resource applied after this call - /// would be treated as an orphan and deleted by the next reconcile run. - pub async fn finish(self) -> Result<()> { - self.cluster_resources - .delete_orphaned_resources(self.client) - .await - .context(DeleteOrphanedResourcesSnafu) - } - async fn add_resources( &mut self, resources: Vec, diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 77191eff..26856a21 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -17,6 +17,7 @@ use crate::{ KubernetesResources, Prepared, ValidatedCluster, build::resource::{ config_map::build_metastore_rolegroup_config_map, + discovery::build_discovery_configmap, listener::build_role_listener, pdb::build_pdb, rbac::{build_role_binding, build_service_account}, @@ -60,6 +61,9 @@ pub enum Error { source: resource::statefulset::Error, role_group: RoleGroupName, }, + + #[snafu(display("failed to build the discovery ConfigMap"))] + DiscoveryConfigMap { source: resource::discovery::Error }, } /// Builds every Kubernetes resource for the given validated cluster. @@ -68,10 +72,6 @@ pub enum Error { /// dereferenced and validated by this point. Cluster configuration is likewise already validated, /// so the errors returned here are resource-assembly failures only. /// -/// The role-level discovery `ConfigMap` is *not* built here: it depends on the *applied* role -/// [`Listener`](stackable_operator::crd::listener::v1alpha1::Listener)'s ingress addresses and is -/// therefore assembled in the reconcile step after the Listener has been applied. -/// /// `cluster_info` carries the Kubernetes cluster domain (needed by the Kerberos config); it is /// static cluster metadata, not a live client, so the build step stays client-free. pub fn build( @@ -84,8 +84,8 @@ pub fn build( let mut config_maps = vec![]; let mut pod_disruption_budgets = vec![]; - // Role-level resources. Hive has the single `metastore` role; its PDB and Listener are built - // here, but the discovery ConfigMap (which needs the applied Listener) is built in reconcile. + // Role-level resources. Hive has the single `metastore` role; its PDB and Listener are + // built here. The discovery ConfigMap is built below, from the dereferenced Listener. let role_config = &cluster.role_config; pod_disruption_budgets.extend(build_pdb(&role_config.pdb, cluster, &HiveRole::MetaStore)); listeners.push(build_role_listener( @@ -113,6 +113,30 @@ pub fn build( } } + // The discovery ConfigMap needs the role Listener's ingress address, which only the + // listener-operator writes. Around the first reconcile runs the dereferenced Listener is + // absent or still address-less; the ConfigMap is skipped then instead of failing the whole + // run -- the Listener watch triggers a new run once the address is set. In that window an + // already existing discovery ConfigMap is deleted as an orphan (only reachable when the + // Listener is deleted and re-created) and re-created by the next run. + if let Some(role_listener) = &cluster.role_listener + && role_listener + .status + .as_ref() + .and_then(|status| status.ingress_addresses.as_ref()?.first()) + .is_some() + { + config_maps.push( + build_discovery_configmap(cluster, HiveRole::MetaStore, role_listener) + .context(DiscoveryConfigMapSnafu)?, + ); + } else { + tracing::debug!( + "the metastore role Listener has no ingress address yet, skipping the discovery \ + ConfigMap" + ); + } + Ok(KubernetesResources { stateful_sets, services, @@ -146,14 +170,21 @@ pub(crate) fn object_meta( #[cfg(test)] mod tests { - use std::str::FromStr; + use std::{collections::BTreeMap, str::FromStr}; use stackable_operator::{ - commons::networking::DomainName, kube::Resource, utils::cluster_info::KubernetesClusterInfo, + commons::networking::DomainName, + crd::listener::{self, v1alpha1::Listener}, + k8s_openapi::api::core::v1::ConfigMap, + kube::{Resource, api::ObjectMeta}, + utils::cluster_info::KubernetesClusterInfo, }; - use super::{RoleGroupName, build, object_meta}; - use crate::controller::test_support::{DERBY_YAML, minimal_hive, validated_cluster}; + use super::{KubernetesResources, Prepared, RoleGroupName, build, object_meta}; + use crate::{ + controller::test_support::{DERBY_YAML, minimal_hive, validated_cluster}, + crd::HIVE_PORT_NAME, + }; fn test_cluster_info() -> KubernetesClusterInfo { KubernetesClusterInfo { @@ -214,6 +245,78 @@ mod tests { ); } + /// A metastore role Listener whose status carries an ingress address, as the + /// listener-operator eventually writes it. + fn role_listener_with_address() -> Listener { + Listener { + metadata: ObjectMeta::default(), + spec: listener::v1alpha1::ListenerSpec::default(), + status: Some(listener::v1alpha1::ListenerStatus { + service_name: None, + ingress_addresses: Some(vec![listener::v1alpha1::ListenerIngress { + address: "hive.example.com".to_string(), + address_type: listener::v1alpha1::AddressType::Hostname, + ports: BTreeMap::from([(HIVE_PORT_NAME.to_string(), 9083)]), + }]), + node_ports: None, + }), + } + } + + /// The built discovery ConfigMap (named after the cluster itself), if any. + fn discovery_config_map(resources: &KubernetesResources) -> Option<&ConfigMap> { + resources + .config_maps + .iter() + .find(|config_map| config_map.metadata.name.as_deref() == Some("simple-hive")) + } + + #[test] + fn build_adds_the_discovery_config_map_once_the_role_listener_has_an_address() { + let hive = minimal_hive(DERBY_YAML); + let mut cluster = validated_cluster(&hive); + cluster.role_listener = Some(role_listener_with_address()); + + let resources = build(&cluster, &test_cluster_info()).expect("build succeeds"); + + let data = discovery_config_map(&resources) + .expect("the discovery ConfigMap is built") + .data + .as_ref() + .expect("the discovery ConfigMap carries data"); + assert_eq!( + data.get("HIVE").map(String::as_str), + Some("thrift://hive.example.com:9083") + ); + } + + /// While the Listener carries no ingress address (the listener-operator has not reconciled + /// it yet), the discovery ConfigMap is skipped, *without* failing the build: the Listener + /// watch triggers a new reconcile run once the address is set. + #[test] + fn build_skips_the_discovery_config_map_while_the_role_listener_has_no_address() { + let hive = minimal_hive(DERBY_YAML); + let mut cluster = validated_cluster(&hive); + + let no_status = None; + let no_addresses = Some(listener::v1alpha1::ListenerStatus { + service_name: None, + ingress_addresses: Some(vec![]), + node_ports: None, + }); + for status in [no_status, no_addresses] { + cluster.role_listener = Some(Listener { + status, + ..role_listener_with_address() + }); + + let resources = + build(&cluster, &test_cluster_info()).expect("build succeeds without an address"); + + assert!(discovery_config_map(&resources).is_none()); + } + } + #[test] fn object_meta_sets_namespace_owner_and_recommended_labels() { let hive = minimal_hive(DERBY_YAML); diff --git a/rust/operator-binary/src/controller/build/resource/listener.rs b/rust/operator-binary/src/controller/build/resource/listener.rs index 03152881..7940f73e 100644 --- a/rust/operator-binary/src/controller/build/resource/listener.rs +++ b/rust/operator-binary/src/controller/build/resource/listener.rs @@ -1,7 +1,12 @@ +use std::str::FromStr; + use snafu::{OptionExt, Snafu}; use stackable_operator::{ crd::listener::v1alpha1::{Listener, ListenerPort, ListenerSpec}, - v2::types::kubernetes::ListenerClassName, + v2::types::{ + kubernetes::{ListenerClassName, ListenerName}, + operator::ClusterName, + }, }; use crate::{ @@ -46,6 +51,15 @@ pub fn build_listener_connection_string( Ok(conn_str) } +/// The name of the per-role [`Listener`] object. +/// +/// Takes the bare cluster name (not [`ValidatedCluster`]) so the dereference step, which runs +/// before validation, can derive the same name. +pub fn role_listener_name(cluster_name: &ClusterName, hive_role: &HiveRole) -> ListenerName { + ListenerName::from_str(&format!("{cluster_name}-{hive_role}")) + .expect("the role listener name is a valid Listener name") +} + // Designed to build a listener per role // In case of Hive we expect only one role: Metastore pub fn build_role_listener( @@ -57,7 +71,7 @@ pub fn build_role_listener( // role-group name; "none" is used as a placeholder for the recommended labels. let metadata = object_meta( cluster, - cluster.role_listener_name(hive_role), + role_listener_name(&cluster.name, hive_role), &PLACEHOLDER_LISTENER_ROLE_GROUP, ) .build(); diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index e79add71..d7b430e5 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -55,6 +55,7 @@ use crate::{ object_meta, opa::{OPA_TLS_VOLUME_NAME, build_opa_tls_ca_cert_mount_path}, properties::product_logging::MAX_HIVE_LOG_FILES_SIZE, + resource::listener::role_listener_name, }, }, crd::{ @@ -330,7 +331,7 @@ pub(crate) fn build_metastore_rolegroup_statefulset( .with_labels(recommended_object_labels) .build(); - let listener_name = cluster.role_listener_name(hive_role); + let listener_name = role_listener_name(&cluster.name, hive_role); let pvc = listener_operator_volume_source_builder_build_pvc( &ListenerReference::Listener(listener_name), &unversioned_recommended_labels, diff --git a/rust/operator-binary/src/controller/dereference.rs b/rust/operator-binary/src/controller/dereference.rs index 994f73c9..e5298e0f 100644 --- a/rust/operator-binary/src/controller/dereference.rs +++ b/rust/operator-binary/src/controller/dereference.rs @@ -4,12 +4,18 @@ use snafu::{ResultExt, Snafu}; use stackable_operator::{ client::Client, commons::opa::{OpaApiVersion, OpaConfig}, - crd::s3, + crd::{listener::v1alpha1::Listener, s3}, k8s_openapi::api::core::v1::ConfigMap, - v2::{controller_utils::get_namespace, types::kubernetes::SecretClassName}, + v2::{ + controller_utils::{get_cluster_name, get_namespace}, + types::kubernetes::SecretClassName, + }, }; -use crate::crd::v1alpha1; +use crate::{ + controller::build::resource::listener::role_listener_name, + crd::{HiveRole, v1alpha1}, +}; #[derive(Snafu, Debug)] pub enum Error { @@ -32,12 +38,29 @@ pub enum Error { ParseOpaTlsSecretClassName { source: stackable_operator::v2::macros::attributed_string_type::Error, }, + + #[snafu(display("failed to determine the cluster's name"))] + ResolveClusterName { + source: stackable_operator::v2::controller_utils::Error, + }, + + #[snafu(display("failed to get the metastore role Listener {listener_name}"))] + GetRoleListener { + source: stackable_operator::client::Error, + listener_name: String, + }, } /// External references resolved during the dereference step. pub struct DereferencedObjects { pub s3_connection_spec: Option, pub hive_opa_config: Option, + /// The metastore role [`Listener`] as currently stored in the cluster, fetched because the + /// discovery `ConfigMap` is built from its ingress address. Unlike the other fields it is not + /// referenced from the spec but created by this operator itself: `None` on the first + /// reconcile run (the apply step has not created it yet), and its status is only populated + /// asynchronously by the listener-operator, so it can still be address-less here. + pub role_listener: Option, } /// OPA settings resolved from the cluster's OPA reference during the dereference step. @@ -105,8 +128,19 @@ pub async fn dereference( None => None, }; + let cluster_name = get_cluster_name(hive).context(ResolveClusterNameSnafu)?; + let namespace = get_namespace(hive).context(ResolveNamespaceSnafu)?; + let listener_name = role_listener_name(&cluster_name, &HiveRole::MetaStore); + let role_listener = client + .get_opt::(listener_name.as_ref(), namespace.as_ref()) + .await + .context(GetRoleListenerSnafu { + listener_name: listener_name.as_ref(), + })?; + Ok(DereferencedObjects { s3_connection_spec, hive_opa_config, + role_listener, }) } diff --git a/rust/operator-binary/src/controller/update_status.rs b/rust/operator-binary/src/controller/update_status.rs index 3e5bdbe1..a4092599 100644 --- a/rust/operator-binary/src/controller/update_status.rs +++ b/rust/operator-binary/src/controller/update_status.rs @@ -1,12 +1,17 @@ //! The update_status step in the HiveCluster controller. +use std::hash::Hasher; + +use fnv::FnvHasher; use snafu::{ResultExt, Snafu}; use stackable_operator::{ client::Client, + k8s_openapi::api::core::v1::ConfigMap, status::condition::{ compute_conditions, operations::ClusterOperationsConditionBuilder, statefulset::StatefulSetConditionBuilder, }, + v2::{controller_utils::get_cluster_name, types::operator::ClusterName}, }; use strum::{EnumDiscriminants, IntoStaticStr}; @@ -23,19 +28,22 @@ pub enum Error { ApplyStatus { source: stackable_operator::client::Error, }, + + #[snafu(display("failed to determine the cluster's name"))] + ResolveClusterName { + source: stackable_operator::v2::controller_utils::Error, + }, } type Result = std::result::Result; /// Computes the cluster status from the applied resources and patches it onto the /// [`v1alpha1::HiveCluster`]. Takes [`KubernetesResources`] so the type system proves -/// the status derives from applied resources, not merely built ones. `discovery_hash` is -/// derived from the applied discovery `ConfigMap`'s resource version in the reconcile step. +/// the status derives from applied resources, not merely built ones. pub async fn update_status( client: &Client, hive: &v1alpha1::HiveCluster, applied: &KubernetesResources, - discovery_hash: u64, ) -> Result<()> { let mut ss_cond_builder = StatefulSetConditionBuilder::default(); for stateful_set in &applied.stateful_sets { @@ -45,10 +53,12 @@ pub async fn update_status( let cluster_operation_cond_builder = ClusterOperationsConditionBuilder::new(&hive.spec.cluster_operation); + let cluster_name = get_cluster_name(hive).context(ResolveClusterNameSnafu)?; + let status = HiveClusterStatus { // Serialize as a string to discourage users from trying to parse the value, // and to keep things flexible if we end up changing the hasher at some point. - discovery_hash: Some(discovery_hash.to_string()), + discovery_hash: Some(discovery_hash(&applied.config_maps, &cluster_name).to_string()), conditions: compute_conditions(hive, &[&ss_cond_builder, &cluster_operation_cond_builder]), }; @@ -59,3 +69,69 @@ pub async fn update_status( Ok(()) } + +/// The hash of the applied discovery `ConfigMap`'s (named after the cluster) resource version, +/// exposed in the status so that dependent clusters can restart on discovery changes. While no +/// discovery ConfigMap has been applied (the role Listener has no ingress address yet), nothing +/// is hashed and the hasher's initial state is returned. +fn discovery_hash(config_maps: &[ConfigMap], cluster_name: &ClusterName) -> u64 { + // std's SipHasher is deprecated, and DefaultHasher is unstable across Rust releases. + // We don't /need/ stability, but it's still nice to avoid spurious changes where possible. + let mut hasher = FnvHasher::with_key(0); + if let Some(resource_version) = config_maps + .iter() + .find(|config_map| config_map.metadata.name.as_deref() == Some(cluster_name.as_ref())) + .and_then(|config_map| config_map.metadata.resource_version.as_ref()) + { + hasher.write(resource_version.as_bytes()); + } + hasher.finish() +} + +#[cfg(test)] +mod tests { + use std::str::FromStr; + + use stackable_operator::{ + k8s_openapi::api::core::v1::ConfigMap, kube::api::ObjectMeta, + v2::types::operator::ClusterName, + }; + + use super::discovery_hash; + + fn config_map(name: &str, resource_version: &str) -> ConfigMap { + ConfigMap { + metadata: ObjectMeta { + name: Some(name.to_string()), + resource_version: Some(resource_version.to_string()), + ..ObjectMeta::default() + }, + ..ConfigMap::default() + } + } + + /// The hash must react to the discovery ConfigMap (named after the cluster) and ignore the + /// role-group ConfigMaps. + #[test] + fn discovery_hash_tracks_only_the_discovery_config_map() { + let cluster_name = ClusterName::from_str("simple-hive").expect("valid cluster name"); + let role_group_cm = config_map("simple-hive-metastore-default", "1"); + + let without_discovery_cm = + discovery_hash(std::slice::from_ref(&role_group_cm), &cluster_name); + let with_discovery_cm = discovery_hash( + &[role_group_cm, config_map("simple-hive", "42")], + &cluster_name, + ); + assert_ne!(without_discovery_cm, with_discovery_cm); + + // A changed resource version changes the hash. + let with_changed_discovery_cm = + discovery_hash(&[config_map("simple-hive", "43")], &cluster_name); + assert_ne!(with_discovery_cm, with_changed_discovery_cm); + + // An absent discovery ConfigMap hashes to the hasher's initial state, matching the + // behaviour before it was skipped (no bytes were written either). + assert_eq!(without_discovery_cm, discovery_hash(&[], &cluster_name)); + } +} diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs index 73ea52ea..da73519d 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -233,6 +233,7 @@ pub fn validate_cluster( kerberos_secret_class, }, role_group_configs, + dereferenced_objects.role_listener, )) } @@ -296,6 +297,7 @@ mod tests { DereferencedObjects { s3_connection_spec: None, hive_opa_config: None, + role_listener: None, }, ) } diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs index e4d6b461..68b5f11c 100644 --- a/rust/operator-binary/src/main.rs +++ b/rust/operator-binary/src/main.rs @@ -10,6 +10,7 @@ use futures::{FutureExt, StreamExt, TryFutureExt}; use stackable_operator::{ YamlSchema, cli::{Command, RunArguments}, + crd::listener::v1alpha1::Listener, eos::EndOfSupportChecker, k8s_openapi::api::{ apps::v1::StatefulSet, @@ -135,6 +136,14 @@ async fn main() -> anyhow::Result<()> { watch_namespace.get_api::(&client), watcher::Config::default(), ) + // The role Listener is created by this operator, but its ingress address (from + // which the discovery ConfigMap is built) is only written asynchronously by the + // listener-operator -- this watch triggers the reconcile run that builds the + // discovery ConfigMap once the address is set. + .owns( + watch_namespace.get_api::(&client), + watcher::Config::default(), + ) .watches( watch_namespace.get_api::>(&client), watcher::Config::default(), From f5b753e94be74868b016cc85dc08f58ebcd6ee39 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Fri, 7 Aug 2026 17:26:00 +0200 Subject: [PATCH 3/3] extract listeneringress and re-use --- .../src/controller/build/mod.rs | 65 ++++++++----------- .../controller/build/resource/discovery.rs | 40 +++++++++--- .../src/controller/build/resource/listener.rs | 12 +--- 3 files changed, 58 insertions(+), 59 deletions(-) diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 26856a21..2e429f02 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -113,28 +113,10 @@ pub fn build( } } - // The discovery ConfigMap needs the role Listener's ingress address, which only the - // listener-operator writes. Around the first reconcile runs the dereferenced Listener is - // absent or still address-less; the ConfigMap is skipped then instead of failing the whole - // run -- the Listener watch triggers a new run once the address is set. In that window an - // already existing discovery ConfigMap is deleted as an orphan (only reachable when the - // Listener is deleted and re-created) and re-created by the next run. - if let Some(role_listener) = &cluster.role_listener - && role_listener - .status - .as_ref() - .and_then(|status| status.ingress_addresses.as_ref()?.first()) - .is_some() + if let Some(discovery_config_map) = + build_discovery_configmap(cluster, HiveRole::MetaStore).context(DiscoveryConfigMapSnafu)? { - config_maps.push( - build_discovery_configmap(cluster, HiveRole::MetaStore, role_listener) - .context(DiscoveryConfigMapSnafu)?, - ); - } else { - tracing::debug!( - "the metastore role Listener has no ingress address yet, skipping the discovery \ - ConfigMap" - ); + config_maps.push(discovery_config_map); } Ok(KubernetesResources { @@ -272,7 +254,7 @@ mod tests { } #[test] - fn build_adds_the_discovery_config_map_once_the_role_listener_has_an_address() { + fn builds_discovery_config_map_with_listener_address() { let hive = minimal_hive(DERBY_YAML); let mut cluster = validated_cluster(&hive); cluster.role_listener = Some(role_listener_with_address()); @@ -290,28 +272,33 @@ mod tests { ); } - /// While the Listener carries no ingress address (the listener-operator has not reconciled - /// it yet), the discovery ConfigMap is skipped, *without* failing the build: the Listener - /// watch triggers a new reconcile run once the address is set. + /// While the Listener is absent (the apply step has not created it yet) or carries no + /// ingress address (the listener-operator has not reconciled it yet), the discovery + /// ConfigMap is skipped, *without* failing the build: the Listener watch triggers a new + /// reconcile run once the address is set. #[test] - fn build_skips_the_discovery_config_map_while_the_role_listener_has_no_address() { + fn skips_discovery_config_map_without_listener_address() { let hive = minimal_hive(DERBY_YAML); let mut cluster = validated_cluster(&hive); - let no_status = None; - let no_addresses = Some(listener::v1alpha1::ListenerStatus { - service_name: None, - ingress_addresses: Some(vec![]), - node_ports: None, + let no_listener = None; + let no_status = Some(Listener { + status: None, + ..role_listener_with_address() }); - for status in [no_status, no_addresses] { - cluster.role_listener = Some(Listener { - status, - ..role_listener_with_address() - }); - - let resources = - build(&cluster, &test_cluster_info()).expect("build succeeds without an address"); + let no_addresses = Some(Listener { + status: Some(listener::v1alpha1::ListenerStatus { + service_name: None, + ingress_addresses: Some(vec![]), + node_ports: None, + }), + ..role_listener_with_address() + }); + for role_listener in [no_listener, no_status, no_addresses] { + cluster.role_listener = role_listener; + + let resources = build(&cluster, &test_cluster_info()) + .expect("build succeeds without a listener address"); assert!(discovery_config_map(&resources).is_none()); } diff --git a/rust/operator-binary/src/controller/build/resource/discovery.rs b/rust/operator-binary/src/controller/build/resource/discovery.rs index 0756daf4..d4e7b981 100644 --- a/rust/operator-binary/src/controller/build/resource/discovery.rs +++ b/rust/operator-binary/src/controller/build/resource/discovery.rs @@ -1,7 +1,7 @@ use snafu::{ResultExt, Snafu}; use stackable_operator::{ - builder::configmap::ConfigMapBuilder, crd::listener::v1alpha1::Listener, - k8s_openapi::api::core::v1::ConfigMap, kube::runtime::reflector::ObjectRef, + builder::configmap::ConfigMapBuilder, k8s_openapi::api::core::v1::ConfigMap, + kube::runtime::reflector::ObjectRef, }; use crate::{ @@ -35,15 +35,33 @@ fn cluster_object_ref(cluster: &ValidatedCluster) -> ObjectRef Result { +) -> Result, Error> { + let Some(listener_address) = cluster + .role_listener + .as_ref() + .and_then(|listener| listener.status.as_ref()) + .and_then(|status| status.ingress_addresses.as_ref()?.first()) + else { + tracing::debug!( + "the metastore role Listener has no ingress address yet, \ + skipping the discovery ConfigMap" + ); + return Ok(None); + }; + let mut discovery_configmap = ConfigMapBuilder::new(); discovery_configmap.metadata( @@ -60,13 +78,15 @@ pub fn build_discovery_configmap( discovery_configmap.add_data( "HIVE".to_string(), - build_listener_connection_string(listener, &hive_role.to_string()) + build_listener_connection_string(listener_address, &hive_role.to_string()) .context(ListenerConfigurationSnafu)?, ); - discovery_configmap + let config_map = discovery_configmap .build() .with_context(|_| DiscoveryConfigMapSnafu { obj_ref: cluster_object_ref(cluster), - }) + })?; + + Ok(Some(config_map)) } diff --git a/rust/operator-binary/src/controller/build/resource/listener.rs b/rust/operator-binary/src/controller/build/resource/listener.rs index 7940f73e..87e57bc8 100644 --- a/rust/operator-binary/src/controller/build/resource/listener.rs +++ b/rust/operator-binary/src/controller/build/resource/listener.rs @@ -2,7 +2,7 @@ use std::str::FromStr; use snafu::{OptionExt, Snafu}; use stackable_operator::{ - crd::listener::v1alpha1::{Listener, ListenerPort, ListenerSpec}, + crd::listener::v1alpha1::{Listener, ListenerIngress, ListenerPort, ListenerSpec}, v2::types::{ kubernetes::{ListenerClassName, ListenerName}, operator::ClusterName, @@ -19,23 +19,15 @@ use crate::{ #[derive(Snafu, Debug)] pub enum Error { - #[snafu(display("{role} listener has no address"))] - RoleListenerHasNoAddress { role: String }, #[snafu(display("could not find port [{port_name}] for rolegroup listener {role}"))] NoServicePort { port_name: String, role: String }, } // Builds the connection string with respect to the listener provided objects pub fn build_listener_connection_string( - listener_ref: &Listener, + listener_address: &ListenerIngress, role: &str, ) -> Result { - // We only need the first address corresponding to the role - let listener_address = listener_ref - .status - .as_ref() - .and_then(|status| status.ingress_addresses.as_ref()?.first()) - .context(RoleListenerHasNoAddressSnafu { role })?; let conn_str = format!( "thrift://{address}:{port}", address = listener_address.address,