added a new condition before launching the self._scan_stale_dags() fu…

airflow/config_templates/config.yml

@@ -530,6 +530,14 @@ core:
       type: integer
       example: ~
       default: "4096"
+    purge_stale_dags:
+      description: |
+        Stale dags are deleted by default, set this to False
+        if you want to keep them.
+      version_added: 2.10.0
+      type: boolean
+      example: ~
+      default: "True"
 database:
   description: ~
   options:

airflow/dag_processing/manager.py

@@ -476,8 +476,8 @@ def start(self):
 
         return self._run_parsing_loop()
 
-    def _scan_stale_dags(self):
-        """Scan at fix internal DAGs which are no longer present in files."""
+    def _purge_stale_dags(self):
+        """Scan at purge internal DAGs which are no longer present in files."""
         now = timezone.utcnow()
         elapsed_time_since_refresh = (now - self.last_deactivate_stale_dags_time).total_seconds()
         if elapsed_time_since_refresh > self.parsing_cleanup_interval:
@@ -600,7 +600,9 @@ def _run_parsing_loop(self):
 
             if self.standalone_dag_processor:
                 self._fetch_callbacks(max_callbacks_per_loop)
-            self._scan_stale_dags()
+
+            if conf.getboolean("core", "purge_stale_dags", fallback=True):
+                self._purge_stale_dags()
             DagWarning.purge_inactive_dag_warnings()
             refreshed_dag_dir = self._refresh_dag_dir()
 

docs/apache-airflow/core-concepts/dags.rst

@@ -924,3 +924,21 @@ if it fails for ``N`` number of times consecutively.
 we can also provide and override these configuration from DAG argument:
 
 - ``max_consecutive_failed_dag_runs``: Overrides :ref:`config:core__max_consecutive_failed_dag_runs_per_dag`.
+
+Disable deletion of stale DAGs
+------------------------------
+
+A small clarification about obsolete DAGs: For the DAGs to become obsolete,
+they must be deleted from the DAG file. And it is at this point that the purge_stale_dags function will carry out the purge of these DAGs in the database.
+
+A use case where we want to keep obsolete DAGs:
+
+In a versioned DAG context, it may be necessary to run two versions of DAG in parallel in two versions of workers.
+
+It can also be useful to keep obsolete DAGs if they are still being used on the version (n-1) of the worker.
+
+However, as the scheduler will not have the DAGs of the version (n-1), it will automatically launch the purge of non-existing DAGs in the DAG file.
+
+To keep obsolete DAGs, we can set the value of the ``AIRFLOW__CORE__PURGE_STALE_DAGS`` variable to ``False``.
+
+By default, it is set to ``True`` in ``config.yml``.

tests/dag_processing/test_job_runner.py

@@ -616,7 +616,7 @@ def test_file_paths_in_queue_sorted_by_priority(
         )
         assert session_delete.call_args[0][0].fileloc == parsing_request.fileloc
 
