diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index a4a834b9acec1..a150b4c1fba5f 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/airflow-core/src/airflow/dag_processing/manager.py @@ -1573,8 +1573,21 @@ def _resort_file_queue(self): # Put callback files at the front, then sorted regular files self._file_queue = OrderedDict.fromkeys(callback_files + sorted_regular_files) + def _resolve_stats_by_presence_key(self) -> dict[tuple[str, Path], DagFileStat]: + """ + Resolve one stat per physical file from the bundle scan's entries. + + A callback pins ``bundle_version``, so its entry is separate from the scan's and does + not track parses of the current file: a callback-only result has no ``last_finish_time``, + and a timed-out callback processor is stamped although nothing parsed the current version. + A file with no scan entry has no stat, so it is queued. + """ + return { + file.presence_key: stat for file, stat in self._file_stats.items() if file.bundle_version is None + } + def _sort_by_mtime(self, files: Iterable[DagFileInfo]): - file_stats_by_presence_key = {file.presence_key: stat for file, stat in self._file_stats.items()} + file_stats_by_presence_key = self._resolve_stats_by_presence_key() files_with_mtime: dict[DagFileInfo, float] = {} changed_recently = set() for file in files: @@ -1624,19 +1637,19 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]): Note this method is only called when the file path queue is empty """ + file_stats_by_presence_key = self._resolve_stats_by_presence_key() # We only emit metrics after processing all files in the queue. If `self._parsing_start_time` is None # when this method is called, no files have yet been added to the queue so we shouldn't emit metrics. if self._parsing_start_time is not None: emit_metrics( parse_time=time.perf_counter() - self._parsing_start_time, - dag_file_stats=list(self._file_stats.values()), + dag_file_stats=list(file_stats_by_presence_key.values()), ) self._parsing_start_time = None # If the file path is already being processed, or if a file was # processed recently, wait until the next batch in_progress_keys = {file.presence_key for file in self._processors} - file_stats_by_presence_key = {file.presence_key: stat for file, stat in self._file_stats.items()} now = timezone.utcnow() # Sort the file paths by the parsing order mode diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index 322d7dd83da5c..275422c78e459 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/airflow-core/tests/unit/dag_processing/test_manager.py @@ -1147,7 +1147,8 @@ def test_prepare_file_queue_skips_file_when_versioned_processor_is_in_progress(s assert manager._file_queue == OrderedDict() @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"}) - def test_prepare_file_queue_skips_file_when_versioned_stat_is_at_run_limit(self): + def test_prepare_file_queue_ignores_versioned_stat_at_run_limit(self): + """A callback's versioned stat doesn't stand for a parse, so it doesn't count towards the run limit.""" manager = DagFileProcessorManager(max_runs=1) versioned_file = _get_versioned_file_info("file_1.py") known_file = _get_file_infos(["file_1.py"])[0] @@ -1156,10 +1157,11 @@ def test_prepare_file_queue_skips_file_when_versioned_stat_is_at_run_limit(self) manager.prepare_file_queue(known_files={"testing": {known_file}}) - assert manager._file_queue == OrderedDict() + assert manager._file_queue == OrderedDict.fromkeys([known_file]) @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"}) - def test_prepare_file_queue_skips_recently_processed_file_with_versioned_stats(self): + def test_prepare_file_queue_ignores_recent_versioned_stat(self): + """A callback's versioned stat doesn't stand for a parse, so a recent one doesn't hold the file back.""" manager = DagFileProcessorManager(max_runs=3) versioned_file = _get_versioned_file_info("file_1.py") known_file = _get_file_infos(["file_1.py"])[0] @@ -1171,11 +1173,11 @@ def test_prepare_file_queue_skips_recently_processed_file_with_versioned_stats(s manager.prepare_file_queue(known_files={"testing": {known_file}}) - assert manager._file_queue == OrderedDict() + assert manager._file_queue == OrderedDict.fromkeys([known_file]) @conf_vars({("dag_processor", "file_parsing_sort_mode"): "modified_time"}) @mock.patch("airflow.utils.file.os.path.getmtime") - def test_recently_modified_file_uses_versioned_stats_without_creating_duplicate_entries( + def test_recently_modified_file_with_versioned_stat_does_not_create_duplicate_entries( self, mock_getmtime ): freezed_base_time = timezone.datetime(2020, 1, 5, 0, 0, 0) @@ -1197,6 +1199,106 @@ def test_recently_modified_file_uses_versioned_stats_without_creating_duplicate_ assert known_file not in manager._file_stats assert versioned_file in manager._file_stats + @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"}) + @pytest.mark.parametrize( + "callback_first", + [ + pytest.param(False, id="scan-recorded-first"), + pytest.param(True, id="callback-recorded-first"), + ], + ) + def test_callback_only_parse_does_not_requeue_recently_parsed_file(self, callback_first): + """A callback-only parse records no ``last_finish_time``, which must not requeue the file.""" + freezed_base_time = timezone.datetime(2020, 1, 5, 0, 0, 0) + known_file = _get_file_infos(["file_1.py"])[0] + versioned_file = _get_versioned_file_info("file_1.py") + scan_stat = DagFileStat(1, 0, freezed_base_time - timedelta(seconds=5), 1.0, 1, 1) + + manager = DagFileProcessorManager(max_runs=3) + processor, _ = self.mock_processor(start_time=time.monotonic() - 1) + processor.had_callbacks = True + processor.parsing_result = None + + with time_machine.travel(freezed_base_time): + if not callback_first: + manager._file_stats[known_file] = scan_stat + manager.handle_parsing_result(versioned_file, processor) + assert manager._file_stats[versioned_file].last_finish_time is None + if callback_first: + manager._file_stats[known_file] = scan_stat + manager.prepare_file_queue(known_files={"testing": {known_file}}) + + assert manager._file_queue == OrderedDict() + + @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"}) + @mock.patch.object(DagFileProcessorProcess, "kill", autospec=True) + def test_timed_out_callback_does_not_delay_due_parse(self, mock_kill): + """A callback processor that times out must not make a due file look recently parsed.""" + freezed_base_time = timezone.datetime(2020, 1, 5, 0, 0, 0) + known_file = _get_file_infos(["file_1.py"])[0] + versioned_file = _get_versioned_file_info("file_1.py") + + manager = DagFileProcessorManager(max_runs=3, processor_timeout=5) + manager._file_stats[known_file] = DagFileStat( + 1, 0, freezed_base_time - timedelta(seconds=60), 1.0, 1, 1 + ) + processor, _ = self.mock_processor(start_time=time.monotonic() - manager.processor_timeout - 1) + manager._processors = {versioned_file: processor} + + with time_machine.travel(freezed_base_time): + manager._kill_timed_out_processors() + mock_kill.assert_called_once_with(processor, signal.SIGKILL) + assert manager._file_stats[versioned_file].last_finish_time is not None + manager.prepare_file_queue(known_files={"testing": {known_file}}) + + assert known_file in manager._file_queue + + @conf_vars({("dag_processor", "file_parsing_sort_mode"): "modified_time"}) + @mock.patch.object(DagFileProcessorProcess, "kill", autospec=True) + @mock.patch("airflow.utils.file.os.path.getmtime") + def test_timed_out_callback_does_not_hide_an_edit(self, mock_getmtime, mock_kill): + """An edit after the last parse is picked up even if a callback processor times out later.""" + freezed_base_time = timezone.datetime(2020, 1, 5, 0, 0, 0) + known_file = _get_file_infos(["file_1.py"])[0] + versioned_file = _get_versioned_file_info("file_1.py") + + manager = DagFileProcessorManager(max_runs=3, processor_timeout=5) + manager._file_stats[known_file] = DagFileStat( + 1, 0, freezed_base_time - timedelta(seconds=10), 1.0, 1, 1 + ) + processor, _ = self.mock_processor(start_time=time.monotonic() - manager.processor_timeout - 1) + manager._processors = {versioned_file: processor} + mock_getmtime.return_value = (freezed_base_time - timedelta(seconds=7)).timestamp() + + with time_machine.travel(freezed_base_time - timedelta(seconds=5)): + manager._kill_timed_out_processors() + mock_kill.assert_called_once_with(processor, signal.SIGKILL) + assert manager._file_stats[versioned_file].last_finish_time is not None + with time_machine.travel(freezed_base_time): + manager.prepare_file_queue(known_files={"testing": {known_file}}) + + assert known_file in manager._file_queue + + @conf_vars({("dag_processor", "file_parsing_sort_mode"): "alphabetical"}) + @mock.patch.object(DagFileProcessorProcess, "kill", autospec=True) + @mock.patch("airflow.dag_processing.manager.stats.gauge") + def test_timed_out_callback_is_not_reported_as_an_import_error(self, mock_gauge, mock_kill): + """The ``import_errors=1`` a timeout leaves on a callback's entry must not reach the gauge.""" + known_file = _get_file_infos(["file_1.py"])[0] + versioned_file = _get_versioned_file_info("file_1.py") + + manager = DagFileProcessorManager(max_runs=3, processor_timeout=5) + manager._file_stats[known_file] = DagFileStat(1, 2, timezone.datetime(2020, 1, 5), 1.0, 1, 1) + processor, _ = self.mock_processor(start_time=time.monotonic() - manager.processor_timeout - 1) + manager._processors = {versioned_file: processor} + + manager._kill_timed_out_processors() + assert manager._file_stats[versioned_file].import_errors == 1 + manager._parsing_start_time = time.perf_counter() + manager.prepare_file_queue(known_files={"testing": {known_file}}) + + mock_gauge.assert_any_call("dag_processing.import_errors", 2) + def test_file_paths_in_queue_sorted_by_priority(self): from airflow.models.dagbag import DagPriorityParsingRequest