Remove scan_stale_dag method in DAG processor manager

This method has been replaced by `handle_removed_files` and `deactivate_deleted_dags`
method. Its present now causes issues as it deactivates DAGs incorrectly. `handle_removed_files`
is a better method more suited to dag bundles as the file's processor is also terminated.

Also removed unused config variable in scheduler and config.yml
Closes: apache#47294

Author: ephraimbuddy

airflow/cli/commands/remote_commands/config_command.py

@@ -379,7 +379,11 @@ def message(self) -> str | None:
     ),
     ConfigChange(
         config=ConfigParameter("scheduler", "deactivate_stale_dags_interval"),
-        renamed_to=ConfigParameter("scheduler", "parsing_cleanup_interval"),
+        was_removed=True,
+    ),
+    ConfigChange(
+        config=ConfigParameter("scheduler", "dag_stale_not_seen_duration"),
+        was_removed=True,
     ),
     ConfigChange(
         config=ConfigParameter("scheduler", "statsd_on"), renamed_to=ConfigParameter("metrics", "statsd_on")
@@ -442,7 +446,7 @@ def message(self) -> str | None:
     ),
     ConfigChange(
         config=ConfigParameter("scheduler", "stale_dag_threshold"),
-        renamed_to=ConfigParameter("dag_processor", "stale_dag_threshold"),
+        was_removed=True,
     ),
     ConfigChange(
         config=ConfigParameter("scheduler", "print_stats_interval"),

airflow/config_templates/config.yml

@@ -2448,13 +2448,6 @@ scheduler:
       type: boolean
       example: ~
       default: "True"
-    dag_stale_not_seen_duration:
-      description: |
-        Time in seconds after which dags, which were not updated by Dag Processor are deactivated.
-      version_added: 2.4.0
-      type: integer
-      example: ~
-      default: "600"
     use_job_schedule:
       description: |
         Turn off scheduler use of cron intervals by setting this to ``False``.
@@ -2726,18 +2719,6 @@ dag_processor:
       type: integer
       example: ~
       default: "30"
-    stale_dag_threshold:
-      description: |
-        How long (in seconds) to wait after we have re-parsed a DAG file before deactivating stale
-        DAGs (DAGs which are no longer present in the expected files). The reason why we need
-        this threshold is to account for the time between when the file is parsed and when the
-        DAG is loaded. The absolute maximum that this could take is
-        ``[dag_processor] dag_file_processor_timeout``, but when you have a long timeout configured,
-        it results in a significant delay in the deactivation of stale dags.
-      version_added: ~
-      type: integer
-      example: ~
-      default: "50"
     dag_file_processor_timeout:
       description: |
         How long before timing out a DagFileProcessor, which processes a dag file

airflow/dag_processing/manager.py

@@ -33,15 +33,15 @@
 from collections import defaultdict, deque
 from collections.abc import Callable, Iterable, Iterator
 from dataclasses import dataclass, field
-from datetime import datetime, timedelta
+from datetime import datetime
 from importlib import import_module
 from operator import attrgetter, itemgetter
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, NamedTuple, cast
 
 import attrs
 import structlog
-from sqlalchemy import select, update
+from sqlalchemy import select
 from sqlalchemy.orm import load_only
 from tabulate import tabulate
 from uuid6 import uuid7
@@ -158,17 +158,9 @@ class DagFileProcessorManager(LoggingMixin):
 
     _parallelism: int = attrs.field(factory=_config_int_factory("dag_processor", "parsing_processes"))
 
-    parsing_cleanup_interval: float = attrs.field(
-        factory=_config_int_factory("scheduler", "parsing_cleanup_interval")
-    )
     _file_process_interval: float = attrs.field(
         factory=_config_int_factory("dag_processor", "min_file_process_interval")
     )
-    stale_dag_threshold: float = attrs.field(
-        factory=_config_int_factory("dag_processor", "stale_dag_threshold")
-    )
-
-    _last_deactivate_stale_dags_time: float = attrs.field(default=0, init=False)
     print_stats_interval: float = attrs.field(
         factory=_config_int_factory("dag_processor", "print_stats_interval")
     )
@@ -251,59 +243,6 @@ def run(self):
 
         return self._run_parsing_loop()
 
-    def _scan_stale_dags(self):
-        """Scan and deactivate DAGs which are no longer present in files."""
-        now = time.monotonic()
-        elapsed_time_since_refresh = now - self._last_deactivate_stale_dags_time
-        if elapsed_time_since_refresh > self.parsing_cleanup_interval:
-            last_parsed = {
-                file_info: stat.last_finish_time
-                for file_info, stat in self._file_stats.items()
-                if stat.last_finish_time
-            }
-            self.deactivate_stale_dags(last_parsed=last_parsed)
-            self._last_deactivate_stale_dags_time = time.monotonic()
-
-    @provide_session
-    def deactivate_stale_dags(
-        self,
-        last_parsed: dict[DagFileInfo, datetime | None],
-        session: Session = NEW_SESSION,
-    ):
-        """Detect and deactivate DAGs which are no longer present in files."""
-        to_deactivate = set()
-        bundle_names = {b.name for b in self._dag_bundles}
-        query = select(
-            DagModel.dag_id,
-            DagModel.bundle_name,
-            DagModel.fileloc,
-            DagModel.last_parsed_time,
-            DagModel.relative_fileloc,
-        ).where(DagModel.is_active, DagModel.bundle_name.in_(bundle_names))
-        dags_parsed = session.execute(query)
-
-        for dag in dags_parsed:
-            # The largest valid difference between a DagFileStat's last_finished_time and a DAG's
-            # last_parsed_time is the processor_timeout. Longer than that indicates that the DAG is
-            # no longer present in the file. We have a stale_dag_threshold configured to prevent a
-            # significant delay in deactivation of stale dags when a large timeout is configured
-            file_info = DagFileInfo(rel_path=Path(dag.relative_fileloc), bundle_name=dag.bundle_name)
-            if last_finish_time := last_parsed.get(file_info, None):
-                if dag.last_parsed_time + timedelta(seconds=self.stale_dag_threshold) < last_finish_time:
-                    self.log.info("DAG %s is missing and will be deactivated.", dag.dag_id)
-                    to_deactivate.add(dag.dag_id)
-
-        if to_deactivate:
-            deactivated_dagmodel = session.execute(
-                update(DagModel)
-                .where(DagModel.dag_id.in_(to_deactivate))
-                .values(is_active=False)
-                .execution_options(synchronize_session="fetch")
-            )
-            deactivated = deactivated_dagmodel.rowcount
-            if deactivated:
-                self.log.info("Deactivated %i DAGs which are no longer present in file.", deactivated)
-
     def _run_parsing_loop(self):
         # initialize cache to mutualize calls to Variable.get in DAGs
         # needs to be done before this process is forked to create the DAG parsing processes.
@@ -342,7 +281,6 @@ def _run_parsing_loop(self):
 
             for callback in self._fetch_callbacks():
                 self._add_callback_to_queue(callback)
-            self._scan_stale_dags()
             DagWarning.purge_inactive_dag_warnings()
 
             # Update number of loop iteration.

airflow/jobs/scheduler_job_runner.py

@@ -179,7 +179,6 @@ def __init__(
         self._task_instance_heartbeat_timeout_secs = conf.getint(
             "scheduler", "task_instance_heartbeat_timeout"
         )
-        self._dag_stale_not_seen_duration = conf.getint("scheduler", "dag_stale_not_seen_duration")
         self._task_queued_timeout = conf.getfloat("scheduler", "task_queued_timeout")
         self._enable_tracemalloc = conf.getboolean("scheduler", "enable_tracemalloc")
 

tests/dag_processing/test_manager.py

@@ -36,7 +36,7 @@
 
 import pytest
 import time_machine
-from sqlalchemy import func, select
+from sqlalchemy import select
 from uuid6 import uuid7
 
 from airflow.callbacks.callback_requests import DagCallbackRequest
@@ -413,83 +413,6 @@ def test_file_paths_in_queue_sorted_by_priority(self):
             parsing_request_after = session2.query(DagPriorityParsingRequest).get(parsing_request.id)
         assert parsing_request_after is None
 
-    def test_scan_stale_dags(self, testing_dag_bundle):
-        """
-        Ensure that DAGs are marked inactive when the file is parsed but the
-        DagModel.last_parsed_time is not updated.
-        """
-        manager = DagFileProcessorManager(
-            max_runs=1,
-            processor_timeout=10 * 60,
-        )
-        bundle = MagicMock()
-        bundle.name = "testing"
-        manager._dag_bundles = [bundle]
-
-        test_dag_path = DagFileInfo(
-            bundle_name="testing",
-            rel_path=Path("test_example_bash_operator.py"),
-            bundle_path=TEST_DAGS_FOLDER,
-        )
-        dagbag = DagBag(
-            test_dag_path.absolute_path,
-            read_dags_from_db=False,
-            include_examples=False,
-            bundle_path=test_dag_path.bundle_path,
-        )
-
-        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.bulk_write_to_db("testing", None, [dag])
-            SerializedDagModel.write_dag(dag, bundle_name="testing")
-
-            # 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=1,
-                run_count=1,
-                last_num_of_db_queries=1,
-            )
-            manager._files = [test_dag_path]
-            manager._file_stats[test_dag_path] = stat
-
-            active_dag_count = (
-                session.query(func.count(DagModel.dag_id))
-                .filter(
-                    DagModel.is_active,
-                    DagModel.relative_fileloc == str(test_dag_path.rel_path),
-                    DagModel.bundle_name == test_dag_path.bundle_name,
-                )
-                .scalar()
-            )
-            assert active_dag_count == 1
-
-            manager._scan_stale_dags()
-
-            active_dag_count = (
-                session.query(func.count(DagModel.dag_id))
-                .filter(
-                    DagModel.is_active,
-                    DagModel.relative_fileloc == str(test_dag_path.rel_path),
-                    DagModel.bundle_name == test_dag_path.bundle_name,
-                )
-                .scalar()
-            )
-            assert active_dag_count == 0
-
-            serialized_dag_count = (
-                session.query(func.count(SerializedDagModel.dag_id))
-                .filter(SerializedDagModel.dag_id == dag.dag_id)
-                .scalar()
-            )
-            # Deactivating the DagModel should not delete the SerializedDagModel
-            # SerializedDagModel gives history about Dags
-            assert serialized_dag_count == 1
-
     def test_kill_timed_out_processors_kill(self):
         manager = DagFileProcessorManager(max_runs=1, processor_timeout=5)