fix: consolidate executor build merging

Author: maltesander

rust/operator-binary/src/airflow_controller.rs

@@ -30,7 +30,7 @@ use strum::{EnumDiscriminants, IntoStaticStr};
 
 use crate::{
     controller::{
-        ValidatedCluster,
+        ValidatedCluster, ValidatedExecutorTemplate,
         build::{
             config_map,
             resource::{
@@ -44,8 +44,7 @@ use crate::{
     },
     controller_commons::LOG_VOLUME_NAME,
     crd::{
-        self, APP_NAME, AirflowClusterStatus, AirflowConfigOverrides, AirflowExecutor,
-        AirflowExecutorCommonConfiguration, Container, OPERATOR_NAME,
+        APP_NAME, AirflowClusterStatus, AirflowConfigOverrides, Container, OPERATOR_NAME,
         internal_secret::{
             FERNET_KEY_SECRET_KEY, INTERNAL_SECRET_SECRET_KEY, JWT_SECRET_SECRET_KEY,
         },
@@ -132,9 +131,6 @@ pub enum Error {
         source: crate::controller::build::config_map::Error,
     },
 
-    #[snafu(display("failed to resolve and merge config for role and role group"))]
-    FailedToResolveConfig { source: crd::Error },
-
     #[snafu(display("invalid git-sync specification"))]
     InvalidGitSyncSpec { source: git_sync::v1alpha2::Error },
 
@@ -303,13 +299,9 @@ pub async fn reconcile_airflow(
 
     // 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 AirflowExecutor::KubernetesExecutors {
-        common_configuration,
-    } = &validated_cluster.cluster_config.executor
-    {
+    if let Some(executor_template) = &validated_cluster.cluster_config.executor_template {
         build_executor_template(
-            airflow,
-            common_configuration,
+            executor_template,
             &validated_cluster,
             &mut cluster_resources,
             client,
@@ -352,7 +344,7 @@ pub async fn reconcile_airflow(
                 &validated_cluster.cluster_config.dags_git_sync,
                 &validated_cluster.image,
                 &Vec::<EnvVar>::from(validated_rg_config.env_overrides.clone()),
-                &airflow.volume_mounts(),
+                &validated_cluster.volume_mounts(),
                 LOG_VOLUME_NAME.as_ref(),
                 &logging.git_sync_container,
             )
@@ -436,35 +428,21 @@ pub async fn reconcile_airflow(
 }
 
 async fn build_executor_template(
-    airflow: &v1alpha2::AirflowCluster,
-    common_config: &AirflowExecutorCommonConfiguration,
+    executor_template: &ValidatedExecutorTemplate,
     validated_cluster: &ValidatedCluster,
     cluster_resources: &mut ClusterResources<'_>,
     client: &stackable_operator::client::Client,
     rbac_sa: &stackable_operator::k8s_openapi::api::core::v1::ServiceAccount,
 ) -> Result<(), Error> {
-    let merged_executor_config = airflow
-        .merged_executor_config(&common_config.config)
-        .context(FailedToResolveConfigSnafu)?;
-    // The Kubernetes-executor pod template is not an `AirflowRole` with role groups, so its logging
-    // is validated here (at build time) via the shared `validate_logging`, mirroring the role-group
-    // path in `validate`. `Container::Base` is the executor's product container.
-    let executor_logging = crate::controller::validate::validate_logging(
-        &merged_executor_config.logging,
-        &Container::Base,
-        &validated_cluster
-            .cluster_config
-            .vector_aggregator_config_map_name,
-    )
-    .context(ValidateSnafu)?;
+    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
         // (preserves prior behaviour, which passed an empty map here).
         &AirflowConfigOverrides::default(),
-        &executor_logging,
+        &executor_config.logging,
         &Container::Base,
     )
     .context(BuildConfigMapSnafu)?;
@@ -478,22 +456,19 @@ async fn build_executor_template(
     let git_sync_resources = git_sync::v1alpha2::GitSyncResources::new(
         &validated_cluster.cluster_config.dags_git_sync,
         &validated_cluster.image,
-        &env_vars_from_overrides(&common_config.env_overrides),
-        &airflow.volume_mounts(),
+        &env_vars_from_overrides(&executor_template.env_overrides),
+        &validated_cluster.volume_mounts(),
         LOG_VOLUME_NAME.as_ref(),
-        &merged_executor_config
-            .logging
-            .for_container(&Container::GitSync),
+        &executor_config.logging.git_sync_container,
     )
     .context(InvalidGitSyncSpecSnafu)?;
 
     let worker_pod_template_config_map = build_executor_template_config_map(
         validated_cluster,
         &rbac_sa.name_unchecked(),
-        &merged_executor_config,
-        &executor_logging,
-        &common_config.env_overrides,
-        &common_config.pod_overrides,
+        executor_config,
+        &executor_template.env_overrides,
+        &executor_template.pod_overrides,
         &git_sync_resources,
     )
     .context(BuildExecutorTemplateSnafu)?;

rust/operator-binary/src/controller/build/properties/env_vars.rs

@@ -13,9 +13,9 @@ use stackable_operator::{
 };
 
 use crate::{
-    controller::ValidatedCluster,
+    controller::{ValidatedCluster, ValidatedLogging},
     crd::{
-        AirflowExecutor, AirflowRole, ExecutorConfig, HTTP_PORT, LOG_CONFIG_DIR, STACKABLE_LOG_DIR,
+        AirflowExecutor, AirflowRole, HTTP_PORT, LOG_CONFIG_DIR, STACKABLE_LOG_DIR,
         TEMPLATE_LOCATION, TEMPLATE_NAME,
         authentication::{
             AirflowAuthenticationClassResolved, AirflowClientAuthenticationDetailsResolved,
@@ -360,7 +360,7 @@ fn static_envs(
 pub fn build_airflow_template_envs(
     cluster: &ValidatedCluster,
     env_overrides: &HashMap<String, String>,
-    config: &ExecutorConfig,
+    logging: &ValidatedLogging,
     git_sync_resources: &git_sync::v1alpha2::GitSyncResources,
 ) -> Vec<EnvVar> {
     let mut env: BTreeMap<String, EnvVar> = BTreeMap::new();
@@ -417,7 +417,7 @@ pub fn build_airflow_template_envs(
     // _STACKABLE_POST_HOOK will contain a command to create a shutdown hook that will be
     // evaluated in the wrapper for each stackable spark container: this is necessary for pods
     // that are created and then terminated (we do a similar thing for spark-k8s).
-    if config.logging.enable_vector_agent {
+    if logging.enable_vector_agent {
         env.insert(
             "_STACKABLE_POST_HOOK".into(),
             EnvVar {

rust/operator-binary/src/controller/build/resource/executor.rs

@@ -24,7 +24,7 @@ use crate::{
         executor_role_group_name, executor_role_name, executor_template_role_group_name,
     },
     controller::{
-        ValidatedCluster, ValidatedLogging,
+        ValidatedAirflowConfig, ValidatedCluster,
         build::{
             graceful_shutdown::add_graceful_shutdown_config,
             properties::env_vars::build_airflow_template_envs,
@@ -35,9 +35,7 @@ use crate::{
         },
     },
     controller_commons::{self, CONFIG_VOLUME_NAME, LOG_CONFIG_VOLUME_NAME, LOG_VOLUME_NAME},
-    crd::{
-        CONFIG_PATH, Container, ExecutorConfig, LOG_CONFIG_DIR, STACKABLE_LOG_DIR, TEMPLATE_NAME,
-    },
+    crd::{CONFIG_PATH, Container, LOG_CONFIG_DIR, STACKABLE_LOG_DIR, TEMPLATE_NAME},
 };
 
 #[derive(Snafu, Debug)]
@@ -79,8 +77,7 @@ type Result<T, E = Error> = std::result::Result<T, E>;
 pub fn build_executor_template_config_map(
     cluster: &ValidatedCluster,
     sa_name: &str,
-    merged_executor_config: &ExecutorConfig,
-    executor_logging: &ValidatedLogging,
+    executor_config: &ValidatedAirflowConfig,
     env_overrides: &HashMap<String, String>,
     pod_overrides: &PodTemplateSpec,
     git_sync_resources: &git_sync::v1alpha2::GitSyncResources,
@@ -99,12 +96,12 @@ pub fn build_executor_template_config_map(
 
     pb.metadata(pb_metadata)
         .image_pull_secrets_from_product_image(resolved_product_image)
-        .affinity(&merged_executor_config.affinity)
+        .affinity(&executor_config.affinity)
         .service_account_name(sa_name)
         .restart_policy("Never")
         .security_context(PodSecurityContextBuilder::new().fs_group(1000).build());
 
-    add_graceful_shutdown_config(merged_executor_config.graceful_shutdown_timeout, &mut pb)
+    add_graceful_shutdown_config(executor_config.graceful_shutdown_timeout, &mut pb)
         .context(GracefulShutdownSnafu)?;
 
     // N.B. this "base" name is an airflow requirement and should not be changed!
@@ -119,11 +116,11 @@ pub fn build_executor_template_config_map(
     .context(PodSnafu)?;
     airflow_container
         .image_from_product_image(resolved_product_image)
-        .resources(merged_executor_config.resources.clone().into())
+        .resources(executor_config.resources.clone().into())
         .add_env_vars(build_airflow_template_envs(
             cluster,
             env_overrides,
-            merged_executor_config,
+            &executor_config.logging,
             git_sync_resources,
         ))
         .add_volume_mounts(cluster.volume_mounts())
@@ -157,11 +154,11 @@ pub fn build_executor_template_config_map(
             .resource_names(&executor_role_name(), &executor_role_group_name())
             .role_group_config_map()
             .as_ref(),
-        &executor_logging.product_container,
+        &executor_config.logging.product_container,
     ))
     .context(AddVolumeSnafu)?;
 
-    if let Some(vector_log_config) = &executor_logging.vector_container {
+    if let Some(vector_log_config) = &executor_config.logging.vector_container {
         pb.add_container(build_logging_container(
             resolved_product_image,
             vector_log_config,

rust/operator-binary/src/controller/mod.rs

@@ -1,4 +1,7 @@
-use std::{collections::BTreeMap, str::FromStr};
+use std::{
+    collections::{BTreeMap, HashMap},
+    str::FromStr,
+};
 
 use stackable_operator::{
     builder::meta::ObjectMetaBuilder,
@@ -11,7 +14,7 @@ use stackable_operator::{
     database_connections::drivers::{
         celery::CeleryDatabaseConnectionDetails, sqlalchemy::SqlAlchemyDatabaseConnectionDetails,
     },
-    k8s_openapi::api::core::v1::{Volume, VolumeMount},
+    k8s_openapi::api::core::v1::{PodTemplateSpec, Volume, VolumeMount},
     kube::{Resource, ResourceExt, api::ObjectMeta},
     kvp::Labels,
     product_logging::spec::ContainerLogConfig,
@@ -38,7 +41,7 @@ use crate::{
     airflow_controller::AIRFLOW_CONTROLLER_NAME,
     crd::{
         APP_NAME, AirflowConfig, AirflowConfigOverrides, AirflowExecutor, AirflowRole,
-        AirflowStorageConfig, OPERATOR_NAME,
+        AirflowStorageConfig, ExecutorConfig, OPERATOR_NAME,
         authentication::AirflowClientAuthenticationDetailsResolved,
         authorization::AirflowAuthorizationResolved, v1alpha2,
     },
@@ -90,6 +93,29 @@ impl ValidatedAirflowConfig {
             graceful_shutdown_timeout: merged.graceful_shutdown_timeout,
         }
     }
+
+    /// Builds the validated config from the merged [`ExecutorConfig`] (Kubernetes-executor pod
+    /// template), swapping in the already-validated logging. [`ExecutorConfig`] is field-identical
+    /// to [`AirflowConfig`].
+    pub(crate) fn from_merged_executor(merged: ExecutorConfig, logging: ValidatedLogging) -> Self {
+        Self {
+            resources: merged.resources,
+            logging,
+            affinity: merged.affinity,
+            graceful_shutdown_timeout: merged.graceful_shutdown_timeout,
+        }
+    }
+}
+
+/// The validated Kubernetes-executor pod-template config, computed during validation so the build
+/// step never merges or validates the raw cluster's executor config itself.
+pub struct ValidatedExecutorTemplate {
+    /// The merged + validated executor config (resources, affinity, logging, …).
+    pub config: ValidatedAirflowConfig,
+    /// Env-var overrides for the executor pod template (`spec.kubernetesExecutors.envOverrides`).
+    pub env_overrides: HashMap<String, String>,
+    /// Pod overrides for the executor pod template (`spec.kubernetesExecutors.podOverrides`).
+    pub pod_overrides: PodTemplateSpec,
 }
 
 /// Validated logging configuration for the containers of a role-group (or Kubernetes-executor) Pod.
@@ -111,6 +137,9 @@ pub struct ValidatedLogging {
 /// here rather than from the raw cluster object.
 pub struct ValidatedClusterConfig {
     pub executor: AirflowExecutor,
+    /// The validated Kubernetes-executor pod-template config (`None` for the Celery executor),
+    /// merged and logging-validated up-front so the build step does not touch the raw cluster.
+    pub executor_template: Option<ValidatedExecutorTemplate>,
     pub authentication_config: AirflowClientAuthenticationDetailsResolved,
     pub authorization_config: AirflowAuthorizationResolved,
     /// The Git-sync definitions for the DAGs (`spec.clusterConfig.dagsGitSync`).
@@ -130,8 +159,6 @@ pub struct ValidatedClusterConfig {
     pub volumes: Vec<Volume>,
     /// User-supplied extra VolumeMounts (`spec.clusterConfig.volumeMounts`).
     pub volume_mounts: Vec<VolumeMount>,
-    /// The validated Vector aggregator discovery ConfigMap name (`None` when no aggregator is set).
-    pub vector_aggregator_config_map_name: Option<ConfigMapName>,
 }
 
 /// The validated cluster: proves that config merging succeeded for every role and

rust/operator-binary/src/controller/validate.rs

@@ -31,7 +31,8 @@ use strum::IntoEnumIterator;
 
 use super::{
     AirflowRoleGroupConfig, ValidatedAirflowConfig, ValidatedCluster, ValidatedClusterConfig,
-    ValidatedLogging, ValidatedRoleConfig, dereference::DereferencedObjects,
+    ValidatedExecutorTemplate, ValidatedLogging, ValidatedRoleConfig,
+    dereference::DereferencedObjects,
 };
 use crate::{
     airflow_controller::CONTAINER_IMAGE_BASE_NAME,
@@ -64,6 +65,9 @@ pub enum Error {
         source: stackable_operator::v2::controller_utils::Error,
     },
 
+    #[snafu(display("failed to merge the Kubernetes-executor pod-template config"))]
+    MergeExecutorConfig { source: crate::crd::Error },
+
     #[snafu(display("invalid role group name {role_group}"))]
     ParseRoleGroupName {
         source: stackable_operator::v2::macros::attributed_string_type::Error,
@@ -174,13 +178,38 @@ pub fn validate_cluster(
     let (metadata_database_connection_details, celery_database_connection_details) =
         database_connection_details(airflow);
 
+    // The Kubernetes-executor pod template is not an `AirflowRole` with role groups, so its config
+    // is merged and its logging validated here, up-front, rather than in the build step.
+    // `Container::Base` is the executor's product container.
+    let executor_template = match &airflow.spec.executor {
+        AirflowExecutor::KubernetesExecutors {
+            common_configuration,
+        } => {
+            let merged = airflow
+                .merged_executor_config(&common_configuration.config)
+                .context(MergeExecutorConfigSnafu)?;
+            let logging = validate_logging(
+                &merged.logging,
+                &Container::Base,
+                &vector_aggregator_config_map_name,
+            )?;
+            Some(ValidatedExecutorTemplate {
+                config: ValidatedAirflowConfig::from_merged_executor(merged, logging),
+                env_overrides: common_configuration.env_overrides.clone(),
+                pod_overrides: common_configuration.pod_overrides.clone(),
+            })
+        }
+        AirflowExecutor::CeleryExecutors { .. } => None,
+    };
+
     Ok(ValidatedCluster::new(
         cluster_name,
         namespace,
         uid,
         resolved_product_image,
         ValidatedClusterConfig {
             executor: airflow.spec.executor.clone(),
+            executor_template,
             authentication_config,
             authorization_config,
             dags_git_sync: airflow.spec.cluster_config.dags_git_sync.clone(),
@@ -196,7 +225,6 @@ pub fn validate_cluster(
                 .enabled,
             volumes: airflow.spec.cluster_config.volumes.clone(),
             volume_mounts: airflow.spec.cluster_config.volume_mounts.clone(),
-            vector_aggregator_config_map_name: vector_aggregator_config_map_name.clone(),
         },
         role_groups,
         role_configs,

rust/operator-binary/src/crd/mod.rs

@@ -429,10 +429,6 @@ impl v1alpha2::AirflowCluster {
         self.get_role(role).map(|r| r.role_config)
     }
 
-    pub fn volume_mounts(&self) -> Vec<VolumeMount> {
-        self.spec.cluster_config.volume_mounts.clone()
-    }
-
     /// Retrieve and merge resource configs for the executor template
     pub fn merged_executor_config(
         &self,