refactor: remove product-config validation from reconcile

Replaces transform_all_roles_to_config / validate_all_roles_and_groups_config
with direct role iteration plus merged_overrides(). Folds the dereferenced
authentication/authorization objects into ValidatedAirflowCluster so downstream
build steps read them from the validated cluster. The product-config crate is
still used for the Flask config writer (removed in a later step).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: adwk67

rust/operator-binary/src/airflow_controller.rs

@@ -6,11 +6,7 @@ use std::{
 };
 
 use const_format::concatcp;
-use product_config::{
-    ProductConfigManager,
-    flask_app_config_writer::{self, FlaskAppConfigWriterError},
-    types::PropertyNameKind,
-};
+use product_config::flask_app_config_writer::{self, FlaskAppConfigWriterError};
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
     builder::{
@@ -61,10 +57,7 @@ use stackable_operator::{
     },
     kvp::{Annotation, Label, LabelError, Labels, ObjectLabels},
     logging::controller::ReconcilerError,
-    product_config_utils::{
-        CONFIG_OVERRIDE_FILE_FOOTER_KEY, CONFIG_OVERRIDE_FILE_HEADER_KEY, env_vars_from,
-        env_vars_from_rolegroup_config,
-    },
+    product_config_utils::env_vars_from,
     product_logging::{
         self,
         framework::LoggingError,
@@ -118,9 +111,11 @@ pub const CONTAINER_IMAGE_BASE_NAME: &str = "airflow";
 pub const AIRFLOW_FULL_CONTROLLER_NAME: &str =
     concatcp!(AIRFLOW_CONTROLLER_NAME, '.', OPERATOR_NAME);
 
+const CONFIG_OVERRIDE_FILE_HEADER_KEY: &str = "FILE_HEADER";
+const CONFIG_OVERRIDE_FILE_FOOTER_KEY: &str = "FILE_FOOTER";
+
 pub struct Ctx {
     pub client: stackable_operator::client::Client,
-    pub product_config: ProductConfigManager,
     pub operator_environment: OperatorEnvironmentOptions,
 }
 
@@ -386,7 +381,7 @@ pub async fn reconcile_airflow(
         CONTAINER_IMAGE_BASE_NAME,
         &ctx.operator_environment.image_repository,
         crate::built_info::PKG_VERSION,
-        &ctx.product_config,
+        dereferenced,
     )
     .context(ValidateSnafu)?;
 
@@ -464,8 +459,8 @@ pub async fn reconcile_airflow(
             common_configuration,
             &metadata_database_connection_details,
             &validated.image,
-            &dereferenced.authentication_config,
-            &dereferenced.authorization_config,
+            &validated.authentication_config,
+            &validated.authorization_config,
             &mut cluster_resources,
             client,
             &rbac_sa,
@@ -515,7 +510,7 @@ pub async fn reconcile_airflow(
             let git_sync_resources = git_sync::v1alpha2::GitSyncResources::new(
                 &airflow.spec.cluster_config.dags_git_sync,
                 &validated.image,
-                &env_vars_from_rolegroup_config(&validated_rg_config.product_config_properties),
+                &env_vars_from(&validated_rg_config.overrides.env_overrides),
                 &airflow.volume_mounts(),
                 LOG_VOLUME_NAME,
                 &validated_rg_config
@@ -574,9 +569,9 @@ pub async fn reconcile_airflow(
                 airflow,
                 &validated.image,
                 &rolegroup,
-                &validated_rg_config.product_config_properties,
-                &dereferenced.authentication_config,
-                &dereferenced.authorization_config,
+                &validated_rg_config.overrides.config_file_overrides,
+                &validated.authentication_config,
+                &validated.authorization_config,
                 &validated_rg_config.merged_config.logging,
                 &Container::Airflow,
             )?;
@@ -592,9 +587,9 @@ pub async fn reconcile_airflow(
                 &validated.image,
                 airflow_role,
                 &rolegroup,
-                &validated_rg_config.product_config_properties,
-                &dereferenced.authentication_config,
-                &dereferenced.authorization_config,
+                &validated_rg_config.overrides.env_overrides,
+                &validated.authentication_config,
+                &validated.authorization_config,
                 &metadata_database_connection_details,
                 &celery_database_connection_details,
                 &rbac_sa,
@@ -659,7 +654,7 @@ async fn build_executor_template(
         airflow,
         resolved_product_image,
         &rolegroup,
-        &HashMap::new(),
+        &BTreeMap::new(),
         authentication_config,
         authorization_config,
         &merged_executor_config.logging,
@@ -709,7 +704,7 @@ fn build_rolegroup_config_map(
     airflow: &v1alpha2::AirflowCluster,
     resolved_product_image: &ResolvedProductImage,
     rolegroup: &RoleGroupRef<v1alpha2::AirflowCluster>,
-    rolegroup_config: &HashMap<PropertyNameKind, BTreeMap<String, String>>,
+    config_file_overrides: &BTreeMap<String, String>,
     authentication_config: &AirflowClientAuthenticationDetailsResolved,
     authorization_config: &AirflowAuthorizationResolved,
     logging: &Logging<Container>,
@@ -732,10 +727,7 @@ fn build_rolegroup_config_map(
         config
     );
 
-    let mut file_config = rolegroup_config
-        .get(&PropertyNameKind::File(AIRFLOW_CONFIG_FILENAME.to_string()))
-        .cloned()
-        .unwrap_or_default();
+    let mut file_config = config_file_overrides.clone();
 
     tracing::debug!(
         "Config overrides for {}: {:?}",
@@ -884,7 +876,7 @@ fn build_server_rolegroup_statefulset(
     resolved_product_image: &ResolvedProductImage,
     airflow_role: &AirflowRole,
     rolegroup_ref: &RoleGroupRef<v1alpha2::AirflowCluster>,
-    rolegroup_config: &HashMap<PropertyNameKind, BTreeMap<String, String>>,
+    env_overrides: &HashMap<String, String>,
     authentication_config: &AirflowClientAuthenticationDetailsResolved,
     authorization_config: &AirflowAuthorizationResolved,
     metadata_database_connection_details: &SqlAlchemyDatabaseConnectionDetails,
@@ -974,7 +966,7 @@ fn build_server_rolegroup_statefulset(
         env_vars::build_airflow_statefulset_envs(
             airflow,
             airflow_role,
-            rolegroup_config,
+            env_overrides,
             executor,
             authentication_config,
             authorization_config,

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

@@ -1,18 +1,18 @@
-use std::{
-    collections::{BTreeMap, HashMap},
-    str::FromStr,
-};
+use std::collections::BTreeMap;
 
-use product_config::{ProductConfigManager, types::PropertyNameKind};
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     commons::product_image_selection::{self, ResolvedProductImage},
-    product_config_utils::{transform_all_roles_to_config, validate_all_roles_and_groups_config},
     role_utils::RoleGroupRef,
 };
 use strum::IntoEnumIterator;
 
-use crate::crd::{AIRFLOW_CONFIG_FILENAME, AirflowConfig, AirflowExecutor, AirflowRole, v1alpha2};
+use super::dereference::DereferencedObjects;
+use crate::crd::{
+    AirflowConfig, AirflowExecutor, AirflowRole, MergedOverrides,
+    authentication::AirflowClientAuthenticationDetailsResolved,
+    authorization::AirflowAuthorizationResolved, v1alpha2,
+};
 
 #[derive(Snafu, Debug)]
 pub enum Error {
@@ -21,24 +21,8 @@ pub enum Error {
         source: product_image_selection::Error,
     },
 
-    #[snafu(display("invalid product config"))]
-    InvalidProductConfig {
-        source: stackable_operator::product_config_utils::Error,
-    },
-
-    #[snafu(display("failed to generate product config"))]
-    GenerateProductConfig {
-        source: stackable_operator::product_config_utils::Error,
-    },
-
     #[snafu(display("failed to resolve and merge config for role and role group"))]
     FailedToResolveConfig { source: crate::crd::Error },
-
-    #[snafu(display("could not parse Airflow role [{role}]"))]
-    UnidentifiedAirflowRole {
-        source: strum::ParseError,
-        role: String,
-    },
 }
 
 /// Per-role configuration extracted during validation.
@@ -49,115 +33,99 @@ pub struct ValidatedRoleConfig {
     pub group_listener_name: Option<String>,
 }
 
-/// Per-rolegroup configuration: the merged CRD config plus the product-config properties.
+/// Per-rolegroup configuration: the merged CRD config plus overrides.
 #[derive(Clone, Debug)]
 pub struct ValidatedRoleGroupConfig {
     pub merged_config: AirflowConfig,
-    pub product_config_properties: HashMap<PropertyNameKind, BTreeMap<String, String>>,
+    pub overrides: MergedOverrides,
 }
 
-/// The validated cluster: proves that product-config validation and config merging
-/// succeeded for every role and role group before any resources are created.
+/// The validated cluster: proves that config merging succeeded for every role and
+/// role group before any resources are created. It also carries the dereferenced
+/// external references, so every downstream build step reads them from here.
 #[derive(Clone, Debug)]
 pub struct ValidatedAirflowCluster {
     pub image: ResolvedProductImage,
     pub role_groups: BTreeMap<AirflowRole, BTreeMap<String, ValidatedRoleGroupConfig>>,
     pub role_configs: BTreeMap<AirflowRole, ValidatedRoleConfig>,
     pub executor: AirflowExecutor,
+    pub authentication_config: AirflowClientAuthenticationDetailsResolved,
+    pub authorization_config: AirflowAuthorizationResolved,
 }
 
 pub fn validate_cluster(
     airflow: &v1alpha2::AirflowCluster,
     image_base_name: &str,
     image_repository: &str,
     pkg_version: &str,
-    product_config_manager: &ProductConfigManager,
+    dereferenced: DereferencedObjects,
 ) -> Result<ValidatedAirflowCluster, Error> {
     let resolved_product_image = airflow
         .spec
         .image
         .resolve(image_base_name, image_repository, pkg_version)
         .context(ResolveProductImageSnafu)?;
 
-    let mut roles = HashMap::new();
+    let mut role_groups = BTreeMap::new();
+    let mut role_configs = BTreeMap::new();
 
     // if the kubernetes executor is specified there will be no worker role as the pods
     // are provisioned by airflow as defined by the task (default: one pod per task)
     for role in AirflowRole::iter() {
-        if let Some(resolved_role) = airflow.get_role(&role) {
-            roles.insert(
-                role.to_string(),
-                (
-                    vec![
-                        PropertyNameKind::Env,
-                        PropertyNameKind::File(AIRFLOW_CONFIG_FILENAME.into()),
-                    ],
-                    resolved_role.clone(),
-                ),
-            );
-        }
-    }
-
-    let role_config = transform_all_roles_to_config(airflow, &roles);
-    let validated_role_config = validate_all_roles_and_groups_config(
-        &resolved_product_image.product_version,
-        &role_config.context(GenerateProductConfigSnafu)?,
-        product_config_manager,
-        false,
-        false,
-    )
-    .context(InvalidProductConfigSnafu)?;
-
-    let mut role_groups = BTreeMap::new();
-    let mut role_configs = BTreeMap::new();
-
-    for (role_name, rolegroup_configs) in validated_role_config.iter() {
-        let airflow_role =
-            AirflowRole::from_str(role_name).context(UnidentifiedAirflowRoleSnafu {
-                role: role_name.to_string(),
-            })?;
+        let Some(resolved_role) = airflow.get_role(&role) else {
+            continue;
+        };
 
         role_configs.insert(
-            airflow_role.clone(),
+            role.clone(),
             ValidatedRoleConfig {
                 pdb: airflow
-                    .role_config(&airflow_role)
+                    .role_config(&role)
                     .map(|rc| rc.pod_disruption_budget),
-                listener_class: airflow_role
-                    .listener_class_name(airflow)
-                    .map(|s| s.to_string()),
-                group_listener_name: airflow.group_listener_name(&airflow_role),
+                listener_class: role.listener_class_name(airflow).map(|s| s.to_string()),
+                group_listener_name: airflow.group_listener_name(&role),
             },
         );
 
         let mut group_configs = BTreeMap::new();
-        for (rolegroup_name, rolegroup_config) in rolegroup_configs.iter() {
+        for rolegroup_name in resolved_role.role_groups.keys() {
             let rolegroup_ref = RoleGroupRef {
                 cluster: stackable_operator::kube::runtime::reflector::ObjectRef::from_obj(airflow),
-                role: role_name.into(),
+                role: role.to_string(),
                 role_group: rolegroup_name.into(),
             };
 
             let merged_config = airflow
-                .merged_config(&airflow_role, &rolegroup_ref)
+                .merged_config(&role, &rolegroup_ref)
+                .context(FailedToResolveConfigSnafu)?;
+
+            let overrides = airflow
+                .merged_overrides(&role, rolegroup_name)
                 .context(FailedToResolveConfigSnafu)?;
 
             group_configs.insert(
                 rolegroup_name.clone(),
                 ValidatedRoleGroupConfig {
                     merged_config,
-                    product_config_properties: rolegroup_config.clone(),
+                    overrides,
                 },
             );
         }
 
-        role_groups.insert(airflow_role, group_configs);
+        role_groups.insert(role, group_configs);
     }
 
+    let DereferencedObjects {
+        authentication_config,
+        authorization_config,
+    } = dereferenced;
+
     Ok(ValidatedAirflowCluster {
         image: resolved_product_image,
         role_groups,
         role_configs,
         executor: airflow.spec.executor.clone(),
+        authentication_config,
+        authorization_config,
     })
 }

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

@@ -2,6 +2,7 @@ use stackable_operator::{client::Client, commons::opa::OpaApiVersion, shared::ti
 
 use crate::crd::{AirflowAuthorization, AirflowOpaConfig, v1alpha2};
 
+#[derive(Clone, Debug)]
 pub struct AirflowAuthorizationResolved {
     pub opa: Option<OpaConfigResolved>,
 }
@@ -24,6 +25,7 @@ impl AirflowAuthorizationResolved {
     }
 }
 
+#[derive(Clone, Debug)]
 pub struct OpaConfigResolved {
     pub connection_string: String,
     pub cache_entry_time_to_live: Duration,