git_sync::v1alpha2::GitSyncResources usage inline with nifi

adwk67 · adwk67 · commit 45470161e295 · 2026-06-24T17:16:35.000+02:00
diff --git a/rust/operator-binary/src/airflow_controller.rs b/rust/operator-binary/src/airflow_controller.rs
@@ -10,7 +10,6 @@ use stackable_operator::{
     cli::OperatorEnvironmentOptions,
     cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
     commons::{random_secret_creation, rbac::build_rbac_resources},
-    crd::git_sync,
     k8s_openapi::api::core::v1::EnvVar,
     kube::{
         ResourceExt,
@@ -34,16 +33,13 @@ 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,
-            },
-            volumes::LOG_VOLUME_NAME,
+        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,
     },
@@ -135,9 +131,6 @@ pub enum Error {
         source: crate::controller::build::resource::config_map::Error,
     },
 
-    #[snafu(display("invalid git-sync specification"))]
-    InvalidGitSyncSpec { source: git_sync::v1alpha2::Error },
-
     #[snafu(display("failed to delete orphaned resources"))]
     DeleteOrphanedResources {
         source: stackable_operator::cluster_resources::Error,
@@ -340,16 +333,6 @@ pub async fn reconcile_airflow(
         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()),
-                &validated_cluster.volume_mounts(),
-                LOG_VOLUME_NAME.as_ref(),
-                &logging.git_sync_container,
-            )
-            .context(InvalidGitSyncSpecSnafu)?;
-
             let rg_headless_service =
                 build_rolegroup_headless_service(&validated_cluster, airflow_role, rolegroup_name);
 
@@ -392,7 +375,6 @@ pub async fn reconcile_airflow(
                 validated_rg_config,
                 logging,
                 &rbac_sa,
-                &git_sync_resources,
             )
             .context(BuildStatefulSetSnafu)?;
 
@@ -452,23 +434,12 @@ async fn build_executor_template(
             role_group: executor_role_group_name(),
         })?;
 
-    let git_sync_resources = git_sync::v1alpha2::GitSyncResources::new(
-        &validated_cluster.cluster_config.dags_git_sync,
-        &validated_cluster.image,
-        &env_vars_from_overrides(&executor_template.env_overrides),
-        &validated_cluster.volume_mounts(),
-        LOG_VOLUME_NAME.as_ref(),
-        &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(),
         executor_config,
         &executor_template.env_overrides,
         &executor_template.pod_overrides,
-        &git_sync_resources,
     )
     .context(BuildExecutorTemplateSnafu)?;
     cluster_resources
@@ -492,7 +463,7 @@ pub fn error_policy(
 }
 
 /// Convert user-supplied `envOverrides` into a list of [`EnvVar`]s.
