This is an automated email from the ASF dual-hosted git repository.
dabla pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new d9bfef031f9 Make sure stale file handler doesn't crash the
DagFileProcessorManager (#69324)
d9bfef031f9 is described below
commit d9bfef031f99ed310d657789b7a2057f2d5f9682
Author: David Blain <[email protected]>
AuthorDate: Thu Jul 9 13:05:02 2026 +0200
Make sure stale file handler doesn't crash the DagFileProcessorManager
(#69324)
* refactor: Make sure state file handler doesn't crash the
DagProcessorManager
* 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
* refactor: Added more context to the warning statement if close of file
handler failed
* refactor: Replaced MagicMock with mock_processor in
test_terminate_orphan_processes_tolerates_stale_file_handle_on_close
---
airflow-core/src/airflow/dag_processing/manager.py | 6 +--
.../src/airflow/dag_processing/processor.py | 13 ++++-
.../tests/unit/dag_processing/test_manager.py | 55 ++++++++++++++++++++++
3 files changed, 70 insertions(+), 4 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index f41883edf31..cd6e48b542c 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -1185,7 +1185,7 @@ class DagFileProcessorManager(LoggingMixin):
),
)
processor.kill(signal.SIGKILL)
- processor.logger_filehandle.close()
+ processor.close()
self._file_stats.pop(file, None)
@provide_session
@@ -1316,7 +1316,7 @@ class DagFileProcessorManager(LoggingMixin):
for file in finished:
processor = self._processors.pop(file)
- processor.logger_filehandle.close()
+ processor.close()
def _get_log_dir(self) -> str:
return os.path.join(self.base_log_dir,
timezone.utcnow().strftime("%Y-%m-%d"))
@@ -1622,7 +1622,7 @@ class DagFileProcessorManager(LoggingMixin):
# Clean up `self._processors` after iterating over it
for proc in processors_to_remove:
processor = self._processors.pop(proc)
- processor.logger_filehandle.close()
+ 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 88a486f99c6..ca7539131f6 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.serialization.serialized_objects import
DagSerialization, LazyDeser
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,13 @@ class DagFileProcessorProcess(WatchedSubprocess):
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,
+ exc_info=True,
+ )
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index ee2d7cd76d4..ae28f3317fb 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -497,6 +497,23 @@ class TestDagFileProcessorManager:
)
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, _ = self.mock_processor()
+ processor.logger_filehandle.close.side_effect = OSError(116, "Stale
file handle")
+
+ manager._processors[versioned_file] = processor
+
+ with (
+ mock.patch.object(type(processor), "kill"),
+ mock.patch("airflow.dag_processing.manager.stats.decr"),
+ ):
+ manager.terminate_orphan_processes(present=set())
+
+ assert manager._processors == {}
+
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 +1201,26 @@ class TestDagFileProcessorManager:
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 +1375,24 @@ class TestDagFileProcessorManager:
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"),
