This is an automated email from the ASF dual-hosted git repository.
eladkal 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 e3f42b7f656 Be a little clearer when task likely OOM error (#55602)
e3f42b7f656 is described below
commit e3f42b7f656fec43dcd51c76f700724df14920b3
Author: Daniel Standish <[email protected]>
AuthorDate: Sat Sep 13 19:56:06 2025 -0700
Be a little clearer when task likely OOM error (#55602)
* Be a little clearer when task likely OOM error
When a task process is terminated with exit code -9 (sigkill) then there's
a good chance it's due to out of memory error (OOM). We can be a little more
direct about indicating that in the logs. And I reorganize the way we present
that in the docs a bit.
* fix test
---
airflow-core/docs/troubleshooting.rst | 22 ++++++++++++++++------
.../src/airflow/sdk/execution_time/supervisor.py | 5 +++--
.../task_sdk/execution_time/test_supervisor.py | 16 ++++++++++++----
3 files changed, 31 insertions(+), 12 deletions(-)
diff --git a/airflow-core/docs/troubleshooting.rst
b/airflow-core/docs/troubleshooting.rst
index 6756ab87e10..b5c904cac96 100644
--- a/airflow-core/docs/troubleshooting.rst
+++ b/airflow-core/docs/troubleshooting.rst
@@ -36,16 +36,26 @@ Below are some example scenarios that could cause a task's
state to change by a
- A user marked the task as successful or failed in the Airflow UI.
- An external script or process used the :doc:`Airflow REST API
<stable-rest-api-ref>` to change the state of a task.
-TaskRunner killed
------------------
+Process terminated by signal
+----------------------------
Sometimes, Airflow or some adjacent system will kill a task instance's
``TaskRunner``, causing the task instance to fail.
-Here are some examples that could cause such an event:
+Below we discuss a few common cases.
-- A Dag run timeout, specified by ``dagrun_timeout`` in the Dag's definition.
-- An Airflow worker running out of memory
- - Usually, Airflow workers that run out of memory receive a SIGKILL, and the
scheduler will fail the corresponding task instance for not having a heartbeat.
However, in some scenarios, Airflow kills the task before that happens.
+Dag run timeout
+"""""""""""""""
+
+A dag run timeout can be specified by ``dagrun_timeout`` in the dag's
definition.
+The task process would likely be killed with SIGTERM (exit code -15).
+
+Out of memory error (OOM)
+"""""""""""""""""""""""""
+
+When a task process consumes too much memory for a worker, the best case
scenario is it is killed
+with SIGTERM (exit code -9). Depending on configuration and infrastructure, it
is also
+possible that the whole worker will be killed due to OOM and then the tasks
would be marked as
+failed after failing to heartbeat.
Lingering task supervisor processes
-----------------------------------
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index e4818e7756f..4969ed8a1d4 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -803,11 +803,12 @@ class WatchedSubprocess:
self.process_log.critical(SIGSEGV_MESSAGE)
# psutil turns signal exit codes into an enum for us. Handy.
(Otherwise it's a plain integer) if exit_code and (name := getattr(exit_code,
"name")):
elif name := getattr(self._exit_code, "name", None):
- message = "Process terminated by signal"
+ message = "Process terminated by signal."
level = logging.ERROR
if self._exit_code == -signal.SIGKILL:
- message += ". For more information, see
https://airflow.apache.org/docs/apache-airflow/stable/troubleshooting.html#TaskRunner-killed";
+ message += " Likely out of memory error (OOM)."
level = logging.CRITICAL
+ message += " For more information, see
https://airflow.apache.org/docs/apache-airflow/stable/troubleshooting.html#process-terminated-by-signal.";
self.process_log.log(level, message,
signal=int(self._exit_code), signal_name=name)
elif self._exit_code:
# Run of the mill exit code (1, 42, etc).
diff --git a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
index f36bc8a191a..ae39595197a 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
@@ -930,21 +930,29 @@ class TestWatchedSubprocess:
mock_logger.warning.assert_not_called()
@pytest.mark.parametrize(
- ["signal_to_raise", "log_pattern"],
+ ["signal_to_raise", "log_pattern", "level"],
(
pytest.param(
signal.SIGKILL,
- re.compile(r"Process terminated by signal. For more
information, see"),
+ re.compile(r"Process terminated by signal. Likely out of
memory error"),
+ "critical",
id="kill",
),
+ pytest.param(
+ signal.SIGTERM,
+ re.compile(r"Process terminated by signal. For more
information"),
+ "error",
+ id="term",
+ ),
pytest.param(
signal.SIGSEGV,
re.compile(r".*SIGSEGV \(Segmentation Violation\) signal
indicates", re.DOTALL),
+ "critical",
id="segv",
),
),
)
- def test_exit_by_signal(self, signal_to_raise, log_pattern, cap_structlog,
client_with_ti_start):
+ def test_exit_by_signal(self, signal_to_raise, log_pattern, level,
cap_structlog, client_with_ti_start):
def subprocess_main():
import faulthandler
import os
@@ -976,7 +984,7 @@ class TestWatchedSubprocess:
rc = proc.wait()
assert {
- "log_level": "critical",
+ "log_level": level,
"event": log_pattern,
} in cap_structlog
assert rc == -signal_to_raise
