feat: Add flag for db init routine

CHANGELOG.md

@@ -2,13 +2,18 @@
 
 ## [Unreleased]
 
+### Added
+
+- Add a flag to determine if database initialization steps should be executed ([#669]).
+
 ### Fixed
 
 - Don't panic on invalid authorization config. Previously, a missing OPA ConfigMap would crash the operator ([#667]).
 - Fix OPA authorization for Airflow 3. Airflow 3 needs to be configured via env variables, the operator now does this correctly ([#668]).
 
 [#667]: https://github.com/stackabletech/airflow-operator/pull/667
 [#668]: https://github.com/stackabletech/airflow-operator/pull/668
+[#669]: https://github.com/stackabletech/airflow-operator/pull/669
 
 ## [25.7.0] - 2025-07-23
 

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.

docs/modules/airflow/pages/usage-guide/db-init.adoc

@@ -0,0 +1,22 @@
+= Database initialization
+:description: Configure Airflow DB start-up.
+
+By default, Airflow will run database initialization routines (checking and/or creating the metadata schema and creating an admin user) on start-up.
+These are idempotent and can be run every time as the overhead is minimal.
+However, if these steps should be skipped, a running Airflow cluster can be patched with a resource like this to deactivate the initialization:
+
+[source,yaml]
+----
+---
+apiVersion: airflow.stackable.tech/v1alpha1
+kind: AirflowCluster
+metadata:
+  name: airflow
+spec:
+  clusterConfig:
+    dbInit: false  # <1>
+----
+<1> Turn off the initialization routine by setting `dbInit` to `false`
+
+NOTE: The field `dbInit` is `true` by default to be backwards-compatible.
+A fresh Airflow cluster cannot be created with this field set to `false` as this results in missing metadata in the Airflow database.

docs/modules/airflow/partials/nav.adoc

@@ -3,6 +3,7 @@
 ** xref:airflow:getting_started/first_steps.adoc[]
 * xref:airflow:required-external-components.adoc[]
 * xref:airflow:usage-guide/index.adoc[]
+** xref:airflow:usage-guide/db-init.adoc[]
 ** xref:airflow:usage-guide/mounting-dags.adoc[]
 ** xref:airflow:usage-guide/applying-custom-resources.adoc[]
 ** xref:airflow:usage-guide/listenerclass.adoc[]

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)

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);
     }
 }

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!"

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

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