This is an automated email from the ASF dual-hosted git repository.
amoghdesai pushed a commit to branch v3-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-0-test by this push:
new 2624939caab [v3-0-test] Improve detection and handling of timed out
DAG processor processes (#49868) (#49925)
2624939caab is described below
commit 2624939caabb85f035853db45127cb6849f5dc57
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Apr 29 12:22:36 2025 +0530
[v3-0-test] Improve detection and handling of timed out DAG processor
processes (#49868) (#49925)
(cherry picked from commit af10644256e923a746a6c82ff6b25f58bb67a742)
Co-authored-by: Amogh Desai <[email protected]>
Co-authored-by: Kaxil Naik <[email protected]>
Co-authored-by: Ash Berlin-Taylor <[email protected]>
---
airflow-core/src/airflow/dag_processing/manager.py | 2 +-
airflow-core/src/airflow/dag_processing/processor.py | 4 ----
airflow-core/tests/unit/dag_processing/test_manager.py | 8 ++++----
task-sdk/src/airflow/sdk/execution_time/supervisor.py | 4 ++++
4 files changed, 9 insertions(+), 9 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index b8ddf688dcb..324813cec38 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -992,7 +992,7 @@ class DagFileProcessorManager(LoggingMixin):
def _kill_timed_out_processors(self):
"""Kill any file processors that timeout to defend against process
hangs."""
- now = time.time()
+ now = time.monotonic()
processors_to_remove = []
for file, processor in self._processors.items():
duration = now - processor.start_time
diff --git a/airflow-core/src/airflow/dag_processing/processor.py
b/airflow-core/src/airflow/dag_processing/processor.py
index 25f9e2a73ed..ef3e5cb69ca 100644
--- a/airflow-core/src/airflow/dag_processing/processor.py
+++ b/airflow-core/src/airflow/dag_processing/processor.py
@@ -312,9 +312,5 @@ class DagFileProcessorProcess(WatchedSubprocess):
return self._num_open_sockets == 0
- @property
- def start_time(self) -> float:
- return self._process.create_time()
-
def wait(self) -> int:
raise NotImplementedError(f"Don't call wait on {type(self).__name__}
objects")
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index 3c180afa76b..ebf684c6788 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -133,7 +133,7 @@ class TestDagFileProcessorManager:
clear_db_import_errors()
clear_db_dag_bundles()
- def mock_processor(self) -> tuple[DagFileProcessorProcess, socket]:
+ def mock_processor(self, start_time: float | None = None) ->
tuple[DagFileProcessorProcess, socket]:
proc = MagicMock()
logger_filehandle = MagicMock()
proc.create_time.return_value = time.time()
@@ -148,6 +148,8 @@ class TestDagFileProcessorManager:
requests_fd=123,
logger_filehandle=logger_filehandle,
)
+ if start_time:
+ ret.start_time = start_time
ret._num_open_sockets = 0
return ret, read_end
@@ -518,9 +520,7 @@ class TestDagFileProcessorManager:
def test_kill_timed_out_processors_kill(self):
manager = DagFileProcessorManager(max_runs=1, processor_timeout=5)
-
- processor, _ = self.mock_processor()
- processor._process.create_time.return_value =
timezone.make_aware(datetime.min).timestamp()
+ processor, _ = self.mock_processor(start_time=16000)
manager._processors = {
DagFileInfo(
bundle_name="testing", rel_path=Path("abc.txt"),
bundle_path=TEST_DAGS_FOLDER
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index db9474e68d5..0b77156578b 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -428,6 +428,9 @@ class WatchedSubprocess:
subprocess_logs_to_stdout: bool = False
"""Duplicate log messages to stdout, or only send them to
``self.process_log``."""
+ start_time: float = attrs.field(factory=time.monotonic)
+ """The start time of the child process."""
+
@classmethod
def start(
cls,
@@ -481,6 +484,7 @@ class WatchedSubprocess:
process=psutil.Process(pid),
requests_fd=requests_fd,
process_log=logger,
+ start_time=time.monotonic(),
**constructor_kwargs,
)