-fn env_vars_from_overrides(env_overrides: &HashMap<String, String>) -> Vec<EnvVar> {
+pub(crate) fn env_vars_from_overrides(env_overrides: &HashMap<String, String>) -> Vec<EnvVar> {
     // Collect into a `BTreeMap` first so the env vars come out in a deterministic (sorted) order;
     // `HashMap` iteration order is randomised per instance and would otherwise churn the containers
     // this feeds between reconciles. Mirrors the override handling in `env_vars.rs`.
diff --git a/rust/operator-binary/src/controller/build/resource/executor.rs b/rust/operator-binary/src/controller/build/resource/executor.rs
@@ -10,7 +10,6 @@ use stackable_operator::{
         meta::ObjectMetaBuilder,
         pod::{PodBuilder, security::PodSecurityContextBuilder},
     },
-    crd::git_sync,
     k8s_openapi::{
         DeepMerge,
         api::core::v1::{ConfigMap, PodTemplateSpec},
@@ -83,10 +82,12 @@ pub fn build_executor_template_config_map(
     executor_config: &ValidatedAirflowConfig,
     env_overrides: &HashMap<String, String>,
     pod_overrides: &PodTemplateSpec,
-    git_sync_resources: &git_sync::v1alpha2::GitSyncResources,
 ) -> Result<ConfigMap> {
     let resolved_product_image = &cluster.image;
     let authentication_config = &cluster.cluster_config.authentication_config;
+    // The git-sync resources were resolved up-front during validation; read them off the validated
+    // executor config rather than reconstructing them here.
+    let git_sync_resources = &executor_config.git_sync_resources;
 
     let mut pb = PodBuilder::new();
     let pb_metadata =
diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs
@@ -7,7 +7,6 @@ use stackable_operator::{
             security::PodSecurityContextBuilder, volume::VolumeBuilder,
         },
     },
-    crd::git_sync,
     k8s_openapi::{
         DeepMerge,
         api::{
@@ -105,10 +104,12 @@ pub fn build_server_rolegroup_statefulset(
     validated_rg_config: &AirflowRoleGroupConfig,
     logging: &ValidatedLogging,
     service_account: &ServiceAccount,
-    git_sync_resources: &git_sync::v1alpha2::GitSyncResources,
 ) -> Result<StatefulSet> {
     let merged_airflow_config = &validated_rg_config.config;
     let env_overrides = &validated_rg_config.env_overrides;
+    // The git-sync resources were resolved up-front during validation; read them off the validated
+    // config rather than reconstructing them here.
+    let git_sync_resources = &merged_airflow_config.git_sync_resources;
 
     let resolved_product_image = &validated_cluster.image;
     let authentication_config = &validated_cluster.cluster_config.authentication_config;
diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs
@@ -82,35 +82,51 @@ pub type AirflowRoleGroupConfig = stackable_operator::v2::role_utils::RoleGroupC
 /// 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)]
+// Not `Clone`/`Debug`/`PartialEq`: `git_sync_resources` (a `GitSyncResources`) implements none of
+// them, mirroring nifi's `ValidatedNifiConfig`.
 pub struct ValidatedAirflowConfig {
     pub resources: Resources<AirflowStorageConfig, NoRuntimeLimits>,
     pub logging: ValidatedLogging,
     pub affinity: StackableAffinity,
     pub graceful_shutdown_timeout: Option<Duration>,
+    /// The git-sync resources (containers, volumes, mounts) for the DAGs, resolved up-front in the
+    /// [`validate`] step (the env vars and logging differ per role group / executor, so they are
+    /// computed there). Consumed by the StatefulSet, executor-template and env-var builders, which
+    /// read it off here rather than reconstructing it.
+    pub git_sync_resources: git_sync::v1alpha2::GitSyncResources,
 }
 
 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 {
+    pub(crate) fn from_merged(
+        merged: AirflowConfig,
+        logging: ValidatedLogging,
+        git_sync_resources: git_sync::v1alpha2::GitSyncResources,
+    ) -> Self {
         Self {
             resources: merged.resources,
             logging,
             affinity: merged.affinity,
             graceful_shutdown_timeout: merged.graceful_shutdown_timeout,
+            git_sync_resources,
         }
     }
 
     /// 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 {
+    pub(crate) fn from_merged_executor(
+        merged: ExecutorConfig,
+        logging: ValidatedLogging,
+        git_sync_resources: git_sync::v1alpha2::GitSyncResources,
+    ) -> Self {
         Self {
             resources: merged.resources,
             logging,
             affinity: merged.affinity,
             graceful_shutdown_timeout: merged.graceful_shutdown_timeout,
+            git_sync_resources,
         }
     }
 }
@@ -150,8 +166,6 @@ pub struct ValidatedClusterConfig {
     pub executor_template: Option<ValidatedExecutorTemplate>,
     pub authentication_config: AirflowClientAuthenticationDetailsResolved,
     pub authorization_config: AirflowAuthorizationResolved,
-    /// The Git-sync definitions for the DAGs (`spec.clusterConfig.dagsGitSync`).
-    pub dags_git_sync: Vec<git_sync::v1alpha2::GitSync>,
     pub credentials_secret_name: SecretName,
     pub load_examples: bool,
     pub expose_config: bool,
diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs
@@ -4,6 +4,8 @@ use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
     commons::product_image_selection,
     config::fragment,
+    crd::git_sync,
+    k8s_openapi::api::core::v1::{EnvVar, VolumeMount},
     kube::ResourceExt,
     product_logging::spec::Logging,
     role_utils::{GenericRoleConfig, RoleGroup},
@@ -25,10 +27,10 @@ use strum::IntoEnumIterator;
 use super::{
     AirflowRoleGroupConfig, ValidatedAirflowConfig, ValidatedCluster, ValidatedClusterConfig,
     ValidatedExecutorTemplate, ValidatedLogging, ValidatedRoleConfig,
-    dereference::DereferencedObjects,
+    build::volumes::LOG_VOLUME_NAME, dereference::DereferencedObjects,
 };
 use crate::{
-    airflow_controller::CONTAINER_IMAGE_BASE_NAME,
+    airflow_controller::{CONTAINER_IMAGE_BASE_NAME, env_vars_from_overrides},
     crd::{
         AirflowConfig, AirflowConfigFragment, AirflowConfigOverrides, AirflowExecutor, AirflowRole,
         AirflowRoleType, Container, v1alpha2,
@@ -87,6 +89,9 @@ pub enum Error {
         "the Vector aggregator discovery ConfigMap name must be set when the Vector agent is enabled"
     ))]
     MissingVectorAggregatorConfigMapName,
+
+    #[snafu(display("invalid git-sync specification"))]
+    InvalidGitSyncSpec { source: git_sync::v1alpha2::Error },
 }
 
 pub fn validate_cluster(
@@ -155,6 +160,9 @@ pub fn validate_cluster(
                 rolegroup,
                 &default_config,
                 &vector_aggregator_config_map_name,
+                &resolved_product_image,
+                &airflow.spec.cluster_config.dags_git_sync,
+                &airflow.spec.cluster_config.volume_mounts,
             )?;
 
             group_configs.insert(role_group_name, config);
@@ -199,8 +207,22 @@ pub fn validate_cluster(
                 &Container::Base,
                 &vector_aggregator_config_map_name,
             )?;
+            // Resolve the executor's git-sync resources up-front too, mirroring the role groups.
+            let git_sync_resources = git_sync::v1alpha2::GitSyncResources::new(
+                &airflow.spec.cluster_config.dags_git_sync,
+                &resolved_product_image,
+                &env_vars_from_overrides(&common_configuration.env_overrides),
+                &airflow.spec.cluster_config.volume_mounts,
+                LOG_VOLUME_NAME.as_ref(),
+                &logging.git_sync_container,
+            )
+            .context(InvalidGitSyncSpecSnafu)?;
             Some(ValidatedExecutorTemplate {
-                config: ValidatedAirflowConfig::from_merged_executor(merged, logging),
+                config: ValidatedAirflowConfig::from_merged_executor(
+                    merged,
+                    logging,
+                    git_sync_resources,
+                ),
                 env_overrides: common_configuration.env_overrides.clone(),
                 pod_overrides: common_configuration.pod_overrides.clone(),
             })
@@ -218,7 +240,6 @@ pub fn validate_cluster(
             executor_template,
             authentication_config,
             authorization_config,
-            dags_git_sync: airflow.spec.cluster_config.dags_git_sync.clone(),
             credentials_secret_name: airflow.spec.cluster_config.credentials_secret_name.clone(),
             metadata_database: airflow.spec.cluster_config.metadata_database.clone(),
             celery_results_backend: airflow.spec.cluster_config.celery_results_backend.clone(),
@@ -249,12 +270,16 @@ pub fn validate_cluster(
 ///
 /// Note the override `Merge` semantics: a role-group `null` inherits the role-level value rather
 /// than unsetting it (config overrides), and env overrides layer role-group on top of role.
+#[allow(clippy::too_many_arguments)]
 fn validate_role_group(
     role: &AirflowRoleType,
     role_group_name: &RoleGroupName,
     rolegroup: &RoleGroup<AirflowConfigFragment, GenericCommonConfig, AirflowConfigOverrides>,
     default_config: &AirflowConfigFragment,
     vector_aggregator_config_map_name: &Option<ConfigMapName>,
+    image: &product_image_selection::ResolvedProductImage,
+    dags_git_sync: &[git_sync::v1alpha2::GitSync],
+    volume_mounts: &[VolumeMount],
 ) -> Result<AirflowRoleGroupConfig, Error> {
     let validated = with_validated_config::<
         AirflowConfig,
@@ -282,9 +307,21 @@ fn validate_role_group(
         vector_aggregator_config_map_name,
     )?;
 
+    // The git-sync resources depend on this role group's env-var overrides and (git-sync) logging
+    // config, so they are resolved (and validated) here, up-front, rather than at build time.
+    let git_sync_resources = git_sync::v1alpha2::GitSyncResources::new(
+        dags_git_sync,
+        image,
+        &Vec::<EnvVar>::from(env_overrides.clone()),
+        volume_mounts,
+        LOG_VOLUME_NAME.as_ref(),
+        &logging.git_sync_container,
+    )
+    .context(InvalidGitSyncSpecSnafu)?;
+
     Ok(RoleGroupConfig {
         replicas: validated.replicas,
-        config: ValidatedAirflowConfig::from_merged(merged_config, logging),
+        config: ValidatedAirflowConfig::from_merged(merged_config, logging, git_sync_resources),
         config_overrides: validated.config.config_overrides,
         env_overrides,
         cli_overrides: validated.config.cli_overrides,
@@ -342,6 +379,20 @@ mod tests {
     use super::validate_role_group;
     use crate::crd::{AirflowConfig, AirflowRole, v1alpha2};
 
+    /// A minimal resolved product image for tests that exercise `validate_role_group` (which needs
+    /// one to resolve git-sync resources; the test role groups configure no git-sync, so the value
+    /// is otherwise unused).
+    fn test_resolved_product_image()
+    -> stackable_operator::commons::product_image_selection::ResolvedProductImage {
+        stackable_operator::commons::product_image_selection::ResolvedProductImage {
+            product_version: "3.0.6".to_string(),
+            app_version_label_value: "3.0.6".parse().expect("valid label value"),
+            image: "oci.example.org/sdp/airflow:3.0.6-stackable0.0.0-dev".to_string(),
+            image_pull_policy: "IfNotPresent".to_string(),
+            pull_secrets: None,
+        }
+    }
+
     fn test_cluster() -> v1alpha2::AirflowCluster {
         let cluster_yaml = r#"
         apiVersion: airflow.stackable.tech/v1alpha2
@@ -404,6 +455,9 @@ mod tests {
             rolegroup,
             &default_config,
             &None,
+            &test_resolved_product_image(),
+            &[],
+            &[],
         )
         .expect("validated role group");
         let config_overrides = validated.config_overrides;
@@ -502,6 +556,9 @@ mod tests {
             rolegroup,
             &default_config,
             &None,
+            &test_resolved_product_image(),
+            &[],
+            &[],
         )
         .expect("validated role group");
 
@@ -576,6 +633,9 @@ mod tests {
             rolegroup,
             &default_config,
             &None,
+            &test_resolved_product_image(),
+            &[],
+            &[],
         )
         .expect("validated role group");
 
