Allow Airflow DB upgrades/migrations to be on-demand

[adwk67]

Please talk to @adwk67 before starting work on this

Related to SUP-174 and SUP-199

At the moment, upgrades to the airflow database occur every time the scheduler starts up: see here and here. This can cause significant overhead, depending on the number and complexity of DAGs (as well as delays in the scheduler being available). This ticket will cover the following:

• replacing airflow db upgrade with airflow db migrate (as the former has been deprecated)
• making the migration on demand via e.g. a new field in the resource
• We need to make sure we don't introduce problems similar to the AirflowDB resource in the past (e.g. don't run a migration after a Airflow version bump), see Removed AirflowDB #322

(from below)

Proposal

• add a flag to bypass db migration and user creation, defaulting to true in both cases (so it's not breaking)
• extend the airflow demo using a DAG factory and setting AIRFLOW__CORE__MIN_SERIALIZED_DAG_UPDATE_INTERVAL to a different default setting (this is better than an integration test as it is somewhat of a corner-case, is defficult to verify in a test but simple to docuemtn as part of the demo)

[lfrancke]

Notes from planning meeting

• Suggestion is to create a separate CRD
  • Could also be useful for OpenSearch which has similar requirements
• Alternative suggestion: Can we just add a boolean flag to skip the migration?
  • OpenSearch would need ~10 boolean flags
• Research needed why the DB job is even slow in the first place

-> Refinement needed

[adwk67]

Findings

The following comments refer to the 3.0.1 code tag.

Current situation with 25.7 code

• the operator calls airflow db migrate, airflow users create and airflow dag-processor on the scheduler role
• airflow db migrate
  • sets up the ORM by creating a DB pool on the scheduler and each celery worker
  • the command is designed to be idempotent: the current and target schema versions are compared and if no migration is required, it exits immediately, as seen below:

[2025-07-09T15:10:58.482+0000] {providers_manager.py:359} DEBUG - Initialization of Providers Manager[config] took 0.02 seconds 
DB: ****
Performing upgrade to the metadata database ****
[2025-07-09T15:10:58.644+0000] {migration.py:207} INFO - Context impl PostgresqlImpl.
[2025-07-09T15:10:58.644+0000] {migration.py:210} INFO - Will assume transactional DDL.
[2025-07-09T15:10:58.653+0000] {db.py:1135} INFO - Migrating the Airflow database 
[2025-07-09T15:10:58.654+0000] {settings.py:494} DEBUG - Disposing DB connection pool (PID 15) 
[2025-07-09T15:10:58.655+0000] {settings.py:345} DEBUG - Setting up DB connection pool (PID 15)
...
[2025-07-09T15:10:58.756+0000] {settings.py:454} DEBUG - settings.prepare_engine_args(): Using pool settings. pool_size=5, max_overflow=10, pool_recycle=1800, pid=15
Database migrating done!  

• airflow users create
  • a check is made if the fab user already exists before adding it
• airflow dag-processor
  • the processor is started

Benchmark tests with multiple DAGs

This ConfigMap creates 2 DAGs, a simple one and a factory that creates 1200 DAGs:

---
apiVersion: v1
kind: ConfigMap
metadata:
  name: test-cm-dag
data:
  example_trigger_target_dag.py: |
    import pendulum
    from airflow import DAG
    from airflow.decorators import task
    from airflow.operators.bash import BashOperator

    @task(task_id="run_this")
    def run_this_func(dag_run=None):
        """
        Print the payload "message" passed to the DagRun conf attribute.

        :param dag_run: The DagRun object
        :type dag_run: DagRun
        """
        print(f"Remotely received value of {dag_run.conf.get('message')} for key=message")

    with DAG(
        dag_id="example_trigger_target_dag",
        start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
        catchup=False,
        schedule=None,
        tags=['example'],
    ) as dag:
        run_this = run_this_func()

        bash_task = BashOperator(
            task_id="bash_task",
            bash_command='echo "Here is the message: $message"',
            env={'message': '{% raw %}{{ dag_run.conf.get("message") }}{% endraw %}'},
        )
  dag_factory.py: |
    from airflow import DAG
    from airflow.operators.empty import EmptyOperator
    from datetime import datetime, timedelta

    # Number of DAGs to generate
    NUM_DAGS = 1200  # Increase for more stress
    DAG_PREFIX = "stress_dag_"

    default_args = {
        'owner': 'airflow',
        'start_date': datetime(2024, 1, 1),
        'retries': 1,
        'retry_delay': timedelta(seconds=5),
    }

    def create_dag(dag_id):
        with DAG(
            dag_id=dag_id,
            default_args=default_args,
            schedule=None,
            catchup=False,
            tags=["stress_test"],
        ) as dag:
            start = EmptyOperator(task_id='start')
            end = EmptyOperator(task_id='end')
            start >> end
        return dag

    for i in range(NUM_DAGS):
        dag_id = f"{DAG_PREFIX}{i:04d}"
        globals()[dag_id] = create_dag(dag_id)

Running an airflow cluster with these DAGs yields the following log entries:

# calls to bulk sync the DAGs to the DB: these correspond to the Sync commands below
[2025-07-09T13:33:25.242+0000] {collection.py:349} DEBUG - Calling the DAG.bulk_sync_to_db method
[2025-07-09T13:33:46.160+0000] {collection.py:349} DEBUG - Calling the DAG.bulk_sync_to_db method
[2025-07-09T13:33:57.916+0000] {collection.py:349} DEBUG - Calling the DAG.bulk_sync_to_db method
[2025-07-09T13:34:20.047+0000] {collection.py:349} DEBUG - Calling the DAG.bulk_sync_to_db method
[2025-07-09T13:34:28.796+0000] {collection.py:349} DEBUG - Calling the DAG.bulk_sync_to_db method
[2025-07-09T13:34:54.054+0000] {collection.py:349} DEBUG - Calling the DAG.bulk_sync_to_db method
[2025-07-09T13:35:01.382+0000] {collection.py:349} DEBUG - Calling the DAG.bulk_sync_to_db method

# Sync calls for the DAG-factory-generated DAGs
[2025-07-09T13:33:46.160+0000] {dag.py:1895} INFO - Sync 1200 DAGs
[2025-07-09T13:34:20.047+0000] {dag.py:1895} INFO - Sync 1200 DAGs
[2025-07-09T13:34:54.054+0000] {dag.py:1895} INFO - Sync 1200 DAGs

# Sync calls for the second, individual DAG in the ConfigMap
[2025-07-09T13:33:25.242+0000] {dag.py:1895} INFO - Sync 1 DAGs
[2025-07-09T13:33:57.916+0000] {dag.py:1895} INFO - Sync 1 DAGs
[2025-07-09T13:34:28.796+0000] {dag.py:1895} INFO - Sync 1 DAGs
[2025-07-09T13:35:01.382+0000] {dag.py:1895} INFO - Sync 1 DAGs

# Inspecting a specific generated DAG shows that it is sync-ed about every 30s
[2025-07-09T13:33:53.017+0000] {serialized_dag.py:414} DEBUG - Checking if DAG (stress_dag_0860) changed
[2025-07-09T13:34:22.266+0000] {serialized_dag.py:414} DEBUG - Checking if DAG (stress_dag_0860) changed
[2025-07-09T13:35:00.134+0000] {serialized_dag.py:414} DEBUG - Checking if DAG (stress_dag_0860) changed

The default interval for syncing DAGs is given by AIRFLOW__CORE__MIN_SERIALIZED_DAG_UPDATE_INTERVAL, which defaults to 30s. This interval may have to be increased if the number of DAGs means that the synchronization is not complete before the next cycle begins.

Proposal

• add a flag to bypass db migration and user creation, defaulting to true in both cases (so it's not breaking)
• aextend the airflow demo using a DAG factory and setting AIRFLOW__CORE__MIN_SERIALIZED_DAG_UPDATE_INTERVAL to a different default setting (this is better than an integration test as it is somewhat of a corner-case, is defficult to verify in a test but simple to docuemtn as part of the demo)

Links

• https://airflow.apache.org/docs/apache-airflow/3.0.1/migrations-ref.html
• https://airflow.apache.org/docs/apache-airflow/3.0.1/administration-and-deployment/dag-serialization.html
• https://airflow.apache.org/docs/apache-airflow/3.0.1/configurations-ref.html#min-serialized-dag-update-interval

[lfrancke]

The proposal looks good to me, thank you.