add flag for db init routine incl. test asserts as part of cluster ops

adwk67 · adwk67 · commit 81e707cf9948 · 2025-08-06T14:30:52.000+02:00
diff --git a/deploy/helm/airflow-operator/crds/crds.yaml b/deploy/helm/airflow-operator/crds/crds.yaml
@@ -591,6 +591,10 @@ spec:
                           - repo
                         type: object
                       type: array
+                    dbInit:
+                      default: true
+                      description: Whether to execute the database initialization routines (a combination of database initialization, upgrade and migration depending on the Airflow version). Defaults to true to be backwward compatible.
+                      type: boolean
                     exposeConfig:
                       default: false
                       description: for internal use only - not for production use.
diff --git a/rust/operator-binary/src/airflow_controller.rs b/rust/operator-binary/src/airflow_controller.rs
@@ -951,8 +951,11 @@ fn build_server_rolegroup_statefulset(
         .context(GracefulShutdownSnafu)?;
 
     let mut airflow_container_args = Vec::new();
-    airflow_container_args
-        .extend(airflow_role.get_commands(authentication_config, resolved_product_image));
+    airflow_container_args.extend(airflow_role.get_commands(
+        airflow,
+        authentication_config,
+        resolved_product_image,
+    ));
 
     airflow_container
         .image_from_product_image(resolved_product_image)
diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs
@@ -251,6 +251,10 @@ pub mod versioned {
         #[serde(default)]
         pub load_examples: bool,
 
+        /// Whether to execute the database initialization routines (a combination of database initialization, upgrade and migration depending on the Airflow version). Defaults to true to be backwward compatible.
+        #[serde(default = "default_db_init")]
+        pub db_init: bool,
+
         /// 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)
@@ -268,7 +272,6 @@ pub mod versioned {
         #[schemars(schema_with = "raw_object_list_schema")]
         pub volume_mounts: Vec<VolumeMount>,
     }
-
     // TODO: move generic version to op-rs?
     #[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
     #[serde(rename_all = "camelCase")]
@@ -282,6 +285,10 @@ pub mod versioned {
     }
 }
 
+fn default_db_init() -> bool {
+    true
+}
+
 impl Default for v1alpha1::WebserverRoleConfig {
     fn default() -> Self {
         v1alpha1::WebserverRoleConfig {
@@ -547,6 +554,7 @@ impl AirflowRole {
     /// if authentication is enabled.
     pub fn get_commands(
         &self,
+        airflow: &v1alpha1::AirflowCluster,
         auth_config: &AirflowClientAuthenticationDetailsResolved,
         resolved_product_image: &ResolvedProductImage,
     ) -> Vec<String> {
@@ -576,21 +584,30 @@ impl AirflowRole {
                         "airflow api-server &".to_string(),
                     ]);
                 }
-                AirflowRole::Scheduler => command.extend(vec![
-                    "airflow db migrate".to_string(),
-                    "airflow users create \
-                        --username \"$ADMIN_USERNAME\" \
-                        --firstname \"$ADMIN_FIRSTNAME\" \
-                        --lastname \"$ADMIN_LASTNAME\" \
-                        --email \"$ADMIN_EMAIL\" \
-                        --password \"$ADMIN_PASSWORD\" \
-                        --role \"Admin\""
-                        .to_string(),
-                    "prepare_signal_handlers".to_string(),
-                    container_debug_command(),
-                    "airflow dag-processor &".to_string(),
-                    "airflow scheduler &".to_string(),
-                ]),
+                AirflowRole::Scheduler => {
+                    if airflow.spec.cluster_config.db_init {
+                        tracing::info!("Database initialization...");
+                        command.extend(vec![
+                            "airflow db migrate".to_string(),
+                            "airflow users create \
+                            --username \"$ADMIN_USERNAME\" \
+                            --firstname \"$ADMIN_FIRSTNAME\" \
+                            --lastname \"$ADMIN_LASTNAME\" \
+                            --email \"$ADMIN_EMAIL\" \
+                            --password \"$ADMIN_PASSWORD\" \
+                            --role \"Admin\""
+                                .to_string(),
+                        ]);
+                    } else {
+                        tracing::info!("Database initialization routines have been skipped!")
+                    }
+                    command.extend(vec![
+                        "prepare_signal_handlers".to_string(),
+                        container_debug_command(),
+                        "airflow dag-processor &".to_string(),
+                        "airflow scheduler &".to_string(),
+                    ]);
+                }
                 AirflowRole::Worker => command.extend(vec![
                     "prepare_signal_handlers".to_string(),
                     container_debug_command(),
@@ -608,22 +625,31 @@ impl AirflowRole {
                         "airflow webserver &".to_string(),
                     ]);
                 }
-                AirflowRole::Scheduler => command.extend(vec![
-                    // Database initialization is limited to the scheduler, see https://github.com/stackabletech/airflow-operator/issues/259
-                    "airflow db init".to_string(),
-                    "airflow db upgrade".to_string(),
-                    "airflow users create \
-                        --username \"$ADMIN_USERNAME\" \
-                        --firstname \"$ADMIN_FIRSTNAME\" \
-                        --lastname \"$ADMIN_LASTNAME\" \
-                        --email \"$ADMIN_EMAIL\" \
-                        --password \"$ADMIN_PASSWORD\" \
-                        --role \"Admin\""
-                        .to_string(),
-                    "prepare_signal_handlers".to_string(),
-                    container_debug_command(),
-                    "airflow scheduler &".to_string(),
-                ]),
+                AirflowRole::Scheduler => {
+                    if airflow.spec.cluster_config.db_init {
+                        tracing::info!("Database initialization...");
+                        command.extend(vec![
+                            // Database initialization is limited to the scheduler, see https://github.com/stackabletech/airflow-operator/issues/259
+                            "airflow db init".to_string(),
+                            "airflow db upgrade".to_string(),
+                            "airflow users create \
+                            --username \"$ADMIN_USERNAME\" \
+                            --firstname \"$ADMIN_FIRSTNAME\" \
+                            --lastname \"$ADMIN_LASTNAME\" \
+                            --email \"$ADMIN_EMAIL\" \
+                            --password \"$ADMIN_PASSWORD\" \
+                            --role \"Admin\""
+                                .to_string(),
+                        ]);
+                    } else {
+                        tracing::info!("Database initialization routines have been skipped!")
+                    }
+                    command.extend(vec![
+                        "prepare_signal_handlers".to_string(),
+                        container_debug_command(),
+                        "airflow scheduler &".to_string(),
+                    ]);
+                }
                 AirflowRole::Worker => command.extend(vec![
                     "prepare_signal_handlers".to_string(),
                     container_debug_command(),
@@ -981,5 +1007,7 @@ mod tests {
         assert_eq!("KubernetesExecutor", cluster.spec.executor.to_string());
         assert!(cluster.spec.cluster_config.load_examples);
         assert!(cluster.spec.cluster_config.expose_config);
+        // defaults to true
+        assert!(cluster.spec.cluster_config.db_init);
     }
 }
diff --git a/tests/templates/kuttl/cluster-operation/09-assert.yaml b/tests/templates/kuttl/cluster-operation/09-assert.yaml
@@ -0,0 +1,8 @@
+---
+# For this assert we expect the database operation to be logged
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 30
+commands:
+- script: |
+    kubectl -n $NAMESPACE logs airflow-scheduler-default-0 | grep "Database migrating done!"
diff --git a/tests/templates/kuttl/cluster-operation/30-restart-airflow.yaml.j2 b/tests/templates/kuttl/cluster-operation/30-restart-airflow.yaml.j2
@@ -25,6 +25,7 @@ spec:
     vectorAggregatorConfigMapName: vector-aggregator-discovery
 {% endif %}
     credentialsSecret: test-airflow-credentials
+    dbInit: false
   webservers:
     roleConfig:
       listenerClass: external-unstable
diff --git a/tests/templates/kuttl/cluster-operation/31-assert.yaml b/tests/templates/kuttl/cluster-operation/31-assert.yaml
@@ -0,0 +1,8 @@
+---
+# For this step we expect the database operation to NOT be logged
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 30
+commands:
+- script: |
+    kubectl -n $NAMESPACE logs airflow-scheduler-default-0 | grep -q "Database migrating done!" && exit 1 || exit 0
