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

airflow/config_templates/config.yml

@@ -500,6 +500,14 @@ core:
       type: integer
       example: ~
       default: "4096"
+    enable_purging_stale_dags:
+      description: |
+        Stale dags are deleted by default, [core] enable_purging_stale_dags is 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

@@ -475,8 +475,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:
@@ -599,7 +599,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", "enable_purging_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,10 @@ 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
+------------------------------
+
+In a versioned DAG context, there may be a need to run two versions of DAGs in parallel in two versions of workers.
+It can also be useful to keep the allowed DAGs if they are still in progress in the worker (n -1).
+To keep the stale DAGs, you can change the value of the variable ``AIRFLOW__CORE__ENABLE_PURGING_STALE_DAGS`` to ``False``. By default, it is set to ``True`` in config.yml.

tests/dag_processing/test_job_runner.py

@@ -576,7 +576,7 @@ def test_recently_modified_file_is_parsed_with_mtime_mode(
                 > (freezed_base_time - manager.processor.get_last_finish_time("file_1.py")).total_seconds()
             )
 
-    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.
@@ -629,7 +629,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))
@@ -652,7 +652,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
         """
@@ -700,7 +700,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
@@ -1370,6 +1370,146 @@ def test_callback_queue(self, tmp_path):
             dag1_req2,
         ]
 
+    @conf_vars({("core", "enable_purging_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  conf.getboolean("core", "enable_purging_stale_dags") == False
+
+        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", "enable_purging_stale_dags") == True
+
+        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()
@@ -1680,4 +1820,4 @@ def test_not_log_to_stdout(self, capfd):
 
         # Capture the stdout and stderr
         out, _ = capfd.readouterr()
-        assert "DAG File Processing Stats" not in out
+        assert "DAG File Processing Stats" not in out