Suspend Airflow API endpoint

[jaketf]

Description

add a global /suspend and /resume endpoint to the REST API (and corresponding CLI command) that would do the following for the entire airflow deployment:

/suspend

• pause all currently unpaused DAGs (keeping track of what was previously unpaused in the metadata db)
• Set AIRFLOW__SCHEDULER__USE_JOB_SCHEDULE=False (if not already false) to prevent scheduler from scheduling more tasks
• (should handle duplicate suspend calls by 409: Airflow is already suspended)

/resume

• unpause all DAGs from a previous /suspend call
• set AIRFLOW__SCHEDULER__USE_JOB_SCHEDULE to whatever it was before the last /suspend call
• (should handle resuming an airflow that is not in the suspended state with 409: Cannot resume an airflow that is not in suspended state)

Use case/motivation

This could be really helpful for people wanting to pause everything:

• before running and upgrade to a newer airflow version
• because this is a dev enviornment and we don't need anything to run outside of working hours to reduce resource consumption
• "archiving an airflow" call suspend then tear down all the airflow components, could be restored by bringing up the components and then calling resume.

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

[potiuk]

literally setting AIRFLOW__SCHEDULER__USE_JOB_SCHEDULE via API is not possible, because we would have to a) change it in env variables for running process (because it is impossible) b) do it in scheduler - and the API/webserver does not have direct way to communicate with scheduler, it would have to be done through database rather than through env variables.

I am not sure if this is a good idea to influence scheduler's behaviour via API. I'd say any cahnge of scheduler behaviour should be done via deployment. What we COULD do instead we could have Airflow scheduler to respond to (for example) USR1 signal and have a way to pass it via the deployment).

Converting it into discussion - maybe someone will have other ideas.

[brki]

It seems like a very good feature to have.

Perhaps instead of controlling it by setting AIRFLOW__SCHEDULER__USE_JOB_SCHEDULE, a boolean variable, named something like airflow_job_schedule_suspended, could be saved to the DB.

Then, in the scheduling code, instead of just checking settings.USE_JOB_SCHEDULE as it currently does, it would call a function like this:

def job_schedule_enabled():
    return settings.USE_JOB_SCHEDULE and not Variable.get('airflow_job_schedule_suspended', False)

Edit: I guess my suggestion doesn't address pausing / unpausing the currently running scheduled DAGS. I'd like to have that too, though. My wish is to be able to pause everything cleanly, do an upgrade of Airflow, and then unpause only those things that I paused before. The context in which I want this: A helm-chart Airflow installation, where I have ~100 DAGs that run often, some of them long-running, and a few DAGs that have already been paused for whatever reason, and I want to upgrade Airflow with a minimum of tasks / DAGs failing because they were interrupted.

[potiuk]

Edit: I guess my suggestion doesn't address pausing / unpausing the currently running scheduled DAGS. I'd like to have that too, though. My wish is to be able to pause everything cleanly, do an upgrade of Airflow, and then unpause only those things that I paused before. The context in which I want this: A helm-chart Airflow installation, where I have ~100 DAGs that run often, some of them long-running, and a few DAGs that have already been paused for whatever reason, and I want to upgrade Airflow with a minimum of tasks / DAGs failing because they were interrupted.

PRss to implement it are always welcome. That is an interesting feature in this context.