From b51391925dddebc3a8aa136e204400c26f852265 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Fri, 24 Jul 2026 16:46:51 +0200 Subject: [PATCH 1/5] refactor: extract apply step (Applier) and move secret creation into it --- .../operator-binary/src/airflow_controller.rs | 150 +++------------ rust/operator-binary/src/controller/apply.rs | 171 ++++++++++++++++++ .../src/controller/build/mod.rs | 7 +- rust/operator-binary/src/controller/mod.rs | 14 +- 4 files changed, 210 insertions(+), 132 deletions(-) create mode 100644 rust/operator-binary/src/controller/apply.rs diff --git a/rust/operator-binary/src/airflow_controller.rs b/rust/operator-binary/src/airflow_controller.rs index 615e18fe..0e52a5b6 100644 --- a/rust/operator-binary/src/airflow_controller.rs +++ b/rust/operator-binary/src/airflow_controller.rs @@ -9,7 +9,6 @@ use snafu::{ResultExt, Snafu}; use stackable_operator::{ cli::OperatorEnvironmentOptions, cluster_resources::ClusterResourceApplyStrategy, - commons::random_secret_creation, k8s_openapi::api::core::v1::EnvVar, kube::{ core::{DeserializeGuard, error_boundary}, @@ -21,19 +20,15 @@ use stackable_operator::{ compute_conditions, operations::ClusterOperationsConditionBuilder, statefulset::StatefulSetConditionBuilder, }, - v2::cluster_resources::cluster_resources_new, }; use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ - controller::{ValidatedCluster, build, controller_name, operator_name, product_name}, - crd::{ - AirflowClusterStatus, OPERATOR_NAME, - internal_secret::{ - FERNET_KEY_SECRET_KEY, INTERNAL_SECRET_SECRET_KEY, JWT_SECRET_SECRET_KEY, - }, - v1alpha2, + controller::{ + apply::{self, Applier, ensure_random_secrets}, + build, }, + crd::{AirflowClusterStatus, OPERATOR_NAME, v1alpha2}, }; pub const AIRFLOW_CONTROLLER_NAME: &str = "airflowcluster"; @@ -51,29 +46,20 @@ pub struct Ctx { #[strum_discriminants(derive(IntoStaticStr))] #[snafu(visibility(pub(crate)))] pub enum Error { - #[snafu(display("failed to apply Kubernetes resource"))] - ApplyResource { - source: stackable_operator::cluster_resources::Error, - }, + #[snafu(display("failed to apply the Kubernetes resources"))] + ApplyResources { source: apply::Error }, #[snafu(display("failed to build the Kubernetes resources"))] BuildResources { source: build::Error }, - #[snafu(display("failed to delete orphaned resources"))] - DeleteOrphanedResources { - source: stackable_operator::cluster_resources::Error, - }, + #[snafu(display("failed to ensure the shared random Secrets exist"))] + EnsureSecrets { source: apply::Error }, #[snafu(display("failed to update status"))] ApplyStatus { source: stackable_operator::client::Error, }, - #[snafu(display("failed to create internal secret"))] - InternalSecret { - source: random_secret_creation::Error, - }, - #[snafu(display("failed to dereference cluster resources"))] Dereference { source: crate::controller::dereference::Error, @@ -126,75 +112,25 @@ pub async fn reconcile_airflow( ) .context(ValidateSnafu)?; - ensure_random_secrets(client, &validated_cluster).await?; + let resources = build::build(&validated_cluster).context(BuildResourcesSnafu)?; - let mut cluster_resources = cluster_resources_new( - &product_name(), - &operator_name(), - &controller_name(), - &validated_cluster.name, - &validated_cluster.namespace, - &validated_cluster.uid, + ensure_random_secrets(client, &validated_cluster) + .await + .context(EnsureSecretsSnafu)?; + let applied = Applier::new( + client, + &validated_cluster, ClusterResourceApplyStrategy::from(&airflow.spec.cluster_operation), &airflow.spec.object_overrides, - ); - - let resources = build::build(&validated_cluster).context(BuildResourcesSnafu)?; + ) + .apply(resources) + .await + .context(ApplyResourcesSnafu)?; let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - - // 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. - 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)?; - } - for listener in resources.listeners { - 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)?; + for statefulset in applied.stateful_sets { + ss_cond_builder.add(statefulset); } - for statefulset in resources.stateful_sets { - ss_cond_builder.add( - cluster_resources - .add(client, statefulset) - .await - .context(ApplyResourceSnafu)?, - ); - } - - cluster_resources - .delete_orphaned_resources(client) - .await - .context(DeleteOrphanedResourcesSnafu)?; let status = AirflowClusterStatus { conditions: compute_conditions( @@ -211,50 +147,6 @@ pub async fn reconcile_airflow( Ok(Action::await_change()) } -/// Ensures the three shared random Secrets (internal / JWT / Fernet) exist, creating any that are -/// missing. These are read-or-create client operations, so they cannot be part of the client-free -/// `build()` step. -async fn ensure_random_secrets( - client: &stackable_operator::client::Client, - cluster: &ValidatedCluster, -) -> Result<(), Error> { - random_secret_creation::create_random_secret_if_not_exists( - cluster.internal_secret_name().as_ref(), - INTERNAL_SECRET_SECRET_KEY, - 256, - cluster, - client, - ) - .await - .context(InternalSecretSnafu)?; - - random_secret_creation::create_random_secret_if_not_exists( - cluster.jwt_secret_name().as_ref(), - JWT_SECRET_SECRET_KEY, - 256, - cluster, - client, - ) - .await - .context(InternalSecretSnafu)?; - - // https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/fernet.html#security-fernet - // does not document how long the fernet key should be, but recommends using - // python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" - // which returns 32 bytes. - random_secret_creation::create_random_secret_if_not_exists( - cluster.fernet_key_name().as_ref(), - FERNET_KEY_SECRET_KEY, - 32, - cluster, - client, - ) - .await - .context(InternalSecretSnafu)?; - - Ok(()) -} - pub fn error_policy( _obj: Arc>, error: &Error, diff --git a/rust/operator-binary/src/controller/apply.rs b/rust/operator-binary/src/controller/apply.rs new file mode 100644 index 00000000..c3be9c58 --- /dev/null +++ b/rust/operator-binary/src/controller/apply.rs @@ -0,0 +1,171 @@ +//! The apply step in the AirflowCluster controller. + +use std::marker::PhantomData; + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + cluster_resources::{ClusterResource, ClusterResourceApplyStrategy, ClusterResources}, + commons::random_secret_creation, + deep_merger::ObjectOverrides, + v2::cluster_resources::cluster_resources_new, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::{ + controller::{ + Applied, KubernetesResources, Prepared, ValidatedCluster, controller_name, operator_name, + product_name, + }, + crd::internal_secret::{ + FERNET_KEY_SECRET_KEY, INTERNAL_SECRET_SECRET_KEY, JWT_SECRET_SECRET_KEY, + }, +}; + +#[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, + }, + + #[snafu(display("failed to create internal secret"))] + InternalSecret { + source: random_secret_creation::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. + pub async fn apply( + mut self, + resources: KubernetesResources, + ) -> Result> { + // 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(resources.service_accounts).await?; + let role_bindings = self.add_resources(resources.role_bindings).await?; + let services = self.add_resources(resources.services).await?; + let listeners = self.add_resources(resources.listeners).await?; + let config_maps = self.add_resources(resources.config_maps).await?; + let pod_disruption_budgets = self.add_resources(resources.pod_disruption_budgets).await?; + let stateful_sets = self.add_resources(resources.stateful_sets).await?; + + self.cluster_resources + .delete_orphaned_resources(self.client) + .await + .context(DeleteOrphanedResourcesSnafu)?; + + Ok(KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + pod_disruption_budgets, + service_accounts, + role_bindings, + status: PhantomData, + }) + } + + 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) + } +} + +/// Ensures the three shared random Secrets (internal / JWT / Fernet) exist, creating any that are +/// missing. These are read-or-create client operations, so they cannot be part of the client-free +/// `build()` step; they are also deliberately not tracked in [`ClusterResources`], so they survive +/// orphan deletion and an existing Secret is never overwritten. +pub async fn ensure_random_secrets(client: &Client, cluster: &ValidatedCluster) -> Result<()> { + random_secret_creation::create_random_secret_if_not_exists( + cluster.internal_secret_name().as_ref(), + INTERNAL_SECRET_SECRET_KEY, + 256, + cluster, + client, + ) + .await + .context(InternalSecretSnafu)?; + + random_secret_creation::create_random_secret_if_not_exists( + cluster.jwt_secret_name().as_ref(), + JWT_SECRET_SECRET_KEY, + 256, + cluster, + client, + ) + .await + .context(InternalSecretSnafu)?; + + // https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/fernet.html#security-fernet + // does not document how long the fernet key should be, but recommends using + // python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" + // which returns 32 bytes. + random_secret_creation::create_random_secret_if_not_exists( + cluster.fernet_key_name().as_ref(), + FERNET_KEY_SECRET_KEY, + 32, + cluster, + client, + ) + .await + .context(InternalSecretSnafu)?; + + Ok(()) +} diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 65d51626..63f67cb6 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -1,11 +1,13 @@ //! Builders that assemble Kubernetes resources from the validated cluster. +use std::marker::PhantomData; + use snafu::{ResultExt, Snafu}; use stackable_operator::v2::types::operator::RoleGroupName; use crate::{ controller::{ - KubernetesResources, ValidatedCluster, + KubernetesResources, Prepared, ValidatedCluster, build::resource::{ config_map, executor::build_executor_template_config_map, @@ -48,7 +50,7 @@ pub enum Error { /// Does not need a Kubernetes client: every reference to another Kubernetes resource is already /// dereferenced and validated by this point. Cluster configuration is likewise already validated, /// so the errors returned here are resource-assembly failures only. -pub fn build(cluster: &ValidatedCluster) -> Result { +pub fn build(cluster: &ValidatedCluster) -> Result, Error> { let mut stateful_sets = vec![]; let mut services = vec![]; let mut listeners = vec![]; @@ -149,6 +151,7 @@ pub fn build(cluster: &ValidatedCluster) -> Result { 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/mod.rs b/rust/operator-binary/src/controller/mod.rs index ea3eddab..2aba0ca8 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -1,5 +1,6 @@ use std::{ collections::{BTreeMap, HashMap}, + marker::PhantomData, str::FromStr, }; @@ -61,6 +62,7 @@ use crate::{ }, }; +pub mod apply; pub mod build; pub mod dereference; pub mod validate; @@ -68,8 +70,17 @@ pub mod validate; // Placeholder version label value for resources whose labels must not change after deployment. stackable_operator::constant!(UNVERSIONED_PRODUCT_VERSION: ProductVersion = "none"); +/// 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 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, @@ -77,6 +88,7 @@ pub struct KubernetesResources { pub pod_disruption_budgets: Vec, pub service_accounts: Vec, pub role_bindings: Vec, + pub status: PhantomData, } /// Per-role configuration extracted during validation. From dede2c0870a026ea8c76c9cc816f384a60519327 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Fri, 24 Jul 2026 16:56:24 +0200 Subject: [PATCH 2/5] refactor: extract update_status step --- .../operator-binary/src/airflow_controller.rs | 33 ++--------- rust/operator-binary/src/controller/mod.rs | 1 + .../src/controller/update_status.rs | 58 +++++++++++++++++++ 3 files changed, 65 insertions(+), 27 deletions(-) create mode 100644 rust/operator-binary/src/controller/update_status.rs diff --git a/rust/operator-binary/src/airflow_controller.rs b/rust/operator-binary/src/airflow_controller.rs index 0e52a5b6..cbe4af8f 100644 --- a/rust/operator-binary/src/airflow_controller.rs +++ b/rust/operator-binary/src/airflow_controller.rs @@ -16,10 +16,6 @@ use stackable_operator::{ }, logging::controller::ReconcilerError, shared::time::Duration, - status::condition::{ - compute_conditions, operations::ClusterOperationsConditionBuilder, - statefulset::StatefulSetConditionBuilder, - }, }; use strum::{EnumDiscriminants, IntoStaticStr}; @@ -27,8 +23,9 @@ use crate::{ controller::{ apply::{self, Applier, ensure_random_secrets}, build, + update_status::{self, update_status}, }, - crd::{AirflowClusterStatus, OPERATOR_NAME, v1alpha2}, + crd::{OPERATOR_NAME, v1alpha2}, }; pub const AIRFLOW_CONTROLLER_NAME: &str = "airflowcluster"; @@ -55,10 +52,8 @@ pub enum Error { #[snafu(display("failed to ensure the shared random Secrets exist"))] EnsureSecrets { source: apply::Error }, - #[snafu(display("failed to update status"))] - ApplyStatus { - source: stackable_operator::client::Error, - }, + #[snafu(display("failed to update the cluster status"))] + UpdateStatus { source: update_status::Error }, #[snafu(display("failed to dereference cluster resources"))] Dereference { @@ -102,9 +97,6 @@ pub async fn reconcile_airflow( .await .context(DereferenceSnafu)?; - let cluster_operation_cond_builder = - ClusterOperationsConditionBuilder::new(&airflow.spec.cluster_operation); - let validated_cluster = crate::controller::validate::validate_cluster( airflow, &ctx.operator_environment.image_repository, @@ -127,22 +119,9 @@ pub async fn reconcile_airflow( .await .context(ApplyResourcesSnafu)?; - let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - for statefulset in applied.stateful_sets { - ss_cond_builder.add(statefulset); - } - - let status = AirflowClusterStatus { - conditions: compute_conditions( - airflow, - &[&ss_cond_builder, &cluster_operation_cond_builder], - ), - }; - - client - .apply_patch_status(OPERATOR_NAME, airflow, &status) + update_status(client, airflow, &applied) .await - .context(ApplyStatusSnafu)?; + .context(UpdateStatusSnafu)?; Ok(Action::await_change()) } diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs index 2aba0ca8..956c539f 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -65,6 +65,7 @@ use crate::{ pub mod apply; pub mod build; pub mod dereference; +pub mod update_status; pub mod validate; // Placeholder version label value for resources whose labels must not change after deployment. 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..ddc961a8 --- /dev/null +++ b/rust/operator-binary/src/controller/update_status.rs @@ -0,0 +1,58 @@ +//! The update_status step in the AirflowCluster controller. + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + status::condition::{ + compute_conditions, operations::ClusterOperationsConditionBuilder, + statefulset::StatefulSetConditionBuilder, + }, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::{ + controller::{Applied, KubernetesResources}, + crd::{AirflowClusterStatus, OPERATOR_NAME, v1alpha2}, +}; + +#[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 +/// [`v1alpha2::AirflowCluster`]. Takes [`KubernetesResources`] so the type system +/// proves the status derives from applied resources, not merely built ones. +pub async fn update_status( + client: &Client, + airflow: &v1alpha2::AirflowCluster, + applied: &KubernetesResources, +) -> 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(&airflow.spec.cluster_operation); + + let status = AirflowClusterStatus { + conditions: compute_conditions( + airflow, + &[&ss_cond_builder, &cluster_operation_cond_builder], + ), + }; + + client + .apply_patch_status(OPERATOR_NAME, airflow, &status) + .await + .context(ApplyStatusSnafu)?; + + Ok(()) +} From 6ab1a55ea5487e2f85a681bfcf06c52032a90490 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Mon, 27 Jul 2026 16:24:33 +0200 Subject: [PATCH 3/5] use app_version_label to ensure labels are release/version-specific --- rust/operator-binary/src/controller/build/mod.rs | 5 +++-- rust/operator-binary/src/controller/mod.rs | 13 +++++++++++++ 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 63f67cb6..275996ee 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -164,7 +164,8 @@ mod tests { use super::build; use crate::{ controller::{ - ValidatedCluster, dereference::DereferencedObjects, validate::validate_cluster, + ValidatedCluster, app_version_label, dereference::DereferencedObjects, + validate::validate_cluster, }, crd::{ authentication::{AirflowClientAuthenticationDetailsResolved, FlaskRolesSyncMoment}, @@ -320,7 +321,7 @@ mod tests { ), ("app.kubernetes.io/name", "airflow"), ("app.kubernetes.io/role-group", "none"), - ("app.kubernetes.io/version", "3.1.6-stackable0.0.0-dev"), + ("app.kubernetes.io/version", &app_version_label("3.1.6")), ("stackable.tech/vendor", "Stackable"), ] .map(|(key, value)| (key.to_string(), value.to_string())), diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs index 956c539f..5e1dd388 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -62,6 +62,19 @@ use crate::{ }, }; +/// The expected `app.kubernetes.io/version` label value for the given product version. +/// +/// The `-stackable` suffix carries the operator's own version, which is `0.0.0-dev` on main +/// but rewritten by the release process — so tests must derive it rather than hardcode it, +/// or they fail on release branches. +#[cfg(test)] +pub(crate) fn app_version_label(product_version: &str) -> String { + format!( + "{product_version}-stackable{}", + crate::built_info::PKG_VERSION + ) +} + pub mod apply; pub mod build; pub mod dereference; From 36494ee3e72b144aff592a3e1e2ca69f7bb60ec5 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Wed, 29 Jul 2026 17:48:58 +0200 Subject: [PATCH 4/5] refactor: move object_meta into the build step Functions which build Kubernetes resources or parts of them belong in the build step, so object_meta becomes a free function in build (the same shape the other operators settled on after review). Also relocates the app_version_label test fixture next to its consumers and imports build_rolegroup_config_map as an item for consistency. (cherry picked from commit 76187c3) --- .../src/controller/build/mod.rs | 71 +++++++++++++++---- .../controller/build/resource/config_map.rs | 29 ++++---- .../src/controller/build/resource/executor.rs | 21 +++--- .../src/controller/build/resource/listener.rs | 18 +++-- .../src/controller/build/resource/service.rs | 52 +++++++------- .../controller/build/resource/statefulset.rs | 12 ++-- rust/operator-binary/src/controller/mod.rs | 27 ------- 7 files changed, 125 insertions(+), 105 deletions(-) diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 275996ee..b41f4dc4 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -3,13 +3,17 @@ use std::marker::PhantomData; use snafu::{ResultExt, Snafu}; -use stackable_operator::v2::types::operator::RoleGroupName; +use stackable_operator::{ + builder::meta::ObjectMetaBuilder, + kvp::Labels, + v2::{builder::meta::ownerreference_from_resource, types::operator::RoleGroupName}, +}; use crate::{ controller::{ KubernetesResources, Prepared, ValidatedCluster, build::resource::{ - config_map, + config_map::build_rolegroup_config_map, executor::build_executor_template_config_map, listener::build_group_listener, pdb::build_pdb, @@ -61,7 +65,7 @@ pub fn build(cluster: &ValidatedCluster) -> Result // executor's workers are a regular role with its own role groups instead). if let Some(executor_template) = &cluster.cluster_config.executor_template { let executor_role_group = executor_role_group_name(); - let executor_config_map = config_map::build_rolegroup_config_map( + let executor_config_map = build_rolegroup_config_map( cluster, &executor_role_name(), &executor_role_group, @@ -116,7 +120,7 @@ pub fn build(cluster: &ValidatedCluster) -> Result role_group_name, )); config_maps.push( - config_map::build_rolegroup_config_map( + build_rolegroup_config_map( cluster, &ValidatedCluster::role_name(role), role_group_name, @@ -155,17 +159,30 @@ pub fn build(cluster: &ValidatedCluster) -> Result }) } -#[cfg(test)] -mod tests { - use std::collections::BTreeMap; - - use stackable_operator::kube::Resource; +/// Returns an [`ObjectMetaBuilder`] pre-filled with the cluster's namespace, the resource +/// `name`, an owner reference back to the cluster, and the given recommended `labels`. +/// +/// Consolidates the metadata chain repeated by the child-resource builders. Call sites that +/// need extra labels/annotations chain them onto the returned builder. +pub(crate) fn object_meta( + cluster: &ValidatedCluster, + name: impl Into, + labels: Labels, +) -> ObjectMetaBuilder { + let mut builder = ObjectMetaBuilder::new(); + builder + .name_and_namespace(cluster) + .name(name) + .ownerreference(ownerreference_from_resource(cluster, None, Some(true))) + .with_labels(labels); + builder +} - use super::build; +#[cfg(test)] +pub(crate) mod test_support { use crate::{ controller::{ - ValidatedCluster, app_version_label, dereference::DereferencedObjects, - validate::validate_cluster, + ValidatedCluster, dereference::DereferencedObjects, validate::validate_cluster, }, crd::{ authentication::{AirflowClientAuthenticationDetailsResolved, FlaskRolesSyncMoment}, @@ -174,12 +191,24 @@ mod tests { }, }; + /// The expected `app.kubernetes.io/version` label value for the given product version. + /// + /// The `-stackable` suffix carries the operator's own version, which is `0.0.0-dev` on main + /// but rewritten by the release process — so tests must derive it rather than hardcode it, + /// or they fail on release branches. + pub fn app_version_label(product_version: &str) -> String { + format!( + "{product_version}-stackable{}", + crate::built_info::PKG_VERSION + ) + } + /// A validated cluster with default `webserver`/`scheduler` role groups and the given executor /// (its `spec` key plus config, as standalone YAML), built via `validate_cluster` from a /// minimal test CR (mirroring `validate::tests::test_cluster`), since `ValidatedCluster` /// carries several resolved types (git-sync resources, validated logging, …) that are /// impractical to construct by hand. - fn validated_cluster(executor_key: &str, executor_config: &str) -> ValidatedCluster { + pub fn validated_cluster(executor_key: &str, executor_config: &str) -> ValidatedCluster { let cluster_yaml = r#" apiVersion: airflow.stackable.tech/v1alpha2 kind: AirflowCluster @@ -241,15 +270,27 @@ mod tests { /// Validated cluster with a Celery executor (its workers are provisioned via the queue, so no /// executor pod template is built). - fn celery_executor_cluster() -> ValidatedCluster { + pub fn celery_executor_cluster() -> ValidatedCluster { validated_cluster("celeryExecutors", "{config: {}, roleGroups: {}}") } /// Validated cluster with a Kubernetes executor, which builds an executor pod-template /// ConfigMap instead of a worker role. - fn kubernetes_executor_cluster() -> ValidatedCluster { + pub fn kubernetes_executor_cluster() -> ValidatedCluster { validated_cluster("kubernetesExecutors", "{config: {}}") } +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use stackable_operator::kube::Resource; + + use super::{ + build, + test_support::{app_version_label, celery_executor_cluster, kubernetes_executor_cluster}, + }; fn sorted_names(resources: &[impl Resource]) -> Vec<&str> { let mut names: Vec<&str> = resources diff --git a/rust/operator-binary/src/controller/build/resource/config_map.rs b/rust/operator-binary/src/controller/build/resource/config_map.rs index d4d10958..5b8dcacc 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -17,10 +17,13 @@ use stackable_operator::{ use crate::{ controller::{ ValidatedCluster, ValidatedLogging, - build::properties::{ - ConfigFileName, - product_logging::{create_airflow_config, vector_config_file_content}, - webserver_config, + build::{ + object_meta, + properties::{ + ConfigFileName, + product_logging::{create_airflow_config, vector_config_file_content}, + webserver_config, + }, }, }, crd::{AirflowConfigOverrides, Container}, @@ -63,15 +66,15 @@ pub fn build_rolegroup_config_map( cm_builder .metadata( - validated_cluster - .object_meta( - validated_cluster - .role_group_resource_names(role_name, role_group_name) - .role_group_config_map() - .to_string(), - validated_cluster.recommended_labels_for(role_name, role_group_name), - ) - .build(), + object_meta( + validated_cluster, + validated_cluster + .role_group_resource_names(role_name, role_group_name) + .role_group_config_map() + .to_string(), + validated_cluster.recommended_labels_for(role_name, role_group_name), + ) + .build(), ) .add_data(ConfigFileName::WebserverConfig.to_string(), config_file); diff --git a/rust/operator-binary/src/controller/build/resource/executor.rs b/rust/operator-binary/src/controller/build/resource/executor.rs index 3bcc320a..5b1a9673 100644 --- a/rust/operator-binary/src/controller/build/resource/executor.rs +++ b/rust/operator-binary/src/controller/build/resource/executor.rs @@ -26,6 +26,7 @@ use crate::{ ValidatedAirflowConfig, ValidatedCluster, build::{ graceful_shutdown::add_graceful_shutdown_config, + object_meta, properties::env_vars::build_airflow_template_envs, resource::pod::{ add_authentication_volumes_and_volume_mounts, add_git_sync_resources, @@ -184,16 +185,16 @@ pub fn build_executor_template_config_map( cm_builder .metadata( - cluster - .object_meta( - cluster.executor_template_configmap_name(), - cluster.recommended_labels_for( - &executor_role_name(), - &executor_template_role_group_name(), - ), - ) - .with_label(restarter_label) - .build(), + object_meta( + cluster, + cluster.executor_template_configmap_name(), + cluster.recommended_labels_for( + &executor_role_name(), + &executor_template_role_group_name(), + ), + ) + .with_label(restarter_label) + .build(), ) .add_data( TEMPLATE_NAME, diff --git a/rust/operator-binary/src/controller/build/resource/listener.rs b/rust/operator-binary/src/controller/build/resource/listener.rs index e42d3b25..e346eb25 100644 --- a/rust/operator-binary/src/controller/build/resource/listener.rs +++ b/rust/operator-binary/src/controller/build/resource/listener.rs @@ -9,7 +9,7 @@ use stackable_operator::{ }; use crate::{ - controller::ValidatedCluster, + controller::{ValidatedCluster, build::object_meta}, crd::{AirflowRole, HTTP_PORT, HTTP_PORT_NAME}, }; @@ -24,15 +24,13 @@ pub fn build_group_listener( listener_group_name: ListenerName, ) -> listener::v1alpha1::Listener { listener::v1alpha1::Listener { - metadata: cluster - .object_meta( - listener_group_name, - cluster.recommended_labels_for( - &ValidatedCluster::role_name(role), - &NONE_ROLE_GROUP_NAME, - ), - ) - .build(), + metadata: object_meta( + cluster, + listener_group_name, + cluster + .recommended_labels_for(&ValidatedCluster::role_name(role), &NONE_ROLE_GROUP_NAME), + ) + .build(), spec: listener::v1alpha1::ListenerSpec { class_name: Some(listener_class.to_string()), ports: Some(listener_ports()), diff --git a/rust/operator-binary/src/controller/build/resource/service.rs b/rust/operator-binary/src/controller/build/resource/service.rs index 4bba1114..6bab8c0c 100644 --- a/rust/operator-binary/src/controller/build/resource/service.rs +++ b/rust/operator-binary/src/controller/build/resource/service.rs @@ -7,7 +7,7 @@ use stackable_operator::{ }; use crate::{ - controller::ValidatedCluster, + controller::{ValidatedCluster, build::object_meta}, crd::{AirflowRole, HTTP_PORT, HTTP_PORT_NAME, METRICS_PORT, METRICS_PORT_NAME}, }; @@ -19,15 +19,15 @@ pub fn build_rolegroup_headless_service( role_group_name: &RoleGroupName, ) -> Service { Service { - metadata: cluster - .object_meta( - cluster - .role_group_resource_names(&ValidatedCluster::role_name(role), role_group_name) - .headless_service_name() - .to_string(), - cluster.recommended_labels(role, role_group_name), - ) - .build(), + metadata: object_meta( + cluster, + cluster + .role_group_resource_names(&ValidatedCluster::role_name(role), role_group_name) + .headless_service_name() + .to_string(), + cluster.recommended_labels(role, role_group_name), + ) + .build(), spec: Some(ServiceSpec { // Internal communication does not need to be exposed type_: Some("ClusterIP".to_string()), @@ -48,22 +48,22 @@ pub fn build_rolegroup_metrics_service( role_group_name: &RoleGroupName, ) -> Service { Service { - metadata: cluster - .object_meta( - cluster - .role_group_resource_names(&ValidatedCluster::role_name(role), role_group_name) - .metrics_service_name() - .to_string(), - cluster.recommended_labels(role, role_group_name), - ) - .with_labels(prometheus_labels(&Scraping::Enabled)) - .with_annotations(prometheus_annotations( - &Scraping::Enabled, - &Scheme::Http, - "/metrics", - &METRICS_PORT, - )) - .build(), + metadata: object_meta( + cluster, + cluster + .role_group_resource_names(&ValidatedCluster::role_name(role), role_group_name) + .metrics_service_name() + .to_string(), + cluster.recommended_labels(role, role_group_name), + ) + .with_labels(prometheus_labels(&Scraping::Enabled)) + .with_annotations(prometheus_annotations( + &Scraping::Enabled, + &Scheme::Http, + "/metrics", + &METRICS_PORT, + )) + .build(), spec: Some(ServiceSpec { // Internal communication does not need to be exposed type_: Some("ClusterIP".to_string()), diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 5e2aa8d8..205683df 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -33,6 +33,7 @@ use crate::{ AirflowRoleGroupConfig, ValidatedCluster, ValidatedLogging, build::{ graceful_shutdown::add_graceful_shutdown_config, + object_meta, properties::env_vars, resource::{ pod::{ @@ -90,10 +91,13 @@ fn build_rolegroup_metadata( prometheus_label: Label, name: String, ) -> ObjectMeta { - cluster - .object_meta(name, cluster.recommended_labels(role, role_group_name)) - .with_label(prometheus_label) - .build() + object_meta( + cluster, + name, + cluster.recommended_labels(role, role_group_name), + ) + .with_label(prometheus_label) + .build() } /// The rolegroup [`StatefulSet`] runs the rolegroup, as configured by the administrator. diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs index 5e1dd388..f4ebef5a 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -5,7 +5,6 @@ use std::{ }; use stackable_operator::{ - builder::meta::ObjectMetaBuilder, commons::{ affinity::StackableAffinity, product_image_selection::ResolvedProductImage, @@ -31,7 +30,6 @@ use stackable_operator::{ shared::time::Duration, v2::{ HasName, HasUid, NameIsValidLabelValue, - builder::meta::ownerreference_from_resource, kvp::label::{recommended_labels, role_group_selector}, product_logging::framework::{ValidatedContainerLogConfigChoice, VectorContainerLogConfig}, role_group_utils::ResourceNames, @@ -62,19 +60,6 @@ use crate::{ }, }; -/// The expected `app.kubernetes.io/version` label value for the given product version. -/// -/// The `-stackable` suffix carries the operator's own version, which is `0.0.0-dev` on main -/// but rewritten by the release process — so tests must derive it rather than hardcode it, -/// or they fail on release branches. -#[cfg(test)] -pub(crate) fn app_version_label(product_version: &str) -> String { - format!( - "{product_version}-stackable{}", - crate::built_info::PKG_VERSION - ) -} - pub mod apply; pub mod build; pub mod dereference; @@ -446,18 +431,6 @@ impl ValidatedCluster { role_group_name, ) } - - /// Returns an [`ObjectMetaBuilder`] pre-filled with the namespace, the resource `name`, an owner - /// reference back to this cluster, and the given recommended `labels`. - pub(crate) fn object_meta(&self, name: impl Into, labels: Labels) -> ObjectMetaBuilder { - let mut builder = ObjectMetaBuilder::new(); - builder - .name_and_namespace(self) - .name(name) - .ownerreference(ownerreference_from_resource(self, None, Some(true))) - .with_labels(labels); - builder - } } /// The product name (`airflow`) as a type-safe label value. From 4ae6f4b4040c493274cf746f7fcb5bf5a1566022 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Thu, 30 Jul 2026 11:30:29 +0200 Subject: [PATCH 5/5] changelog --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 365bf054..0e5c1a77 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,10 +9,13 @@ - The RBAC ServiceAccount and RoleBinding are now built with the operator-rs `v2::rbac` functions and carry the full set of recommended labels ([#821]). - Bump stackable-operator to 0.114.0 ([#827]). +- The reconciler now applies resources and derives the cluster status in discrete + apply and update_status steps ([#828]). [#814]: https://github.com/stackabletech/airflow-operator/pull/814 [#821]: https://github.com/stackabletech/airflow-operator/pull/821 [#827]: https://github.com/stackabletech/airflow-operator/pull/827 +[#828]: https://github.com/stackabletech/airflow-operator/pull/828 ## [26.7.0] - 2026-07-21