From ac9876a0d4e1f519c07153bdd1c4110288b45794 Mon Sep 17 00:00:00 2001 From: David Blain Date: Fri, 3 Jul 2026 12:00:41 +0200 Subject: [PATCH 1/3] refactor: Make sure state file handler doesn't crash the DagProcessorManager --- .../src/airflow/dag_processing/manager.py | 15 ++++-- .../tests/unit/dag_processing/test_manager.py | 53 +++++++++++++++++++ 2 files changed, 65 insertions(+), 3 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index b2428985a6506..afd3b8a2188d1 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/airflow-core/src/airflow/dag_processing/manager.py @@ -1181,7 +1181,10 @@ def terminate_orphan_processes(self, present: set[DagFileInfo]): ), ) processor.kill(signal.SIGKILL) - processor.logger_filehandle.close() + try: + processor.logger_filehandle.close() + except OSError: + self.log.warning("Failed to close log file handle for %s", file_name) self._file_stats.pop(file, None) @provide_session @@ -1312,7 +1315,10 @@ def _collect_results(self): for file in finished: processor = self._processors.pop(file) - processor.logger_filehandle.close() + try: + processor.logger_filehandle.close() + except OSError: + self.log.warning("Failed to close log file handle for %s", file) def _get_log_dir(self) -> str: return os.path.join(self.base_log_dir, timezone.utcnow().strftime("%Y-%m-%d")) @@ -1618,7 +1624,10 @@ def _kill_timed_out_processors(self): # Clean up `self._processors` after iterating over it for proc in processors_to_remove: processor = self._processors.pop(proc) - processor.logger_filehandle.close() + try: + processor.logger_filehandle.close() + except OSError: + self.log.warning("Failed to close log file handle for %s", proc) def _add_files_to_queue( self, diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index 4b55b6b2f1ee9..3c763fc5e676b 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/airflow-core/tests/unit/dag_processing/test_manager.py @@ -497,6 +497,21 @@ def test_terminate_orphan_processes_kills_processor_when_file_is_truly_absent(se ) processor.kill.assert_called_once_with(signal.SIGKILL) + def test_terminate_orphan_processes_tolerates_stale_file_handle_on_close(self): + """A stale NFS file handle on close (e.g. OpenShift) must not crash the manager.""" + manager = DagFileProcessorManager(max_runs=1) + versioned_file = _get_versioned_file_info("callbacks.py") + processor = MagicMock() + processor.logger_filehandle.close.side_effect = OSError(116, "Stale file handle") + + manager._processors[versioned_file] = processor + + with mock.patch("airflow.dag_processing.manager.stats.decr"): + manager.terminate_orphan_processes(present=set()) + + assert manager._processors == {} + processor.kill.assert_called_once_with(signal.SIGKILL) + def test_remove_orphaned_file_stats_keeps_versioned_callback_stats_when_unversioned_file_is_present(self): manager = DagFileProcessorManager(max_runs=1) versioned_file = _get_versioned_file_info("callbacks.py") @@ -1184,6 +1199,26 @@ def test_kill_timed_out_processors_kill(self): assert len(manager._processors) == 0 processor.logger_filehandle.close.assert_called() + def test_kill_timed_out_processors_tolerates_stale_file_handle_on_close(self): + """A stale NFS file handle on close (e.g. OpenShift) must not crash the manager.""" + manager = DagFileProcessorManager(max_runs=1, processor_timeout=5) + start_time = time.monotonic() - manager.processor_timeout - 1 + processor, _ = self.mock_processor(start_time=start_time) + processor.logger_filehandle.close.side_effect = OSError(116, "Stale file handle") + manager._processors = { + DagFileInfo( + bundle_name="testing", rel_path=Path("abc.py"), bundle_path=TEST_DAGS_FOLDER + ): processor + } + with ( + mock.patch.object(type(processor), "kill"), + mock.patch("airflow.dag_processing.manager.stats.decr"), + mock.patch("airflow.dag_processing.manager.stats.incr"), + ): + manager._kill_timed_out_processors() + + assert len(manager._processors) == 0 + def test_kill_timed_out_processors_no_kill(self): manager = DagFileProcessorManager( max_runs=1, @@ -1338,6 +1373,24 @@ def test_collect_results_processes_remaining_files_when_one_persist_fails(self, assert manager._file_stats[file_b].run_count == 2 assert len(manager._processors) == 0 + def test_collect_results_tolerates_stale_file_handle_on_close(self): + """A stale NFS file handle on close (e.g. OpenShift) must not crash the manager.""" + manager = DagFileProcessorManager(max_runs=1) + file = DagFileInfo(bundle_name="testing", rel_path=Path("a.py"), bundle_path=TEST_DAGS_FOLDER) + manager._file_stats[file] = DagFileStat() + manager._bundle_versions["testing"] = "v1" + + proc, _ = self.mock_processor(start_time=time.monotonic() - 1) + proc.had_callbacks = False + proc.parsing_result = DagFileParsingResult(fileloc="a.py", serialized_dags=[]) + proc.logger_filehandle.close.side_effect = OSError(116, "Stale file handle") + manager._processors = {file: proc} + + with mock.patch.object(manager, "persist_parsing_result"): + manager._collect_results() + + assert len(manager._processors) == 0 + @pytest.mark.usefixtures("testing_dag_bundle") @pytest.mark.parametrize( ("callbacks", "path", "expected_body"), From 2cb5083c4c79775252d3a482a37a81111b7d69c4 Mon Sep 17 00:00:00 2001 From: David Blain Date: Fri, 3 Jul 2026 22:42:25 +0200 Subject: [PATCH 2/3] refactor: Added close method on DagFileProcessorProcess so we have a more DRY approach on how to close the logger file handler and catch OSError if file handler is stale --- .../src/airflow/dag_processing/manager.py | 15 +++------------ .../src/airflow/dag_processing/processor.py | 9 ++++++++- 2 files changed, 11 insertions(+), 13 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index afd3b8a2188d1..71968d2cb2bdc 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/airflow-core/src/airflow/dag_processing/manager.py @@ -1181,10 +1181,7 @@ def terminate_orphan_processes(self, present: set[DagFileInfo]): ), ) processor.kill(signal.SIGKILL) - try: - processor.logger_filehandle.close() - except OSError: - self.log.warning("Failed to close log file handle for %s", file_name) + processor.close() self._file_stats.pop(file, None) @provide_session @@ -1315,10 +1312,7 @@ def _collect_results(self): for file in finished: processor = self._processors.pop(file) - try: - processor.logger_filehandle.close() - except OSError: - self.log.warning("Failed to close log file handle for %s", file) + processor.close() def _get_log_dir(self) -> str: return os.path.join(self.base_log_dir, timezone.utcnow().strftime("%Y-%m-%d")) @@ -1624,10 +1618,7 @@ def _kill_timed_out_processors(self): # Clean up `self._processors` after iterating over it for proc in processors_to_remove: processor = self._processors.pop(proc) - try: - processor.logger_filehandle.close() - except OSError: - self.log.warning("Failed to close log file handle for %s", proc) + processor.close() def _add_files_to_queue( self, diff --git a/airflow-core/src/airflow/dag_processing/processor.py b/airflow-core/src/airflow/dag_processing/processor.py index 88a486f99c695..589981c682b9f 100644 --- a/airflow-core/src/airflow/dag_processing/processor.py +++ b/airflow-core/src/airflow/dag_processing/processor.py @@ -93,6 +93,7 @@ from airflow.utils.dag_version_inflation_checker import check_dag_file_stability from airflow.utils.file import iter_airflow_imports from airflow.utils.helpers import prune_dict +from airflow.utils.log.logging_mixin import LoggingMixin from airflow.utils.state import TaskInstanceState if TYPE_CHECKING: @@ -550,7 +551,7 @@ def in_process_api_server() -> InProcessExecutionAPI: @attrs.define(kw_only=True) -class DagFileProcessorProcess(WatchedSubprocess): +class DagFileProcessorProcess(WatchedSubprocess, LoggingMixin): """ Parses dags with Task SDK API. @@ -730,3 +731,9 @@ def is_ready(self) -> bool: def wait(self) -> int: raise NotImplementedError(f"Don't call wait on {type(self).__name__} objects") + + def close(self): + try: + self.logger_filehandle.close() + except OSError: + self.log.warning("Failed to close log file handle for %s", self.dag_file_rel_path) From 3e89c1aef338d52f51ae4e90c2e153898ece2ee1 Mon Sep 17 00:00:00 2001 From: David Blain Date: Fri, 3 Jul 2026 22:46:04 +0200 Subject: [PATCH 3/3] refactor: Added more context to the warning statement if close of file handler failed --- airflow-core/src/airflow/dag_processing/processor.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/airflow-core/src/airflow/dag_processing/processor.py b/airflow-core/src/airflow/dag_processing/processor.py index 589981c682b9f..ca7539131f60e 100644 --- a/airflow-core/src/airflow/dag_processing/processor.py +++ b/airflow-core/src/airflow/dag_processing/processor.py @@ -736,4 +736,8 @@ def close(self): try: self.logger_filehandle.close() except OSError: - self.log.warning("Failed to close log file handle for %s", self.dag_file_rel_path) + self.log.warning( + "Failed to close log file handle for %s", + self.dag_file_rel_path, + exc_info=True, + )