From aee1646bc59f9c74ade4cf6351b917bf9b7e1db0 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Fri, 31 Jul 2026 16:29:14 +0200 Subject: [PATCH] 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(()) +}