-    def test_scan_stale_dags(self):
+    def test_purge_stale_dags(self):
         """
         Ensure that DAGs are marked inactive when the file is parsed but the
         DagModel.last_parsed_time is not updated.
@@ -669,7 +669,7 @@ def test_scan_stale_dags(self):
             )
             assert serialized_dag_count == 1
 
-            manager.processor._scan_stale_dags()
+            manager.processor._purge_stale_dags()
 
             active_dag_count = (
                 session.query(func.count(DagModel.dag_id))
@@ -692,7 +692,7 @@ def test_scan_stale_dags(self):
             ("scheduler", "stale_dag_threshold"): "50",
         }
     )
-    def test_scan_stale_dags_standalone_mode(self):
+    def test_purge_stale_dags_standalone_mode(self):
         """
         Ensure only dags from current dag_directory are updated
         """
@@ -740,7 +740,7 @@ def test_scan_stale_dags_standalone_mode(self):
             active_dag_count = session.query(func.count(DagModel.dag_id)).filter(DagModel.is_active).scalar()
             assert active_dag_count == 2
 
-            manager.processor._scan_stale_dags()
+            manager.processor._purge_stale_dags()
 
             active_dag_count = session.query(func.count(DagModel.dag_id)).filter(DagModel.is_active).scalar()
             assert active_dag_count == 1
@@ -1410,6 +1410,147 @@ def test_callback_queue(self, tmp_path):
             dag1_req2,
         ]
 
+    @conf_vars({("core", "purge_stale_dags"): "False"})
+    @conf_vars({("core", "load_examples"): "False"})
+    def test_disable_purge_stale_dags(self):
+        """
+        Test disable purge_stale_dags.
+        """
+        manager = DagProcessorJobRunner(
+            job=Job(),
+            processor=DagFileProcessorManager(
+                dag_directory=TEST_DAG_FOLDER,
+                max_runs=1,
+                processor_timeout=timedelta(minutes=10),
+                dag_ids=[],
+                pickle_dags=False,
+                async_mode=True,
+            ),
+        )
+
+        test_dag_path = str(TEST_DAG_FOLDER / "test_example_bash_operator.py")
+        dagbag = DagBag(test_dag_path, read_dags_from_db=False, include_examples=False)
+
+        assert not conf.getboolean("core", "purge_stale_dags")
+
+        with create_session() as session:
+            # Add stale DAG to the DB
+            dag = dagbag.get_dag("test_example_bash_operator")
+            dag.last_parsed_time = timezone.utcnow()
+            dag.sync_to_db()
+            SerializedDagModel.write_dag(dag)
+
+            # Add DAG to the file_parsing_stats
+            stat = DagFileStat(
+                num_dags=1,
+                import_errors=0,
+                last_finish_time=timezone.utcnow() + timedelta(hours=1),
+                last_duration=timedelta(seconds=1),
+                run_count=1,
+            )
+            manager.processor._file_paths = [test_dag_path]
+            manager.processor._file_stats[test_dag_path] = stat
+
+            active_dag_count = (
+                session.query(func.count(DagModel.dag_id))
+                .filter(DagModel.is_active, DagModel.fileloc == test_dag_path)
+                .scalar()
+            )
+            assert active_dag_count == 1
+
+            serialized_dag_count = (
+                session.query(func.count(SerializedDagModel.dag_id))
+                .filter(SerializedDagModel.fileloc == test_dag_path)
+                .scalar()
+            )
+            assert serialized_dag_count == 1
+
+            manager.processor._run_parsing_loop()
+
+            active_dag_count = (
+                session.query(func.count(DagModel.dag_id))
+                .filter(DagModel.is_active, DagModel.fileloc == test_dag_path)
+                .scalar()
+            )
+            assert active_dag_count == 1
+
+            serialized_dag_count = (
+                session.query(func.count(SerializedDagModel.dag_id))
+                .filter(SerializedDagModel.fileloc == test_dag_path)
+                .scalar()
+            )
+            assert serialized_dag_count == 1
+
+    @conf_vars({("core", "load_examples"): "False"})
+    def test_enable_purge_stale_dags(self):
+        """
+        Test enable purge_stale_dags with the default variable.
+        """
+        manager = DagProcessorJobRunner(
+            job=Job(),
+            processor=DagFileProcessorManager(
+                dag_directory=TEST_DAG_FOLDER,
+                max_runs=1,
+                processor_timeout=timedelta(minutes=10),
+                dag_ids=[],
+                pickle_dags=False,
+                async_mode=True,
+            ),
+        )
+
+        test_dag_path = str(TEST_DAG_FOLDER / "test_example_bash_operator.py")
+        dagbag = DagBag(test_dag_path, read_dags_from_db=False, include_examples=False)
+
+        assert conf.getboolean("core", "purge_stale_dags")
+
+        with create_session() as session:
+            # Add stale DAG to the DB
+            dag = dagbag.get_dag("test_example_bash_operator")
+            dag.last_parsed_time = timezone.utcnow()
+            dag.sync_to_db()
+            SerializedDagModel.write_dag(dag)
+
+            # Add DAG to the file_parsing_stats
+            stat = DagFileStat(
+                num_dags=1,
+                import_errors=0,
+                last_finish_time=timezone.utcnow() + timedelta(hours=1),
+                last_duration=timedelta(seconds=1),
+                run_count=1,
+            )
+            manager.processor._file_paths = [test_dag_path]
+            manager.processor._file_stats[test_dag_path] = stat
+
+            active_dag_count = (
+                session.query(func.count(DagModel.dag_id))
+                .filter(DagModel.is_active, DagModel.fileloc == test_dag_path)
+                .scalar()
+            )
+            assert active_dag_count == 1
+
+            serialized_dag_count = (
+                session.query(func.count(SerializedDagModel.dag_id))
+                .filter(SerializedDagModel.fileloc == test_dag_path)
+                .scalar()
+            )
+            assert serialized_dag_count == 1
+
+            manager.processor._run_parsing_loop()
+
+            active_dag_count = (
+                session.query(func.count(DagModel.dag_id))
+                .filter(DagModel.is_active, DagModel.fileloc == test_dag_path)
+                .scalar()
+            )
+            assert active_dag_count == 0
+
+            serialized_dag_count = (
+                session.query(func.count(SerializedDagModel.dag_id))
+                .filter(SerializedDagModel.fileloc == test_dag_path)
+                .scalar()
+            )
+            assert serialized_dag_count == 0
+
 
 def _wait_for_processor_agent_to_complete_in_async_mode(processor_agent: DagFileProcessorAgent):
     start_timer = time.monotonic()