Scheduler restarting after many dag insert/serialization failed in dynamic dag creation

[rdjouder]

Hello,
Almost same use case as "dynamic dag creation race condition".
Adding details, refreshing/recycling discussion thread.
Using Airflow 2.5.2, postgresql 13, python 3.8, (max_tis_per_query = 256) running two schedulers instances, dags are created dynamically from dedicated psql table. Number of DAGs may reach 20k. When creating too many (300), dags at a time, Scheduler is experiencing race conditions during DAG serialisation and restarts:

[SQL: INSERT INTO dag (dag_id, root_dag_id, is_paused, is_subdag, is_active, last_parsed_time, last_pickled, last_expired, scheduler_lock, pickle_id, fileloc, processor_subdir, owners, description, default_view, schedule_interval, timetable_description, max_active_tasks, max_active_runs, has_task_concurrency_limits, has_import_errors, next_dagrun, next_dagrun_data_interval_start, next_dagrun_data_interval_end, next_dagrun_create_after) VALUES (%(dag_id)s, %(root_dag_id)s, %(is_paused)s, %(is_subdag)s, %(is_active)s, %(last_parsed_time)s, %(last_pickled)s, %(last_expired)s, %(scheduler_lock)s, %(pickle_id)s, %(fileloc)s, %(processor_subdir)s, %(owners)s, %(description)s, %(default_view)s, %(schedule_interval)s, %(timetable_description)s, %(max_active_tasks)s, %(max_active_runs)s, %(has_task_concurrency_limits)s, %(has_import_errors)s, %(next_dagrun)s, %(next_dagrun_data_interval_start)s, %(next_dagrun_data_interval_end)s, %(next_dagrun_create_after)s)]
[parameters: ({'dag_id': ...]
(Background on this error at: https://sqlalche.me/e/14/gkpj)
ERROR [airflow.models.dagbag.DagBag] Failed to write serialized DAG: /usr/local/airflow/dags/dynamic_dag_creator_chunk2.py
Traceback (most recent call last):
File "/home/airflow/.local/bin/airflow", line 8, in
sys.exit(main())
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/main.py", line 48, in main
args.func(args)
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/cli/cli_parser.py", line 52, in command
return func(*args, **kwargs)
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/cli/commands/db_command.py", line 37, in initdb
db.initdb()
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/session.py", line 75, in wrapper
return func(*args, session=session, **kwargs)
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/db.py", line 697, in initdb
upgradedb(session=session)
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/session.py", line 72, in wrapper
return func(*args, **kwargs)
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/db.py", line 1588, in upgradedb
_reserialize_dags(session=session)
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/db.py", line 840, in _reserialize_dags
dagbag.sync_to_db(session=session)
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/session.py", line 72, in wrapper
return func(*args, **kwargs)
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/dagbag.py", line 639, in sync_to_db

...
sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "dag_pkey"
DETAIL: Key (dag_id)=(PredictiveThreshold_os.cpu.load_XXX_0) already exists.

Seeing 4/5 occurrences of this kind of error then Scheduler restarts.

Is there a way to tune my scheduler config to prevent such issues ?
This issue does not occur when DAGs are submitted at a lower pace, one DAG every 0.1s.
DAGs parameters are submitted through REST service, which back-pressure framework would you recommend ?

[potiuk]

This looks like your "something" clashes with scheduler while running "initdb". It looks like you are doing something that you should absolutely not do - which is running airflow db init in parallel to regular operations of scheduler which you should absolutely not do. This means that you are trying to reinitialize the database while something else is running.

Why are you doing this?

[rdjouder]

Airflow is up and running, airflow, 20 mlworkers, 2 schedulers, postgresql , pgbouncer.
Submitting up to 500 DAGs at a time, loaded from DB, dedicated table.
This issue does not occur when DAGs are submitted at a lower pace, one DAG every 0.1s.
DAGs parameters are submitted through REST service, which back-pressure framework would you recommend ?

[potiuk]

No idea. you have stream of thoughts that probably only you understand. I have no idea what "Dag submit is", what it means loaded from the DB, what dedicated table is.

Airflow does not allow for "submitting dags" - dags are supposed to be ONLY parsed by scheduler by scanning DAGS folder and let it parse it at it's own speed. Full stop. Anything else you try to do is wrong and not supported.

[rdjouder]

Hello,
I am using Airflow feature dynamic-dag-generation. Airflow Scheduler calls a python script that creates airflow.DAG instances, lets call this script dynamic_dag_creator.
dynamic_dag_creator uses "with DAG(..)".
The number of created airflow.DAG instances increases over time. When dynamic_dag_creator creates more that 300 new DAGs, airflow scheduler logs shows error quoted above and it restarts after 4 or 5 occurrences of this error.
This error is dependant of the rate at which dynamic_dag_creator creates new DAGs instances.

My Airflow deployment uses two Airflow schedulers.

[schanjr]

I have same issue when running this on MWAA airflow. I created my dags via decorators. Were you able to have this resolved? @rdjouder