alter structure to attempt to being it mroe in line with e.g. hdfs. Added test for merge/extend difference

adwk67 · adwk67 · commit 5a6f9dcc3434 · 2026-06-09T18:17:13.000+02:00
diff --git a/rust/operator-binary/src/airflow_controller.rs b/rust/operator-binary/src/airflow_controller.rs
@@ -1,6 +1,6 @@
 //! Ensures that `Pod`s are configured and running for each [`v1alpha2::AirflowCluster`]
 use std::{
-    collections::{BTreeMap, BTreeSet, HashMap},
+    collections::{BTreeSet, HashMap},
     sync::Arc,
 };
 
@@ -73,10 +73,11 @@ use crate::{
     },
     controller_commons::{self, CONFIG_VOLUME_NAME, LOG_CONFIG_VOLUME_NAME, LOG_VOLUME_NAME},
     crd::{
-        self, APP_NAME, AirflowClusterStatus, AirflowExecutor, AirflowExecutorCommonConfiguration,
-        AirflowRole, CONFIG_PATH, Container, ExecutorConfig, HTTP_PORT, HTTP_PORT_NAME,
-        LISTENER_VOLUME_DIR, LISTENER_VOLUME_NAME, LOG_CONFIG_DIR, METRICS_PORT, METRICS_PORT_NAME,
-        OPERATOR_NAME, STACKABLE_LOG_DIR, TEMPLATE_LOCATION, TEMPLATE_NAME, TEMPLATE_VOLUME_NAME,
+        self, APP_NAME, AirflowClusterStatus, AirflowConfigOverrides, AirflowExecutor,
+        AirflowExecutorCommonConfiguration, AirflowRole, CONFIG_PATH, Container, ExecutorConfig,
+        HTTP_PORT, HTTP_PORT_NAME, LISTENER_VOLUME_DIR, LISTENER_VOLUME_NAME, LOG_CONFIG_DIR,
+        METRICS_PORT, METRICS_PORT_NAME, OPERATOR_NAME, STACKABLE_LOG_DIR, TEMPLATE_LOCATION,
+        TEMPLATE_NAME, TEMPLATE_VOLUME_NAME,
         authentication::{
             AirflowAuthenticationClassResolved, AirflowClientAuthenticationDetailsResolved,
         },
@@ -475,7 +476,7 @@ pub async fn reconcile_airflow(
             let git_sync_resources = git_sync::v1alpha2::GitSyncResources::new(
                 &airflow.spec.cluster_config.dags_git_sync,
                 &validated_cluster.image,
-                &env_vars_from_overrides(&validated_rg_config.overrides.env_overrides),
+                &env_vars_from_overrides(&validated_rg_config.env_overrides),
                 &airflow.volume_mounts(),
                 LOG_VOLUME_NAME,
                 &validated_rg_config
@@ -534,7 +535,7 @@ pub async fn reconcile_airflow(
                 airflow,
                 &validated_cluster,
                 &rolegroup,
-                &validated_rg_config.overrides.config_file_overrides,
+                &validated_rg_config.config_overrides,
                 &validated_rg_config.merged_config.logging,
                 &Container::Airflow,
             )
@@ -611,7 +612,9 @@ async fn build_executor_template(
         airflow,
         validated_cluster,
         &rolegroup,
-        &BTreeMap::new(),
+        // The kubernetes-executor pod template does not apply webserver_config.py overrides
+        // (preserves prior behaviour, which passed an empty map here).
+        &AirflowConfigOverrides::default(),
         &merged_executor_config.logging,
         &Container::Base,
     )
@@ -730,7 +733,7 @@ fn build_server_rolegroup_statefulset(
     git_sync_resources: &git_sync::v1alpha2::GitSyncResources,
 ) -> Result<StatefulSet> {
     let merged_airflow_config = &validated_rg_config.merged_config;
-    let env_overrides = &validated_rg_config.overrides.env_overrides;
+    let env_overrides = &validated_rg_config.env_overrides;
 
     let resolved_product_image = &validated_cluster.image;
     let authentication_config = &validated_cluster.authentication_config;
diff --git a/rust/operator-binary/src/controller/build/config_map.rs b/rust/operator-binary/src/controller/build/config_map.rs
@@ -19,7 +19,8 @@ use crate::{
     config::webserver_config,
     controller::validate::ValidatedAirflowCluster,
     crd::{
-        AIRFLOW_CONFIG_FILENAME, Container, STACKABLE_LOG_DIR, build_recommended_labels, v1alpha2,
+        AIRFLOW_CONFIG_FILENAME, AirflowConfigOverrides, Container, STACKABLE_LOG_DIR,
+        build_recommended_labels, v1alpha2,
     },
     product_logging::{LOG_CONFIG_FILE, create_airflow_config},
 };
@@ -60,15 +61,24 @@ pub fn build_rolegroup_config_map(
     airflow: &v1alpha2::AirflowCluster,
     validated_cluster: &ValidatedAirflowCluster,
     rolegroup: &RoleGroupRef<v1alpha2::AirflowCluster>,
-    config_file_overrides: &BTreeMap<String, String>,
+    config_overrides: &AirflowConfigOverrides,
     logging: &Logging<Container>,
     container: &Container,
 ) -> Result<ConfigMap, Error> {
+    // Flatten the typed `webserver_config.py` overrides into a plain map for the file writer,
+    // dropping entries whose value is unset (`null`).
+    let config_file_overrides: BTreeMap<String, String> = config_overrides
+        .webserver_config_py
+        .overrides
+        .iter()
+        .filter_map(|(key, value)| value.clone().map(|value| (key.clone(), value)))
+        .collect();
+
     let config_file = webserver_config::build(
         &validated_cluster.authentication_config,
         &validated_cluster.authorization_config,
         &validated_cluster.image.product_version,
-        config_file_overrides,
+        &config_file_overrides,
     )
     .with_context(|_| BuildWebserverConfigSnafu {
         rolegroup: rolegroup.clone(),
diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs
@@ -1,8 +1,9 @@
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashMap};
 
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     commons::product_image_selection::{self, ResolvedProductImage},
+    config::merge::Merge,
     role_utils::RoleGroupRef,
 };
 use strum::IntoEnumIterator;
@@ -11,7 +12,7 @@ use super::dereference::DereferencedObjects;
 use crate::{
     airflow_controller::CONTAINER_IMAGE_BASE_NAME,
     crd::{
-        AirflowConfig, AirflowExecutor, AirflowRole, MergedOverrides,
+        AirflowConfig, AirflowConfigOverrides, AirflowExecutor, AirflowRole, AirflowRoleType,
         authentication::AirflowClientAuthenticationDetailsResolved,
         authorization::AirflowAuthorizationResolved, v1alpha2,
     },
@@ -37,10 +38,15 @@ pub struct ValidatedRoleConfig {
 }
 
 /// Per-rolegroup configuration: the merged CRD config plus overrides.
+///
+/// `config_overrides` is kept as the typed [`AirflowConfigOverrides`] (role-group merged over
+/// role); it is flattened into the rendered config file later, in the build step. This mirrors
+/// hdfs-operator. `env_overrides` is already a flat map.
 #[derive(Clone, Debug)]
 pub struct ValidatedRoleGroupConfig {
     pub merged_config: AirflowConfig,
-    pub overrides: MergedOverrides,
+    pub config_overrides: AirflowConfigOverrides,
+    pub env_overrides: HashMap<String, String>,
 }
 
 /// The validated cluster: proves that config merging succeeded for every role and
@@ -104,15 +110,15 @@ pub fn validate_cluster(
                 .merged_config(&role, &rolegroup_ref)
                 .context(FailedToResolveConfigSnafu)?;
 
-            let overrides = airflow
-                .merged_overrides(&role, rolegroup_name)
-                .context(FailedToResolveConfigSnafu)?;
+            let (config_overrides, env_overrides) =
+                merge_role_group_overrides(&resolved_role, rolegroup_name);
 
             group_configs.insert(
                 rolegroup_name.clone(),
                 ValidatedRoleGroupConfig {
                     merged_config,
-                    overrides,
+                    config_overrides,
+                    env_overrides,
                 },
             );
         }
@@ -134,3 +140,217 @@ pub fn validate_cluster(
         authorization_config,
     })
 }
+
+/// Merge a role group's config overrides over the role-level ones (role-group wins per key) via
+/// the `Merge` impl on [`AirflowConfigOverrides`], and combine env overrides (role first, then
+/// role-group on top). Mirrors hdfs-operator's `validate_role_group_config`.
+///
+/// The merged overrides are returned *typed*; flattening into the rendered `webserver_config.py`
+/// happens later, in the build step. Note the `Merge` semantics: a role-group `null` inherits the
+/// role-level value rather than unsetting it.
+fn merge_role_group_overrides(
+    role: &AirflowRoleType,
+    rolegroup_name: &str,
+) -> (AirflowConfigOverrides, HashMap<String, String>) {
+    let rolegroup = role.role_groups.get(rolegroup_name);
+
+    let mut config_overrides = rolegroup
+        .map(|rg| rg.config.config_overrides.clone())
+        .unwrap_or_default();
+    config_overrides.merge(&role.config.config_overrides);
+
+    let mut env_overrides = role.config.env_overrides.clone();
+    if let Some(rg) = rolegroup {
+        env_overrides.extend(rg.config.env_overrides.clone());
+    }
+
+    (config_overrides, env_overrides)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeMap;
+
+    use super::merge_role_group_overrides;
+    use crate::crd::{AirflowRole, v1alpha2};
+
+    fn test_cluster() -> v1alpha2::AirflowCluster {
+        let cluster_yaml = r#"
+        apiVersion: airflow.stackable.tech/v1alpha2
+        kind: AirflowCluster
+        metadata:
+          name: airflow
+        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: {}
+            configOverrides:
+              webserver_config.py:
+                AUTH_TYPE: "AUTH_OID"
+                ROLE_ONLY_KEY: "role-value"
+            envOverrides:
+              ROLE_ENV_VAR: "role-env-value"
+            roleGroups:
+              default:
+                config: {}
+                configOverrides:
+                  webserver_config.py:
+                    AUTH_TYPE: "AUTH_DB"
+                    GROUP_ONLY_KEY: "group-value"
+                envOverrides:
+                  GROUP_ENV_VAR: "group-env-value"
+          schedulers:
+            config: {}
+            roleGroups:
+              default:
+                config: {}
+          kubernetesExecutors:
+            config: {}
+        "#;
+        let deserializer = serde_yaml::Deserializer::from_str(cluster_yaml);
+        serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap()
+    }
+
+    #[test]
+    fn role_group_overrides_merge_over_role_overrides() {
+        let cluster = test_cluster();
+        let role = cluster
+            .get_role(&AirflowRole::Webserver)
+            .expect("webserver role");
+
+        let (config_overrides, env_overrides) = merge_role_group_overrides(&role, "default");
+
+        // configOverrides are kept typed (values are `Option<String>`). The role-group AUTH_TYPE
+        // overrides the role-level one; both role-only and group-only keys are kept.
+        assert_eq!(
+            config_overrides.webserver_config_py.overrides,
+            BTreeMap::from([
+                ("AUTH_TYPE".to_string(), Some("AUTH_DB".to_string())),
+                ("ROLE_ONLY_KEY".to_string(), Some("role-value".to_string())),
+                (
+                    "GROUP_ONLY_KEY".to_string(),
+                    Some("group-value".to_string())
+                ),
+            ])
+        );
+
+        assert_eq!(env_overrides.len(), 2);
+        assert_eq!(env_overrides.get("ROLE_ENV_VAR").unwrap(), "role-env-value");
+        assert_eq!(
+            env_overrides.get("GROUP_ENV_VAR").unwrap(),
+            "group-env-value"
+        );
+    }
+
+    /// A role-group `null` override inherits the role-level value instead of unsetting it — the
+    /// behavioural consequence of merging with `Merge` rather than `.extend()`. `main`'s
+    /// product-config used `.extend()`, where the same input would have *removed* the key. This
+    /// test pins that choice (mirrors the kafka-operator test).
+    #[test]
+    fn role_group_null_inherits_role_value_rather_than_unsetting_it() {
+        let cluster_yaml = r#"
+        apiVersion: airflow.stackable.tech/v1alpha2
+        kind: AirflowCluster
+        metadata:
+          name: airflow
+        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: {}
+            configOverrides:
+              webserver_config.py:
+                AUTH_TYPE: "AUTH_OID"
+            roleGroups:
+              default:
+                config: {}
+                configOverrides:
+                  webserver_config.py:
+                    AUTH_TYPE: null
+          schedulers:
+            config: {}
+            roleGroups:
+              default:
+                config: {}
+          kubernetesExecutors:
+            config: {}
+        "#;
+        let deserializer = serde_yaml::Deserializer::from_str(cluster_yaml);
+        let cluster: v1alpha2::AirflowCluster =
+            serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap();
+        let role = cluster
+            .get_role(&AirflowRole::Webserver)
+            .expect("webserver role");
+
+        // For contrast: under the old `.extend()` layering the role-group `null` overwrites the
+        // role value, and the key is dropped on flatten — i.e. AUTH_TYPE is unset entirely.
+        let old_extend_behaviour: BTreeMap<String, String> = {
+            let mut combined = role
+                .config
+                .config_overrides
+                .webserver_config_py
+                .overrides
+                .clone();
+            let rg = role.role_groups.get("default").expect("default role group");
+            combined.extend(
+                rg.config
+                    .config_overrides
+                    .webserver_config_py
+                    .overrides
+                    .clone(),
+            );
+            combined
+                .into_iter()
+                .filter_map(|(key, value)| value.map(|value| (key, value)))
+                .collect()
+        };
+        assert!(
+            !old_extend_behaviour.contains_key("AUTH_TYPE"),
+            "under the old `.extend()` behaviour the role-group `null` unsets AUTH_TYPE"
+        );
+
+        // What we do now (Merge): the role-group `null` inherits the role-level value, so
+        // AUTH_TYPE survives as the role's "AUTH_OID".
+        let (config_overrides, _env_overrides) = merge_role_group_overrides(&role, "default");
+        assert_eq!(
+            config_overrides
+                .webserver_config_py
+                .overrides
+                .get("AUTH_TYPE"),
+            Some(&Some("AUTH_OID".to_string())),
+            "role-group `null` should inherit the role-level AUTH_TYPE under Merge semantics"
+        );
+    }
+
+    #[test]
+    fn role_without_overrides_yields_empty() {
+        let cluster = test_cluster();
+        let role = cluster
+            .get_role(&AirflowRole::Scheduler)
+            .expect("scheduler role");
+
+        let (config_overrides, env_overrides) = merge_role_group_overrides(&role, "default");
+
+        assert!(config_overrides.webserver_config_py.overrides.is_empty());
+        assert!(env_overrides.is_empty());
+    }
+}
diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs
