chore: move results backend and broker to clusterConfig

Author: maltesander

rust/operator-binary/src/airflow_controller.rs

@@ -393,23 +393,23 @@ pub async fn reconcile_airflow(
         .cluster_config
         .metadata_database
         .sqlalchemy_connection_details_with_templating("METADATA", &templating_mechanism);
-    let celery_database_connection_details = match &airflow.spec.executor {
-        AirflowExecutor::CeleryExecutors {
-            result_backend: celery_result_backend,
-            broker: celery_broker,
-            ..
-        } => {
-            let celery_result_backend = celery_result_backend
+
+    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,
+        ) {
+            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_result_backend, celery_broker))
-        }
-        _ => None,
-    };
+            Some((celery_results_backend, celery_broker))
+        } else {
+            None
+        };
 
     let mut roles = HashMap::new();
 

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

@@ -309,6 +309,14 @@ pub mod versioned {
         #[serde(default)]
         pub database_initialization: DatabaseInitializationConfig,
 
+        #[serde(skip_serializing_if = "Option::is_none")]
+        /// Connection information for the celery backend database.
+        pub celery_results_backend: Option<CeleryResultBackendConnection>,
+
+        #[serde(skip_serializing_if = "Option::is_none")]
+        /// Connection information for the celery broker queue.
+        pub celery_broker: Option<CeleryBrokerConnection>,
+
         /// Name of the Vector aggregator [discovery ConfigMap](DOCS_BASE_URL_PLACEHOLDER/concepts/service_discovery).
         /// It must contain the key `ADDRESS` with the address of the Vector aggregator.
         /// Follow the [logging tutorial](DOCS_BASE_URL_PLACEHOLDER/tutorials/logging-vector-aggregator)
@@ -857,12 +865,6 @@ pub enum AirflowExecutor {
     CeleryExecutors {
         #[serde(flatten)]
         config: Box<AirflowRoleType>,
-
-        /// Connection information for the celery backend database.
-        result_backend: CeleryResultBackendConnection,
-
-        /// Connection information for the celery broker queue.
-        broker: CeleryBrokerConnection,
     },
 
     /// With the Kubernetes executor, executor Pods are created on demand.