fix: introduce ValidatedAirflowConfig, fold logging

Author: maltesander

rust/operator-binary/src/airflow_controller.rs

@@ -345,20 +345,16 @@ pub async fn reconcile_airflow(
             }
         }
 
-        for (rolegroup_name, validated_rg) in role_group_configs {
-            let validated_rg_config = &validated_rg.config;
-            let logging = &validated_rg.logging;
+        for (rolegroup_name, validated_rg_config) in role_group_configs {
+            let logging = &validated_rg_config.config.logging;
 
             let git_sync_resources = git_sync::v1alpha2::GitSyncResources::new(
                 &validated_cluster.cluster_config.dags_git_sync,
                 &validated_cluster.image,
                 &Vec::<EnvVar>::from(validated_rg_config.env_overrides.clone()),
                 &airflow.volume_mounts(),
                 LOG_VOLUME_NAME.as_ref(),
-                &validated_rg_config
-                    .config
-                    .logging
-                    .for_container(&Container::GitSync),
+                &logging.git_sync_container,
             )
             .context(InvalidGitSyncSpecSnafu)?;
 

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

@@ -157,10 +157,7 @@ pub fn build_executor_template_config_map(
             .resource_names(&executor_role_name(), &executor_role_group_name())
             .role_group_config_map()
             .as_ref(),
-        merged_executor_config
-            .logging
-            .containers
-            .get(&Container::Airflow),
+        &executor_logging.product_container,
     ))
     .context(AddVolumeSnafu)?;
 

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

@@ -3,10 +3,8 @@ use stackable_operator::{
     builder::{
         meta::ObjectMetaBuilder,
         pod::{
-            PodBuilder,
-            resources::ResourceRequirementsBuilder,
-            security::PodSecurityContextBuilder,
-            volume::VolumeBuilder,
+            PodBuilder, resources::ResourceRequirementsBuilder,
+            security::PodSecurityContextBuilder, volume::VolumeBuilder,
         },
     },
     crd::git_sync,
@@ -47,8 +45,8 @@ use crate::{
     },
     controller_commons::{self, CONFIG_VOLUME_NAME, LOG_CONFIG_VOLUME_NAME, LOG_VOLUME_NAME},
     crd::{
-        AirflowExecutor, AirflowRole, CONFIG_PATH, Container, HTTP_PORT_NAME, LISTENER_VOLUME_DIR,
-        LISTENER_PVC_NAME, LOG_CONFIG_DIR, METRICS_CONTAINER_NAME, METRICS_PORT,
+        AirflowExecutor, AirflowRole, CONFIG_PATH, Container, HTTP_PORT_NAME, LISTENER_PVC_NAME,
+        LISTENER_VOLUME_DIR, LOG_CONFIG_DIR, METRICS_CONTAINER_NAME, METRICS_PORT,
         METRICS_PORT_NAME, STACKABLE_LOG_DIR, TEMPLATE_LOCATION, TEMPLATE_VOLUME_NAME,
     },
 };
@@ -304,10 +302,7 @@ pub fn build_server_rolegroup_statefulset(
         .context(AddVolumeSnafu)?;
     pb.add_volumes(controller_commons::create_volumes(
         resource_names.role_group_config_map().as_ref(),
-        merged_airflow_config
-            .logging
-            .containers
-            .get(&Container::Airflow),
+        &logging.product_container,
     ))
     .context(AddVolumeSnafu)?;
 

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

@@ -2,14 +2,20 @@ use std::{collections::BTreeMap, str::FromStr};
 
 use stackable_operator::{
     builder::meta::ObjectMetaBuilder,
-    commons::product_image_selection::ResolvedProductImage,
+    commons::{
+        affinity::StackableAffinity,
+        product_image_selection::ResolvedProductImage,
+        resources::{NoRuntimeLimits, Resources},
+    },
     crd::git_sync,
     database_connections::drivers::{
         celery::CeleryDatabaseConnectionDetails, sqlalchemy::SqlAlchemyDatabaseConnectionDetails,
     },
     k8s_openapi::api::core::v1::{Volume, VolumeMount},
     kube::{Resource, ResourceExt, api::ObjectMeta},
     kvp::Labels,
+    product_logging::spec::ContainerLogConfig,
+    shared::time::Duration,
     v2::{
         HasName, HasUid, NameIsValidLabelValue,
         builder::meta::ownerreference_from_resource,
@@ -32,7 +38,8 @@ use crate::{
     airflow_controller::AIRFLOW_CONTROLLER_NAME,
     crd::{
         APP_NAME, AirflowConfig, AirflowConfigOverrides, AirflowExecutor, AirflowRole,
-        OPERATOR_NAME, authentication::AirflowClientAuthenticationDetailsResolved,
+        AirflowStorageConfig, OPERATOR_NAME,
+        authentication::AirflowClientAuthenticationDetailsResolved,
         authorization::AirflowAuthorizationResolved, v1alpha2,
     },
 };
@@ -56,32 +63,45 @@ pub struct ValidatedRoleConfig {
 /// `env_overrides`/`cli_overrides`/`pod_overrides`. The config overrides are kept typed
 /// ([`AirflowConfigOverrides`]) and flattened into the rendered config file later, in the build step.
 pub type AirflowRoleGroupConfig = stackable_operator::v2::role_utils::RoleGroupConfig<
-    AirflowConfig,
+    ValidatedAirflowConfig,
     stackable_operator::v2::role_utils::GenericCommonConfig,
     AirflowConfigOverrides,
 >;
 
-/// A validated role group: the merged [`AirflowRoleGroupConfig`] paired with its up-front-validated
-/// [`ValidatedLogging`], so the build step reads both from here rather than re-deriving from the raw
-/// cluster. (Superset folds logging into a single rolegroup struct; airflow keeps the generic merged
-/// config as-is — retaining all its fields without bespoke `#[allow(dead_code)]` — and carries
-/// logging alongside it.)
-#[derive(Clone, Debug)]
-pub struct AirflowRoleGroup {
-    pub config: AirflowRoleGroupConfig,
+/// A validated, merged Airflow role-group config: the merged [`AirflowConfig`] with its raw
+/// `logging` replaced by the up-front-validated [`ValidatedLogging`] (so an invalid custom log
+/// ConfigMap name or a missing Vector aggregator name fails reconciliation during validation).
+#[derive(Clone, Debug, PartialEq)]
+pub struct ValidatedAirflowConfig {
+    pub resources: Resources<AirflowStorageConfig, NoRuntimeLimits>,
     pub logging: ValidatedLogging,
+    pub affinity: StackableAffinity,
+    pub graceful_shutdown_timeout: Option<Duration>,
 }
 
-/// Validated logging configuration for the product container and the (optional) Vector container.
-///
+impl ValidatedAirflowConfig {
+    /// Builds the validated config from the merged [`AirflowConfig`], swapping in the
+    /// already-validated logging.
+    pub(crate) fn from_merged(merged: AirflowConfig, logging: ValidatedLogging) -> Self {
+        Self {
+            resources: merged.resources,
+            logging,
+            affinity: merged.affinity,
+            graceful_shutdown_timeout: merged.graceful_shutdown_timeout,
+        }
+    }
+}
+
+/// Validated logging configuration for the containers of a role-group (or Kubernetes-executor) Pod.
 ///
 /// `product_container` holds the validated log-config choice of the product's main container
 /// (`Container::Airflow` for the role groups, `Container::Base` for the Kubernetes-executor pod
-/// template).
-#[derive(Clone, Debug, PartialEq, Eq)]
+/// template). `git_sync_container` holds the log config of the git-sync sidecar (DAG fetching).
+#[derive(Clone, Debug, PartialEq)]
 pub struct ValidatedLogging {
     pub product_container: ValidatedContainerLogConfigChoice,
     pub vector_container: Option<VectorContainerLogConfig>,
+    pub git_sync_container: ContainerLogConfig,
     pub enable_vector_agent: bool,
 }
 
@@ -132,7 +152,7 @@ pub struct ValidatedCluster {
     pub product_version: ProductVersion,
     pub image: ResolvedProductImage,
     pub cluster_config: ValidatedClusterConfig,
-    pub role_groups: BTreeMap<AirflowRole, BTreeMap<RoleGroupName, AirflowRoleGroup>>,
+    pub role_groups: BTreeMap<AirflowRole, BTreeMap<RoleGroupName, AirflowRoleGroupConfig>>,
     pub role_configs: BTreeMap<AirflowRole, ValidatedRoleConfig>,
 }
 
@@ -143,7 +163,7 @@ impl ValidatedCluster {
         uid: Uid,
         image: ResolvedProductImage,
         cluster_config: ValidatedClusterConfig,
-        role_groups: BTreeMap<AirflowRole, BTreeMap<RoleGroupName, AirflowRoleGroup>>,
+        role_groups: BTreeMap<AirflowRole, BTreeMap<RoleGroupName, AirflowRoleGroupConfig>>,
         role_configs: BTreeMap<AirflowRole, ValidatedRoleConfig>,
     ) -> Self {
         // `app_version_label_value` is constructed to be a valid label value, so it is also a valid

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

@@ -30,7 +30,7 @@ use stackable_operator::{
 use strum::IntoEnumIterator;
 
 use super::{
-    AirflowRoleGroup, AirflowRoleGroupConfig, ValidatedCluster, ValidatedClusterConfig,
+    AirflowRoleGroupConfig, ValidatedAirflowConfig, ValidatedCluster, ValidatedClusterConfig,
     ValidatedLogging, ValidatedRoleConfig, dereference::DereferencedObjects,
 };
 use crate::{
@@ -152,15 +152,15 @@ pub fn validate_cluster(
                     role_group: rolegroup_name.clone(),
                 }
             })?;
-            let config =
-                validate_role_group(&resolved_role, &role_group_name, rolegroup, &default_config)?;
-            let logging = validate_logging(
-                &config.config.logging,
-                &Container::Airflow,
+            let config = validate_role_group(
+                &resolved_role,
+                &role_group_name,
+                rolegroup,
+                &default_config,
                 &vector_aggregator_config_map_name,
             )?;
 
-            group_configs.insert(role_group_name, AirflowRoleGroup { config, logging });
+            group_configs.insert(role_group_name, config);
         }
 
         role_groups.insert(role, group_configs);
@@ -274,6 +274,7 @@ fn validate_role_group(
     role_group_name: &RoleGroupName,
     rolegroup: &RoleGroup<AirflowConfigFragment, GenericCommonConfig, AirflowConfigOverrides>,
     default_config: &AirflowConfigFragment,
+    vector_aggregator_config_map_name: &Option<ConfigMapName>,
 ) -> Result<AirflowRoleGroupConfig, Error> {
     let validated = with_validated_config::<
         AirflowConfig,
@@ -294,9 +295,16 @@ fn validate_role_group(
         );
     }
 
+    let merged_config = validated.config.config;
+    let logging = validate_logging(
+        &merged_config.logging,
+        &Container::Airflow,
+        vector_aggregator_config_map_name,
+    )?;
+
     Ok(RoleGroupConfig {
         replicas: validated.replicas,
-        config: validated.config.config,
+        config: ValidatedAirflowConfig::from_merged(merged_config, logging),
         config_overrides: validated.config.config_overrides,
         env_overrides,
         cli_overrides: validated.config.cli_overrides,
@@ -340,6 +348,7 @@ pub(crate) fn validate_logging(
     Ok(ValidatedLogging {
         product_container,
         vector_container,
+        git_sync_container: logging.for_container(&Container::GitSync).into_owned(),
         enable_vector_agent: logging.enable_vector_agent,
     })
 }
@@ -414,6 +423,7 @@ mod tests {
             &"default".parse().expect("valid role group name"),
             rolegroup,
             &default_config,
+            &None,
         )
         .expect("validated role group");
         let config_overrides = validated.config_overrides;
@@ -511,6 +521,7 @@ mod tests {
             &"default".parse().expect("valid role group name"),
             rolegroup,
             &default_config,
+            &None,
         )
         .expect("validated role group");
 
@@ -584,6 +595,7 @@ mod tests {
             &"default".parse().expect("valid role group name"),
             rolegroup,
             &default_config,
+            &None,
         )
         .expect("validated role group");
 

rust/operator-binary/src/controller_commons.rs

@@ -3,14 +3,11 @@ use std::str::FromStr;
 use stackable_operator::{
     builder::pod::volume::VolumeBuilder,
     k8s_openapi::api::core::v1::{ConfigMapVolumeSource, EmptyDirVolumeSource, Volume},
-    product_logging::{
-        self,
-        spec::{
-            ConfigMapLogConfig, ContainerLogConfig, ContainerLogConfigChoice,
-            CustomContainerLogConfig,
-        },
+    product_logging,
+    v2::{
+        product_logging::framework::ValidatedContainerLogConfigChoice,
+        types::kubernetes::VolumeName,
     },
-    v2::types::kubernetes::VolumeName,
 };
 
 use crate::crd::MAX_LOG_FILES_SIZE;
@@ -21,7 +18,7 @@ stackable_operator::constant!(pub LOG_VOLUME_NAME: VolumeName = "log");
 
 pub fn create_volumes(
     config_map_name: &str,
-    log_config: Option<&ContainerLogConfig>,
+    product_log_config: &ValidatedContainerLogConfigChoice,
 ) -> Vec<Volume> {
     let mut volumes = Vec::new();
 
@@ -41,31 +38,22 @@ pub fn create_volumes(
         ..Volume::default()
     });
 
-    if let Some(ContainerLogConfig {
-        choice:
-            Some(ContainerLogConfigChoice::Custom(CustomContainerLogConfig {
-                custom: ConfigMapLogConfig { config_map },
-            })),
-    }) = log_config
-    {
-        volumes.push(Volume {
-            name: LOG_CONFIG_VOLUME_NAME.to_string(),
-            config_map: Some(ConfigMapVolumeSource {
-                name: config_map.into(),
-                ..ConfigMapVolumeSource::default()
-            }),
-            ..Volume::default()
-        });
-    } else {
-        volumes.push(Volume {
-            name: LOG_CONFIG_VOLUME_NAME.to_string(),
-            config_map: Some(ConfigMapVolumeSource {
-                name: config_map_name.into(),
-                ..ConfigMapVolumeSource::default()
-            }),
-            ..Volume::default()
-        });
-    }
+    // A custom log config is mounted from its own ConfigMap; an automatic one is rendered into the
+    // rolegroup ConfigMap.
+    let log_config_config_map = match product_log_config {
+        ValidatedContainerLogConfigChoice::Custom(custom_config_map) => {
+            custom_config_map.to_string()
+        }
+        ValidatedContainerLogConfigChoice::Automatic(_) => config_map_name.to_string(),
+    };
+    volumes.push(Volume {
+        name: LOG_CONFIG_VOLUME_NAME.to_string(),
+        config_map: Some(ConfigMapVolumeSource {
+            name: log_config_config_map,
+            ..ConfigMapVolumeSource::default()
+        }),
+        ..Volume::default()
+    });
 
     volumes
 }