From bc5c6403b369462e69fc92574a6903cc368b4602 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Thu, 9 Jul 2026 16:39:33 +0200 Subject: [PATCH 1/3] refactor: Introduce build aggregator --- .../operator-binary/src/airflow_controller.rs | 336 +++++------------- .../src/controller/build/mod.rs | 299 ++++++++++++++++ .../src/controller/build/resource/executor.rs | 4 +- .../controller/build/resource/statefulset.rs | 8 +- rust/operator-binary/src/controller/mod.rs | 44 ++- 5 files changed, 431 insertions(+), 260 deletions(-) diff --git a/rust/operator-binary/src/airflow_controller.rs b/rust/operator-binary/src/airflow_controller.rs index 6cda6f0c..0c4bcb37 100644 --- a/rust/operator-binary/src/airflow_controller.rs +++ b/rust/operator-binary/src/airflow_controller.rs @@ -8,7 +8,7 @@ use const_format::concatcp; use snafu::{ResultExt, Snafu}; use stackable_operator::{ cli::OperatorEnvironmentOptions, - cluster_resources::{ClusterResourceApplyStrategy, ClusterResources}, + cluster_resources::ClusterResourceApplyStrategy, commons::{random_secret_creation, rbac::build_rbac_resources}, k8s_openapi::api::core::v1::EnvVar, kube::{ @@ -23,28 +23,14 @@ use stackable_operator::{ compute_conditions, operations::ClusterOperationsConditionBuilder, statefulset::StatefulSetConditionBuilder, }, - v2::{ - cluster_resources::cluster_resources_new, - types::operator::{RoleGroupName, RoleName}, - }, + v2::cluster_resources::cluster_resources_new, }; use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ - controller::{ - ValidatedCluster, ValidatedExecutorTemplate, - build::resource::{ - config_map, - executor::build_executor_template_config_map, - listener::build_group_listener, - pdb::build_pdb, - service::{build_rolegroup_headless_service, build_rolegroup_metrics_service}, - statefulset::build_server_rolegroup_statefulset, - }, - controller_name, operator_name, product_name, - }, + controller::{ValidatedCluster, build, controller_name, operator_name, product_name}, crd::{ - APP_NAME, AirflowClusterStatus, AirflowConfigOverrides, Container, OPERATOR_NAME, + APP_NAME, AirflowClusterStatus, OPERATOR_NAME, internal_secret::{ FERNET_KEY_SECRET_KEY, INTERNAL_SECRET_SECRET_KEY, JWT_SECRET_SECRET_KEY, }, @@ -55,32 +41,6 @@ use crate::{ pub const AIRFLOW_CONTROLLER_NAME: &str = "airflowcluster"; pub const CONTAINER_IMAGE_BASE_NAME: &str = "airflow"; -/// Pseudo role/role-group names for the Kubernetes executor's resources (it is not a real -/// AirflowRole). Used to derive its labels and ConfigMap name. -pub const EXECUTOR_ROLE_NAME: &str = "executor"; -pub const EXECUTOR_ROLE_GROUP_NAME: &str = "kubernetes"; - -/// The executor pseudo-role name (`executor`) as a type-safe value. -pub fn executor_role_name() -> RoleName { - EXECUTOR_ROLE_NAME - .parse() - .expect("'executor' is a valid role name") -} - -/// The executor's role-group name (`kubernetes`), used for its role-group ConfigMap. -pub fn executor_role_group_name() -> RoleGroupName { - EXECUTOR_ROLE_GROUP_NAME - .parse() - .expect("'kubernetes' is a valid role group name") -} - -/// The executor *pod-template* role-group name (`executor-template`), used for the template -/// ConfigMap/pod labels. -pub fn executor_template_role_group_name() -> RoleGroupName { - "executor-template" - .parse() - .expect("'executor-template' is a valid role group name") -} pub const AIRFLOW_FULL_CONTROLLER_NAME: &str = concatcp!(AIRFLOW_CONTROLLER_NAME, '.', OPERATOR_NAME); @@ -93,22 +53,9 @@ pub struct Ctx { #[strum_discriminants(derive(IntoStaticStr))] #[snafu(visibility(pub(crate)))] pub enum Error { - #[snafu(display("failed to apply Service for role group {role_group}"))] - ApplyRoleGroupService { - source: stackable_operator::cluster_resources::Error, - role_group: RoleGroupName, - }, - - #[snafu(display("failed to apply ConfigMap for role group {role_group}"))] - ApplyRoleGroupConfig { + #[snafu(display("failed to apply Kubernetes resource"))] + ApplyResource { source: stackable_operator::cluster_resources::Error, - role_group: RoleGroupName, - }, - - #[snafu(display("failed to apply StatefulSet for role group {role_group}"))] - ApplyRoleGroupStatefulSet { - source: stackable_operator::cluster_resources::Error, - role_group: RoleGroupName, }, #[snafu(display("failed to patch service account"))] @@ -126,10 +73,8 @@ pub enum Error { source: stackable_operator::commons::rbac::Error, }, - #[snafu(display("failed to build rolegroup ConfigMap"))] - BuildConfigMap { - source: crate::controller::build::resource::config_map::Error, - }, + #[snafu(display("failed to build the Kubernetes resources"))] + BuildResources { source: build::Error }, #[snafu(display("failed to delete orphaned resources"))] DeleteOrphanedResources { @@ -156,21 +101,6 @@ pub enum Error { source: crate::controller::validate::Error, }, - #[snafu(display("failed to apply executor template ConfigMap"))] - ApplyExecutorTemplateConfig { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to build the executor pod-template ConfigMap"))] - BuildExecutorTemplate { - source: crate::controller::build::resource::executor::Error, - }, - - #[snafu(display("failed to apply PodDisruptionBudget"))] - ApplyPdb { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to build label"))] BuildLabel { source: LabelError }, @@ -178,16 +108,6 @@ pub enum Error { InvalidAirflowCluster { source: error_boundary::InvalidObject, }, - - #[snafu(display("failed to build the rolegroup StatefulSet"))] - BuildStatefulSet { - source: crate::controller::build::resource::statefulset::Error, - }, - - #[snafu(display("failed to apply group listener"))] - ApplyGroupListener { - source: stackable_operator::cluster_resources::Error, - }, } pub(crate) type Result = std::result::Result; @@ -226,40 +146,7 @@ pub async fn reconcile_airflow( ) .context(ValidateSnafu)?; - // TODO: Move secret creation to a dedicated apply step once it exists. - random_secret_creation::create_random_secret_if_not_exists( - validated_cluster.internal_secret_name().as_ref(), - INTERNAL_SECRET_SECRET_KEY, - 256, - &validated_cluster, - client, - ) - .await - .context(InternalSecretSnafu)?; - - random_secret_creation::create_random_secret_if_not_exists( - validated_cluster.jwt_secret_name().as_ref(), - JWT_SECRET_SECRET_KEY, - 256, - &validated_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 `jUm21LuA76YZmrIa9u4eXRg0h0P24MDC9IDOmDvJbfw=`, which has 44 characters, which makes 32 bytes. - random_secret_creation::create_random_secret_if_not_exists( - validated_cluster.fernet_key_name().as_ref(), - FERNET_KEY_SECRET_KEY, - 32, - &validated_cluster, - client, - ) - .await - .context(InternalSecretSnafu)?; + ensure_random_secrets(client, &validated_cluster).await?; let mut cluster_resources = cluster_resources_new( &product_name(), @@ -279,8 +166,12 @@ pub async fn reconcile_airflow( let (rbac_sa, rbac_rolebinding) = build_rbac_resources(airflow, APP_NAME, required_labels).context(BuildRBACObjectsSnafu)?; - let rbac_sa = cluster_resources - .add(client, rbac_sa.clone()) + // The ServiceAccount name is deterministic on the built object, so the build step does not + // depend on the applied ServiceAccount. + let service_account_name = rbac_sa.name_any(); + + cluster_resources + .add(client, rbac_sa) .await .context(ApplyServiceAccountSnafu)?; cluster_resources @@ -288,105 +179,44 @@ pub async fn reconcile_airflow( .await .context(ApplyRoleBindingSnafu)?; + let resources = + build::build(&validated_cluster, &service_account_name).context(BuildResourcesSnafu)?; + let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - // if the kubernetes executor is specified, in place of a worker role that will be in the role - // collection there will be a pod template created to be used for pod provisioning - if let Some(executor_template) = &validated_cluster.cluster_config.executor_template { - build_executor_template( - executor_template, - &validated_cluster, - &mut cluster_resources, - client, - &rbac_sa, - ) - .await?; + // Apply order mirrors opensearch: everything before StatefulSets, StatefulSets last (a + // changed mounted ConfigMap/Secret must exist first, else Pods restart -- commons-operator#111). + for service in resources.services { + cluster_resources + .add(client, service) + .await + .context(ApplyResourceSnafu)?; } - - for (airflow_role, role_group_configs) in &validated_cluster.role_groups { - if let Some(role_config) = validated_cluster.role_configs.get(airflow_role) { - if let Some(pdb_config) = &role_config.pdb - && let Some(pdb) = build_pdb(pdb_config, &validated_cluster, airflow_role) - { - cluster_resources - .add(client, pdb) - .await - .context(ApplyPdbSnafu)?; - } - - if let Some(listener_class) = &role_config.listener_class - && let Some(listener_group_name) = &role_config.group_listener_name - { - let rg_group_listener = build_group_listener( - &validated_cluster, - airflow_role, - listener_class.clone(), - listener_group_name.clone(), - ); - cluster_resources - .add(client, rg_group_listener) - .await - .context(ApplyGroupListenerSnafu)?; - } - } - - for (rolegroup_name, validated_rg_config) in role_group_configs { - let logging = &validated_rg_config.config.logging; - - let rg_headless_service = - build_rolegroup_headless_service(&validated_cluster, airflow_role, rolegroup_name); - - cluster_resources - .add(client, rg_headless_service) - .await - .context(ApplyRoleGroupServiceSnafu { - role_group: rolegroup_name.clone(), - })?; - - let rg_metrics_service = - build_rolegroup_metrics_service(&validated_cluster, airflow_role, rolegroup_name); - cluster_resources - .add(client, rg_metrics_service) - .await - .context(ApplyRoleGroupServiceSnafu { - role_group: rolegroup_name.clone(), - })?; - - let rg_configmap = config_map::build_rolegroup_config_map( - &validated_cluster, - &airflow_role.role_name(), - rolegroup_name, - &validated_rg_config.config_overrides, - logging, - &Container::Airflow, - ) - .context(BuildConfigMapSnafu)?; + 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 resources.stateful_sets { + ss_cond_builder.add( cluster_resources - .add(client, rg_configmap) + .add(client, statefulset) .await - .with_context(|_| ApplyRoleGroupConfigSnafu { - role_group: rolegroup_name.clone(), - })?; - - let rg_statefulset = build_server_rolegroup_statefulset( - &validated_cluster, - airflow_role, - rolegroup_name, - validated_rg_config, - logging, - &rbac_sa, - ) - .context(BuildStatefulSetSnafu)?; - - ss_cond_builder.add( - cluster_resources - .add(client, rg_statefulset) - .await - .context(ApplyRoleGroupStatefulSetSnafu { - role_group: rolegroup_name.clone(), - })?, - ); - } + .context(ApplyResourceSnafu)?, + ); } cluster_resources @@ -409,43 +239,47 @@ pub async fn reconcile_airflow( Ok(Action::await_change()) } -async fn build_executor_template( - executor_template: &ValidatedExecutorTemplate, - validated_cluster: &ValidatedCluster, - cluster_resources: &mut ClusterResources<'_>, +/// 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, - rbac_sa: &stackable_operator::k8s_openapi::api::core::v1::ServiceAccount, + cluster: &ValidatedCluster, ) -> Result<(), Error> { - let executor_config = &executor_template.config; - let rg_configmap = config_map::build_rolegroup_config_map( - validated_cluster, - &executor_role_name(), - &executor_role_group_name(), - // The kubernetes-executor pod template does not apply webserver_config.py overrides - &AirflowConfigOverrides::default(), - &executor_config.logging, - &Container::Base, + random_secret_creation::create_random_secret_if_not_exists( + cluster.internal_secret_name().as_ref(), + INTERNAL_SECRET_SECRET_KEY, + 256, + cluster, + client, ) - .context(BuildConfigMapSnafu)?; - cluster_resources - .add(client, rg_configmap) - .await - .with_context(|_| ApplyRoleGroupConfigSnafu { - role_group: executor_role_group_name(), - })?; - - let worker_pod_template_config_map = build_executor_template_config_map( - validated_cluster, - &rbac_sa.name_unchecked(), - executor_config, - &executor_template.env_overrides, - &executor_template.pod_overrides, + .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, ) - .context(BuildExecutorTemplateSnafu)?; - cluster_resources - .add(client, worker_pod_template_config_map) - .await - .with_context(|_| ApplyExecutorTemplateConfigSnafu {})?; + .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 `jUm21LuA76YZmrIa9u4eXRg0h0P24MDC9IDOmDvJbfw=`, which has 44 characters, which makes 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 a5828891..f148f167 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -1,6 +1,305 @@ //! Builders that assemble Kubernetes resources from the validated cluster. +use snafu::{ResultExt, Snafu}; +use stackable_operator::v2::types::operator::RoleGroupName; + +use crate::{ + controller::{ + KubernetesResources, ValidatedCluster, + build::resource::{ + config_map, + executor::build_executor_template_config_map, + listener::build_group_listener, + pdb::build_pdb, + service::{build_rolegroup_headless_service, build_rolegroup_metrics_service}, + statefulset::build_server_rolegroup_statefulset, + }, + executor_role_group_name, executor_role_name, + }, + crd::{AirflowConfigOverrides, Container}, +}; + pub mod graceful_shutdown; pub mod properties; pub mod resource; pub mod volumes; + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to build ConfigMap for role group {role_group}"))] + ConfigMap { + source: resource::config_map::Error, + role_group: RoleGroupName, + }, + + #[snafu(display("failed to build StatefulSet for role group {role_group}"))] + StatefulSet { + source: resource::statefulset::Error, + role_group: RoleGroupName, + }, + + #[snafu(display("failed to build the Kubernetes-executor pod-template ConfigMap"))] + ExecutorTemplate { source: resource::executor::Error }, +} + +/// Builds every Kubernetes resource for the given validated cluster. +/// +/// 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. +/// +/// `service_account_name` is the name of the RBAC `ServiceAccount` the role-group Pods and the +/// Kubernetes-executor pod template run under (RBAC resources are built and applied separately, +/// in the reconcile step). +pub fn build( + cluster: &ValidatedCluster, + service_account_name: &str, +) -> Result { + let mut stateful_sets = vec![]; + let mut services = vec![]; + let mut listeners = vec![]; + let mut config_maps = vec![]; + let mut pod_disruption_budgets = vec![]; + + // The Kubernetes-executor pod template (only built for the Kubernetes executor; the Celery + // 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( + cluster, + &executor_role_name(), + &executor_role_group, + // The Kubernetes-executor pod template does not apply webserver_config.py overrides. + &AirflowConfigOverrides::default(), + &executor_template.config.logging, + &Container::Base, + ) + .context(ConfigMapSnafu { + role_group: executor_role_group, + })?; + config_maps.push(executor_config_map); + + let executor_template_config_map = build_executor_template_config_map( + cluster, + service_account_name, + &executor_template.config, + &executor_template.env_overrides, + &executor_template.pod_overrides, + ) + .context(ExecutorTemplateSnafu)?; + config_maps.push(executor_template_config_map); + } + + for (role, role_group_configs) in &cluster.role_groups { + if let Some(role_config) = cluster.role_configs.get(role) { + if let Some(pdb_config) = &role_config.pdb { + pod_disruption_budgets.extend(build_pdb(pdb_config, cluster, role)); + } + if let Some(listener_class) = &role_config.listener_class + && let Some(group_listener_name) = &role_config.group_listener_name + { + listeners.push(build_group_listener( + cluster, + role, + listener_class.clone(), + group_listener_name.clone(), + )); + } + } + + for (role_group_name, rg_config) in role_group_configs { + let logging = &rg_config.config.logging; + + services.push(build_rolegroup_headless_service( + cluster, + role, + role_group_name, + )); + services.push(build_rolegroup_metrics_service( + cluster, + role, + role_group_name, + )); + config_maps.push( + config_map::build_rolegroup_config_map( + cluster, + &role.role_name(), + role_group_name, + &rg_config.config_overrides, + logging, + &Container::Airflow, + ) + .context(ConfigMapSnafu { + role_group: role_group_name.clone(), + })?, + ); + stateful_sets.push( + build_server_rolegroup_statefulset( + cluster, + role, + role_group_name, + rg_config, + logging, + service_account_name, + ) + .context(StatefulSetSnafu { + role_group: role_group_name.clone(), + })?, + ); + } + } + + Ok(KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + pod_disruption_budgets, + }) +} + +#[cfg(test)] +mod tests { + use stackable_operator::kube::Resource; + + use super::build; + use crate::{ + controller::{ + ValidatedCluster, dereference::DereferencedObjects, validate::validate_cluster, + }, + crd::{ + authentication::{AirflowClientAuthenticationDetailsResolved, FlaskRolesSyncMoment}, + authorization::AirflowAuthorizationResolved, + v1alpha2, + }, + }; + + /// 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 { + let cluster_yaml = r#" + apiVersion: airflow.stackable.tech/v1alpha2 + kind: AirflowCluster + metadata: + name: airflow + namespace: default + uid: e6ac237d-a6d4-43a1-8135-f36506110912 + spec: + image: + productVersion: 3.1.6 + clusterConfig: + loadExamples: false + exposeConfig: false + credentialsSecretName: airflow-admin-credentials + metadataDatabase: + postgresql: + host: airflow-postgresql + database: airflow + credentialsSecretName: airflow-postgresql-credentials + webservers: + config: {} + roleGroups: + default: + config: {} + schedulers: + config: {} + roleGroups: + default: + config: {} + "#; + // The executor block is inserted into the parsed document rather than spliced into the + // YAML text, so the fixture does not depend on matching the template's indentation. + let mut cluster_value: serde_yaml::Value = + serde_yaml::from_str(cluster_yaml).expect("the test CR is valid YAML"); + cluster_value["spec"] + .as_mapping_mut() + .expect("the test CR has a spec mapping") + .insert( + executor_key.into(), + serde_yaml::from_str(executor_config).expect("the executor config is valid YAML"), + ); + let cluster: v1alpha2::AirflowCluster = + serde_yaml::with::singleton_map_recursive::deserialize(cluster_value) + .expect("the test CR deserialises"); + + let dereferenced = DereferencedObjects { + authentication_config: AirflowClientAuthenticationDetailsResolved { + authentication_classes_resolved: vec![], + user_registration: true, + user_registration_role: "Public".to_string(), + sync_roles_at: FlaskRolesSyncMoment::default(), + }, + authorization_config: AirflowAuthorizationResolved { opa: None }, + }; + + validate_cluster(&cluster, "oci.stackable.tech/sdp", dereferenced) + .expect("test cluster validates") + } + + /// 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 { + 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 { + validated_cluster("kubernetesExecutors", "{config: {}}") + } + + fn sorted_names(resources: &[impl Resource]) -> Vec<&str> { + let mut names: Vec<&str> = resources + .iter() + .filter_map(|resource| resource.meta().name.as_deref()) + .collect(); + names.sort(); + names + } + + #[test] + fn build_produces_expected_resource_names() { + let cluster = celery_executor_cluster(); + let resources = build(&cluster, "airflow-serviceaccount").expect("build succeeds"); + + assert_eq!( + sorted_names(&resources.stateful_sets), + ["airflow-scheduler-default", "airflow-webserver-default"] + ); + // One headless and one metrics Service per role group. + assert_eq!(resources.services.len(), 4); + assert_eq!( + sorted_names(&resources.config_maps), + ["airflow-scheduler-default", "airflow-webserver-default"] + ); + // The webserver is the only role with a group Listener. + assert_eq!(sorted_names(&resources.listeners), ["airflow-webserver"]); + // A default PDB per role (the Celery worker included). + assert_eq!( + sorted_names(&resources.pod_disruption_budgets), + ["airflow-scheduler", "airflow-webserver", "airflow-worker"] + ); + } + + /// The Kubernetes-executor branch of `build()` (moved here from `reconcile`) additionally emits + /// the executor role-group ConfigMap and the executor pod-template ConfigMap; the Celery case + /// does not. + #[test] + fn build_kubernetes_executor_adds_pod_template_config_maps() { + let cluster = kubernetes_executor_cluster(); + let resources = build(&cluster, "airflow-serviceaccount").expect("build succeeds"); + + assert_eq!( + sorted_names(&resources.config_maps), + [ + "airflow-executor-kubernetes", + "airflow-executor-pod-template", + "airflow-scheduler-default", + "airflow-webserver-default", + ] + ); + } +} diff --git a/rust/operator-binary/src/controller/build/resource/executor.rs b/rust/operator-binary/src/controller/build/resource/executor.rs index 31c00d7b..b4b6830d 100644 --- a/rust/operator-binary/src/controller/build/resource/executor.rs +++ b/rust/operator-binary/src/controller/build/resource/executor.rs @@ -22,9 +22,6 @@ use stackable_operator::{ }; use crate::{ - airflow_controller::{ - executor_role_group_name, executor_role_name, executor_template_role_group_name, - }, controller::{ ValidatedAirflowConfig, ValidatedCluster, build::{ @@ -36,6 +33,7 @@ use crate::{ }, volumes::{self, CONFIG_VOLUME_NAME, LOG_CONFIG_VOLUME_NAME, LOG_VOLUME_NAME}, }, + executor_role_group_name, executor_role_name, executor_template_role_group_name, }, crd::{CONFIG_PATH, Container, LOG_CONFIG_DIR, TEMPLATE_NAME}, }; diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index b4be976c..b357b93a 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -11,11 +11,11 @@ use stackable_operator::{ DeepMerge, api::{ apps::v1::{StatefulSet, StatefulSetSpec}, - core::v1::{PersistentVolumeClaim, Probe, ServiceAccount, TCPSocketAction}, + core::v1::{PersistentVolumeClaim, Probe, TCPSocketAction}, }, apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString}, }, - kube::{ResourceExt, api::ObjectMeta}, + kube::api::ObjectMeta, kvp::{Annotation, Label, LabelError}, utils::COMMON_BASH_TRAP_FUNCTIONS, v2::{ @@ -103,7 +103,7 @@ pub fn build_server_rolegroup_statefulset( role_group_name: &RoleGroupName, validated_rg_config: &AirflowRoleGroupConfig, logging: &ValidatedLogging, - service_account: &ServiceAccount, + service_account_name: &str, ) -> Result { let merged_airflow_config = &validated_rg_config.config; let env_overrides = &validated_rg_config.env_overrides; @@ -140,7 +140,7 @@ pub fn build_server_rolegroup_statefulset( pb.metadata(pb_metadata) .image_pull_secrets_from_product_image(resolved_product_image) .affinity(&merged_airflow_config.affinity) - .service_account_name(service_account.name_any()) + .service_account_name(service_account_name) .security_context(PodSecurityContextBuilder::new().fs_group(1000).build()); let mut airflow_container = new_container_builder(&Container::Airflow.to_container_name()); diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs index 03c5f350..d59e7b39 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -10,7 +10,7 @@ use stackable_operator::{ product_image_selection::ResolvedProductImage, resources::{NoRuntimeLimits, Resources}, }, - crd::git_sync, + crd::{git_sync, listener}, database_connections::{ TemplatingMechanism, drivers::{ @@ -18,7 +18,11 @@ use stackable_operator::{ sqlalchemy::SqlAlchemyDatabaseConnectionDetails, }, }, - k8s_openapi::api::core::v1::{PodTemplateSpec, Volume, VolumeMount}, + k8s_openapi::api::{ + apps::v1::StatefulSet, + core::v1::{ConfigMap, PodTemplateSpec, Service, Volume, VolumeMount}, + policy::v1::PodDisruptionBudget, + }, kube::{Resource, ResourceExt, api::ObjectMeta}, kvp::Labels, product_logging::spec::ContainerLogConfig, @@ -59,6 +63,15 @@ pub mod build; pub mod dereference; pub mod validate; +/// Every Kubernetes resource produced by the build step. +pub struct KubernetesResources { + pub stateful_sets: Vec, + pub services: Vec, + pub listeners: Vec, + pub config_maps: Vec, + pub pod_disruption_budgets: Vec, +} + /// Per-role configuration extracted during validation. #[derive(Clone, Debug)] pub struct ValidatedRoleConfig { @@ -407,6 +420,33 @@ pub(crate) fn controller_name() -> ControllerName { .expect("the controller name is a valid label value") } +/// Pseudo role/role-group names for the Kubernetes executor's resources (it is not a real +/// AirflowRole). Used to derive its labels and ConfigMap name. +pub const EXECUTOR_ROLE_NAME: &str = "executor"; +pub const EXECUTOR_ROLE_GROUP_NAME: &str = "kubernetes"; + +/// The executor pseudo-role name (`executor`) as a type-safe value. +pub fn executor_role_name() -> RoleName { + EXECUTOR_ROLE_NAME + .parse() + .expect("'executor' is a valid role name") +} + +/// The executor's role-group name (`kubernetes`), used for its role-group ConfigMap. +pub fn executor_role_group_name() -> RoleGroupName { + EXECUTOR_ROLE_GROUP_NAME + .parse() + .expect("'kubernetes' is a valid role group name") +} + +/// The executor *pod-template* role-group name (`executor-template`), used for the template +/// ConfigMap/pod labels. +pub fn executor_template_role_group_name() -> RoleGroupName { + "executor-template" + .parse() + .expect("'executor-template' is a valid role group name") +} + /// Lets [`ValidatedCluster`] stand in for the raw [`v1alpha2::AirflowCluster`] when building owner /// references and metadata for child objects. Kind/group/version are delegated to the CRD; the /// `metadata` (name, namespace, uid) is captured during validation. From e2e8d729cb05b5dd80ed62b6698201313e054e7c Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Fri, 10 Jul 2026 09:44:32 +0200 Subject: [PATCH 2/3] refined comment --- rust/operator-binary/src/airflow_controller.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rust/operator-binary/src/airflow_controller.rs b/rust/operator-binary/src/airflow_controller.rs index 0c4bcb37..6125b470 100644 --- a/rust/operator-binary/src/airflow_controller.rs +++ b/rust/operator-binary/src/airflow_controller.rs @@ -184,8 +184,8 @@ pub async fn reconcile_airflow( let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - // Apply order mirrors opensearch: everything before StatefulSets, StatefulSets last (a - // changed mounted ConfigMap/Secret must exist first, else Pods restart -- commons-operator#111). + // Apply order is: StatefulSets last (a changed mounted ConfigMap/Secret + // must exist first, else Pods restart -- commons-operator#111). for service in resources.services { cluster_resources .add(client, service) From f17d88b859c564e5aba619e082d7f72a7a6bf5b8 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Fri, 10 Jul 2026 10:15:01 +0200 Subject: [PATCH 3/3] linting --- rust/operator-binary/src/airflow_controller.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rust/operator-binary/src/airflow_controller.rs b/rust/operator-binary/src/airflow_controller.rs index 6125b470..2fda4260 100644 --- a/rust/operator-binary/src/airflow_controller.rs +++ b/rust/operator-binary/src/airflow_controller.rs @@ -184,7 +184,7 @@ pub async fn reconcile_airflow( let mut ss_cond_builder = StatefulSetConditionBuilder::default(); - // Apply order is: StatefulSets last (a changed mounted ConfigMap/Secret + // Apply order is: StatefulSets last (a changed mounted ConfigMap/Secret // must exist first, else Pods restart -- commons-operator#111). for service in resources.services { cluster_resources