[AIRFLOW-5088][AIP-24] Persisting serialized DAG in DB for webserver scalability

airflow/api/common/experimental/delete_dag.py

@@ -22,8 +22,10 @@
 
 from airflow import models
 from airflow.exceptions import DagNotFound
-from airflow.models import DagModel, TaskFail
+from airflow.models import DagModel, SerializedDagModel, TaskFail
+from airflow.settings import STORE_SERIALIZED_DAGS
 from airflow.utils.db import provide_session
+from airflow.utils.log.logging_mixin import LoggingMixin
 
 
 @provide_session
@@ -36,10 +38,17 @@ def delete_dag(dag_id: str, keep_records_in_log: bool = True, session=None) -> i
     :param session: session used
     :return count of deleted dags
     """
+    logger = LoggingMixin()
+    logger.log.info("Deleting DAG: %s", dag_id)
     dag = session.query(DagModel).filter(DagModel.dag_id == dag_id).first()
     if dag is None:
         raise DagNotFound("Dag id {} not found".format(dag_id))
 
+    # Scheduler removes DAGs without files from serialized_dag table every dag_dir_list_interval.
+    # There may be a lag, so explicitly removes serialized DAG here.
+    if STORE_SERIALIZED_DAGS and SerializedDagModel.has_dag(dag_id=dag_id, session=session):
+        SerializedDagModel.remove_dag(dag_id=dag_id, session=session)
+
     count = 0
 
     # noinspection PyUnresolvedReferences,PyProtectedMember

airflow/config_templates/default_airflow.cfg

@@ -208,6 +208,13 @@ dag_discovery_safe_mode = True
 # The number of retries each task is going to have by default. Can be overridden at dag or task level.
 default_task_retries = 0
 
+# Whether to serialises DAGs and persist them in DB.
+# If set to True, Webserver reads from DB instead of parsing DAG files
+# More details: https://airflow.apache.org/howto/enable-dag-serialization.html
+store_serialized_dags = False
+
+# Updating serialized DAG can not be faster than a minimum interval to reduce database write rate.
+min_serialized_dag_update_interval = 30
 
 [cli]
 # In what way should the cli access the API. The LocalClient will use the