Indicate if the database is been updated

[onzyone]

Description

I would like add a new flag to airflow db that will allow to check if the db is been updated.

Use case/motivation

When using Airflow in HA mode with a replica of 2 or more schedulers hosted within Kubernetes we are running into an issue where both of the scheduler pods try to upgrade the db.

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[potiuk]

Next time please double check what is available before opening issue.

Usage: airflow db check-migrations [-h] [-t MIGRATION_WAIT_TIMEOUT] [-v]

Check if migration have finished (or continually check until timeout)

Optional Arguments:
  -h, --help            show this help message and exit
  -t, --migration-wait-timeout MIGRATION_WAIT_TIMEOUT
                        timeout to wait for db to migrate
  -v, --verbose         Make logging output more verbose

You can look at our Helm Chart https://github.com/apache/airflow/tree/main/chart to see how this can be implemented (you ca use helm --dry-run. Or even better, use the helm-chart rather than implement things on your own.

[potiuk]

Surely you should have a separate job doing the migration and all other schedulers should wait for migration to complete. Look at our helm chart as a reference.

[potiuk]

And just to add to it.

Adding a separate migration job and using wait-for-migrations is one strategy you can use. But this is not strictly necessary.
I am not sure which version of Database or Airlfow you use, but all the migrations use a global migration lock.

    with create_global_lock(session=session, lock=DBLocks.MIGRATIONS):
        import sqlalchemy.pool

        log.info("Creating tables")
        val = os.environ.get("AIRFLOW__DATABASE__SQL_ALCHEMY_MAX_SIZE")
        try:
            # Reconfigure the ORM to use _EXACTLY_ one connection, otherwise some db engines hang forever
            # trying to ALTER TABLEs
            os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_MAX_SIZE"] = "1"
            settings.reconfigure_orm(pool_class=sqlalchemy.pool.SingletonThreadPool)
            command.upgrade(config, revision=to_revision or "heads")
        finally:
            if val is None:
                os.environ.pop("AIRFLOW__DATABASE__SQL_ALCHEMY_MAX_SIZE")
            else:
                os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_MAX_SIZE"] = val
            settings.reconfigure_orm()

So even if they start to do upgrade, only one of them will actually do the migration. The other one will wait until the lock is released.

Is it not working for you @onzyone ? Why ? What are the problems you experience in your version of the DB that prevents the lock from working?

[onzyone]

We are running Airflow 2.6.0, with a PostgreSQL backend, on OCP4. One of the main issues that we are facing is our pipeline is using ArgoCD. With the way our Argo is implemented, we are seeing issues with kind:job. This is the main reason I wanted it just to run as part of the schedulers deployment.

Thank you for providing this code snippet, I have a better idea how the migration job works now. I was going to build a "check for migration" script so that the secondary scheduler would just wait. Give me some time to explore a different deployment approach (likely a migration deployment with a sleep forever, yes yes not ideal)

[onzyone]

Okay so we got this working with a combination of the migration script and argocd wave annotation. ArgoCd Waves

[potiuk]

Have you seen these docs @onzyone https://airflow.apache.org/docs/helm-chart/stable/index.html#installing-the-chart-with-argo-cd-flux-rancher-or-terraform - I have no idea about Argo, but we do have a section about Argo installation in our docs. Maybe you could provide a PR to improve that sections if what we have is not enough ?

It's super-simple. Just click "Suggest a change on this page" button and you will get PR open where you will be able to update the documentation. Nice way to contribute back your findings.