move db connection resolution out of validate step

Author: adwk67

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

@@ -80,10 +80,8 @@ pub fn build_airflow_statefulset_envs(
     let auth_config = &cluster.cluster_config.authentication_config;
     let authorization_config = &cluster.cluster_config.authorization_config;
     let resolved_product_image = &cluster.image;
-    let metadata_database_connection_details =
-        &cluster.cluster_config.metadata_database_connection_details;
-    let celery_database_connection_details =
-        &cluster.cluster_config.celery_database_connection_details;
+    let metadata_database_connection_details = cluster.metadata_database_connection_details();
+    let celery_database_connection_details = cluster.celery_database_connection_details();
 
     let mut env: BTreeMap<String, EnvVar> = BTreeMap::new();
     let internal_secret_name = cluster.internal_secret_name();
@@ -370,8 +368,7 @@ pub fn build_airflow_template_envs(
             name: AIRFLOW_DATABASE_SQL_ALCHEMY_CONN.into(),
             value: Some(
                 cluster
-                    .cluster_config
-                    .metadata_database_connection_details
+                    .metadata_database_connection_details()
                     .url_template
                     .clone(),
             ),

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

@@ -145,8 +145,7 @@ pub fn build_executor_template_config_map(
     .context(PodSnafu)?;
 
     cluster
-        .cluster_config
-        .metadata_database_connection_details
+        .metadata_database_connection_details()
         .add_to_container(&mut airflow_container);
 
     pb.add_container(airflow_container.build());

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

@@ -256,12 +256,10 @@ pub fn build_server_rolegroup_statefulset(
     .context(PodSnafu)?;
 
     validated_cluster
-        .cluster_config
-        .metadata_database_connection_details
+        .metadata_database_connection_details()
         .add_to_container(&mut airflow_container);
-    if let Some((celery_result_backend, celery_broker)) = &validated_cluster
-        .cluster_config
-        .celery_database_connection_details
+    if let Some((celery_result_backend, celery_broker)) =
+        validated_cluster.celery_database_connection_details()
     {
         celery_result_backend.add_to_container(&mut airflow_container);
         celery_broker.add_to_container(&mut airflow_container);

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

@@ -11,8 +11,12 @@ use stackable_operator::{
         resources::{NoRuntimeLimits, Resources},
     },
     crd::git_sync,
-    database_connections::drivers::{
-        celery::CeleryDatabaseConnectionDetails, sqlalchemy::SqlAlchemyDatabaseConnectionDetails,
+    database_connections::{
+        TemplatingMechanism,
+        drivers::{
+            celery::CeleryDatabaseConnectionDetails,
+            sqlalchemy::SqlAlchemyDatabaseConnectionDetails,
+        },
     },
     k8s_openapi::api::core::v1::{PodTemplateSpec, Volume, VolumeMount},
     kube::{Resource, ResourceExt, api::ObjectMeta},
@@ -43,7 +47,11 @@ use crate::{
         APP_NAME, AirflowConfig, AirflowConfigOverrides, AirflowExecutor, AirflowRole,
         AirflowStorageConfig, ExecutorConfig, OPERATOR_NAME,
         authentication::AirflowClientAuthenticationDetailsResolved,
-        authorization::AirflowAuthorizationResolved, v1alpha2,
+        authorization::AirflowAuthorizationResolved,
+        databases::{
+            CeleryBrokerConnection, CeleryResultBackendConnection, MetadataDatabaseConnection,
+        },
+        v1alpha2,
     },
 };
 
@@ -148,13 +156,16 @@ pub struct ValidatedClusterConfig {
     pub load_examples: bool,
     pub expose_config: bool,
     pub database_initialization_enabled: bool,
-    /// The templated SQLAlchemy connection details for the metadata database.
-    pub metadata_database_connection_details: SqlAlchemyDatabaseConnectionDetails,
-    /// The templated Celery result-backend and broker connection details, when configured.
-    pub celery_database_connection_details: Option<(
-        CeleryDatabaseConnectionDetails,
-        CeleryDatabaseConnectionDetails,
-    )>,
+    /// The metadata database connection (`spec.clusterConfig.metadataDatabase`), as taken from the
+    /// CRD. The templated connection details are derived on demand, see
+    /// [`ValidatedCluster::metadata_database_connection_details`].
+    pub metadata_database: MetadataDatabaseConnection,
+    /// The Celery result-backend connection (`spec.clusterConfig.celeryResultsBackend`), when
+    /// configured. See [`ValidatedCluster::celery_database_connection_details`].
+    pub celery_results_backend: Option<CeleryResultBackendConnection>,
+    /// The Celery broker connection (`spec.clusterConfig.celeryBroker`), when configured. See
+    /// [`ValidatedCluster::celery_database_connection_details`].
+    pub celery_broker: Option<CeleryBrokerConnection>,
     /// User-supplied extra Volumes (`spec.clusterConfig.volumes`).
     pub volumes: Vec<Volume>,
     /// User-supplied extra VolumeMounts (`spec.clusterConfig.volumeMounts`).
@@ -255,6 +266,48 @@ impl ValidatedCluster {
         self.cluster_config.volume_mounts.clone()
     }
 
+    /// The templated SQLAlchemy connection details for the metadata database, derived from the
+    /// CRD connection ([`ValidatedClusterConfig::metadata_database`]).
+    pub fn metadata_database_connection_details(&self) -> SqlAlchemyDatabaseConnectionDetails {
+        self.cluster_config
+            .metadata_database
+            .sqlalchemy_connection_details_with_templating(
+                "METADATA",
+                &TemplatingMechanism::BashEnvSubstitution,
+            )
+    }
+
+    /// The templated Celery result-backend and broker connection details, derived from the CRD
+    /// connections ([`ValidatedClusterConfig::celery_results_backend`] /
+    /// [`ValidatedClusterConfig::celery_broker`]). `Some` only when both are configured, as the
+    /// Celery executor needs both.
+    pub fn celery_database_connection_details(
+        &self,
+    ) -> Option<(
+        CeleryDatabaseConnectionDetails,
+        CeleryDatabaseConnectionDetails,
+    )> {
+        let templating_mechanism = TemplatingMechanism::BashEnvSubstitution;
+        match (
+            &self.cluster_config.celery_results_backend,
+            &self.cluster_config.celery_broker,
+        ) {
+            (Some(celery_results_backend), Some(celery_broker)) => {
+                let celery_results_backend = celery_results_backend
+                    .celery_connection_details_with_templating(
+                        "CELERY_RESULT_BACKEND",
+                        &templating_mechanism,
+                    );
+                let celery_broker = celery_broker.celery_connection_details_with_templating(
+                    "CELERY_BROKER",
+                    &templating_mechanism,
+                );
+                Some((celery_results_backend, celery_broker))
+            }
+            _ => None,
+        }
+    }
+
     /// Type-safe names for the resources of a role group.
     ///
     /// Infallible: the combined name length was validated during cluster validation

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

@@ -4,13 +4,6 @@ use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
     commons::product_image_selection,
     config::fragment,
-    database_connections::{
-        TemplatingMechanism,
-        drivers::{
-            celery::CeleryDatabaseConnectionDetails,
-            sqlalchemy::SqlAlchemyDatabaseConnectionDetails,
-        },
-    },
     kube::ResourceExt,
     product_logging::spec::Logging,
     role_utils::{GenericRoleConfig, RoleGroup},
@@ -175,8 +168,21 @@ pub fn validate_cluster(
         authorization_config,
     } = dereferenced;
 
-    let (metadata_database_connection_details, celery_database_connection_details) =
-        database_connection_details(airflow);
+    // The Celery result-backend and broker only work with configured celeryExecutors. Emit a
+    // warning if they are provided without a Celery executor. The connection details themselves are
+    // derived from the CRD later, in the build step (see
+    // `ValidatedCluster::celery_database_connection_details`).
+    if airflow.spec.cluster_config.celery_results_backend.is_some()
+        && airflow.spec.cluster_config.celery_broker.is_some()
+        && !matches!(
+            &airflow.spec.executor,
+            AirflowExecutor::CeleryExecutors { .. }
+        )
+    {
+        tracing::warn!(
+            "No `spec.celeryExecutors` configured, but `spec.clusterConfig.celeryResultsBackend` and `spec.clusterConfig.celeryBroker` are provided. This only works in combination with a celery executor!"
+        )
+    }
 
     // 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.
@@ -214,8 +220,9 @@ pub fn validate_cluster(
             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_connection_details,
-            celery_database_connection_details,
+            metadata_database: airflow.spec.cluster_config.metadata_database.clone(),
+            celery_results_backend: airflow.spec.cluster_config.celery_results_backend.clone(),
+            celery_broker: airflow.spec.cluster_config.celery_broker.clone(),
             load_examples: airflow.spec.cluster_config.load_examples,
             expose_config: airflow.spec.cluster_config.expose_config,
             database_initialization_enabled: airflow
@@ -231,61 +238,6 @@ pub fn validate_cluster(
     ))
 }
 
-/// Builds the templated metadata-database (and optional Celery result-backend/broker) connection
-/// details from the cluster spec.
-fn database_connection_details(
-    airflow: &v1alpha2::AirflowCluster,
-) -> (
-    SqlAlchemyDatabaseConnectionDetails,
-    Option<(
-        CeleryDatabaseConnectionDetails,
-        CeleryDatabaseConnectionDetails,
-    )>,
-) {
-    let templating_mechanism = TemplatingMechanism::BashEnvSubstitution;
-
-    let metadata_database_connection_details = airflow
-        .spec
-        .cluster_config
-        .metadata_database
-        .sqlalchemy_connection_details_with_templating("METADATA", &templating_mechanism);
-
-    let celery_database_connection_details = if let (
-        Some(celery_results_backend),
-        Some(celery_broker),
-    ) = (
-        &airflow.spec.cluster_config.celery_results_backend,
-        &airflow.spec.cluster_config.celery_broker,
-    ) {
-        // The celery results backend and celery broker only work with configured celeryExecutors.
-        // Emit a warning if celery executors were not configured properly.
-        if !matches!(
-            &airflow.spec.executor,
-            AirflowExecutor::CeleryExecutors { .. }
-        ) {
-            tracing::warn!(
-                "No `spec.celeryExecutors` configured, but `spec.clusterConfig.celeryResultsBackend` and `spec.clusterConfig.celeryBroker` are provided. This only works in combination with a celery executor!"
-            )
-        }
-
-        let celery_results_backend = celery_results_backend
-            .celery_connection_details_with_templating(
-                "CELERY_RESULT_BACKEND",
-                &templating_mechanism,
-            );
-        let celery_broker = celery_broker
-            .celery_connection_details_with_templating("CELERY_BROKER", &templating_mechanism);
-        Some((celery_results_backend, celery_broker))
-    } else {
-        None
-    };
-
-    (
-        metadata_database_connection_details,
-        celery_database_connection_details,
-    )
-}
-
 /// Validate and merge one role group against its role, via the shared
 /// [`with_validated_config`] from `operator-rs`, returning the generic
 /// [`stackable_operator::v2::role_utils::RoleGroupConfig`].