This is an automated email from the ASF dual-hosted git repository.
jason810496 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 97d4d4e2aa9 Skip log accumulator disk spill when reading finished task
logs (#70671)
97d4d4e2aa9 is described below
commit 97d4d4e2aa920cb2cbaf7b7783217d320068bde7
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Thu Oct 29 20:25:30 2026 +0800
Skip log accumulator disk spill when reading finished task logs (#70671)
* Skip log accumulator disk spill when reading finished task logs
* Version-gate remote log handler tests for terminal-state metadata
Finished-task reads drop the log_pos continuation token from 3.4.0
onward, but the provider test matrix still runs against released cores
(3.0-3.3) that emit it. Gate the S3, GCS, and WASB task-handler metadata
assertions on AIRFLOW_V_3_4_0_PLUS so both the source run and the
older-core compatibility runs pass.
* Add plan for incremental UI log fetch via continuation token
* Wrap only unfinished log reads in the stream accumulator
Apply the log_pos skip and header once for both paths instead of
duplicating it in the finished-read branch.
* Drop the accidentally committed planning doc
* Test both accumulator paths of FileTaskHandler._read
Cover running, deferred, older-try, and finished reads, including a running
task's follow-up poll with an inbound log_pos. Update the _read docstring
for
the log_pos contract.
* Drop the misleading log_pos note in the log read test
* Drop the redundant accumulator comment in the log read test
* Assert only end_of_log in the remote log handler tests
log_pos is a core detail covered by the core tests, so the provider tests no
longer need a version gate for it.
* Apply suggestion from @uranusjr to fix the comment
Co-authored-by: Tzu-ping Chung <[email protected]>
---------
Co-authored-by: Tzu-ping Chung <[email protected]>
---
.../src/airflow/utils/log/file_task_handler.py | 40 ++++----
.../tests/unit/utils/log/test_log_reader.py | 4 +-
airflow-core/tests/unit/utils/test_log_handlers.py | 109 ++++++++++++++++++---
.../unit/amazon/aws/log/test_s3_task_handler.py | 4 +-
.../unit/google/cloud/log/test_gcs_task_handler.py | 4 +-
.../microsoft/azure/log/test_wasb_task_handler.py | 2 +-
6 files changed, 124 insertions(+), 39 deletions(-)
diff --git a/airflow-core/src/airflow/utils/log/file_task_handler.py
b/airflow-core/src/airflow/utils/log/file_task_handler.py
index c4790fe206b..e7fe50854ff 100644
--- a/airflow-core/src/airflow/utils/log/file_task_handler.py
+++ b/airflow-core/src/airflow/utils/log/file_task_handler.py
@@ -545,18 +545,16 @@ class FileTaskHandler(logging.Handler):
:param ti: task instance record
:param try_number: current try_number to read log from
:param metadata: log metadata,
- can be used for steaming log reading and auto-tailing.
+ can be used for streaming log reading and
auto-tailing.
Following attributes are used:
- log_pos: (absolute) Char position to which the log
- which was retrieved in previous calls, this
- part will be skipped and only following test
- returned to be added to tail.
- :return: log message as a string and metadata.
+ log_pos: Number of log records returned by previous
calls.
+ These records are skipped and only the
following ones are returned.
+ :return: log stream and metadata.
Following attributes are used in metadata:
end_of_log: Boolean, True if end of log is reached or False
if further calls might get more log text.
- This is determined by the status of the
TaskInstance
- log_pos: (absolute) Char position to which the log is
retrieved
+ This is determined by the state and try number of
the TaskInstance
+ log_pos: Number of log records read so far. Only present when
end_of_log is False.
"""
# Task instance here might be different from task instance when
# initializing the handler. Thus explicitly getting log location
@@ -639,21 +637,21 @@ class FileTaskHandler(logging.Handler):
TaskInstanceState.DEFERRED,
)
- with LogStreamAccumulator(out_stream, HEAP_DUMP_SIZE) as
stream_accumulator:
- log_pos = stream_accumulator.total_lines
- out_stream = stream_accumulator.stream
+ out_metadata: LogMetadata = {"end_of_log": end_of_log}
+ if not end_of_log:
+ # Counting lines for log_pos drains the stream; skip when no token
is needed.
+ with LogStreamAccumulator(out_stream, HEAP_DUMP_SIZE) as
stream_accumulator:
+ out_metadata["log_pos"] = stream_accumulator.total_lines
+ out_stream = stream_accumulator.stream
- # skip log stream until the last position
- if metadata and "log_pos" in metadata:
- out_stream = islice(out_stream, metadata["log_pos"], None)
- else:
- # first time reading log, add messages before interleaved log
stream
- out_stream = chain(header, out_stream)
+ # skip log stream until the last position
+ if metadata and "log_pos" in metadata:
+ out_stream = islice(out_stream, metadata["log_pos"], None)
+ else:
+ # first time reading log, add messages before interleaved log
stream
+ out_stream = chain(header, out_stream)
- return out_stream, {
- "end_of_log": end_of_log,
- "log_pos": log_pos,
- }
+ return out_stream, out_metadata
@staticmethod
def _get_pod_namespace(ti: TaskInstance):
diff --git a/airflow-core/tests/unit/utils/log/test_log_reader.py
b/airflow-core/tests/unit/utils/log/test_log_reader.py
index 1c65cb1806b..01ba3aa6b56 100644
--- a/airflow-core/tests/unit/utils/log/test_log_reader.py
+++ b/airflow-core/tests/unit/utils/log/test_log_reader.py
@@ -136,7 +136,7 @@ class TestLogView:
)
assert logs[2].event == "::endgroup::"
assert logs[3].event == "try_number=1."
- assert metadata == {"end_of_log": True, "log_pos": 1}
+ assert metadata == {"end_of_log": True}
def test_test_read_log_chunks_should_read_latest_files(self):
task_log_reader = TaskLogReader()
@@ -151,7 +151,7 @@ class TestLogView:
)
assert logs[2].event == "::endgroup::"
assert logs[3].event == f"try_number={ti.try_number}."
- assert metadata == {"end_of_log": True, "log_pos": 1}
+ assert metadata == {"end_of_log": True}
def test_test_test_read_log_stream_should_read_one_try(self):
task_log_reader = TaskLogReader()
diff --git a/airflow-core/tests/unit/utils/test_log_handlers.py
b/airflow-core/tests/unit/utils/test_log_handlers.py
index f779ac8b44b..412323cbf7e 100644
--- a/airflow-core/tests/unit/utils/test_log_handlers.py
+++ b/airflow-core/tests/unit/utils/test_log_handlers.py
@@ -25,7 +25,7 @@ import os
from http import HTTPStatus
from importlib import reload
from pathlib import Path
-from typing import cast
+from typing import TYPE_CHECKING, cast
from unittest import mock
from unittest.mock import patch
@@ -62,6 +62,7 @@ from airflow.utils.log.file_task_handler import (
_log_stream_to_parsed_log_stream,
_stream_lines_by_chunk,
)
+from airflow.utils.log.log_stream_accumulator import LogStreamAccumulator
from airflow.utils.log.logging_mixin import set_context
from airflow.utils.net import get_hostname
from airflow.utils.session import create_session
@@ -77,6 +78,9 @@ from tests_common.test_utils.file_task_handler import (
)
from tests_common.test_utils.markers import
skip_if_force_lowest_dependencies_marker
+if TYPE_CHECKING:
+ from collections.abc import Generator
+
pytestmark = [pytest.mark.db_test]
DEFAULT_DATE = pendulum.datetime(2016, 1, 1)
@@ -93,6 +97,18 @@ def cleanup_tables():
clear_db_connections()
+def _tracking_stream(lines: list[str]) -> tuple[Generator[str, None, None],
list[str]]:
+ """Return a stream of ``lines`` and the list of lines pulled from it so
far."""
+ pulled: list[str] = []
+
+ def gen() -> Generator[str, None, None]:
+ for line in lines:
+ pulled.append(line)
+ yield line
+
+ return gen(), pulled
+
+
class TestFileTaskLogHandler:
def clean_up(self):
with create_session() as session:
@@ -503,7 +519,7 @@ class TestFileTaskLogHandler:
log_handler_output_stream, metadata =
fth._read(ti=local_log_file_read, try_number=1)
mock_read_local.assert_called_with(path)
assert extract_events(log_handler_output_stream) == ["the log"]
- assert metadata == {"end_of_log": True, "log_pos": 1}
+ assert metadata == {"end_of_log": True}
@patch("airflow.utils.log.file_task_handler.FileTaskHandler._read_from_local")
def test__read_when_local_respects_log_pos_metadata(self, mock_read_local,
create_task_instance):
@@ -532,10 +548,74 @@ class TestFileTaskLogHandler:
# Should resume from the third line only.
assert extract_events(log_handler_output_stream) == ["line 3"]
- assert metadata == {"end_of_log": True, "log_pos": 3}
+ assert metadata == {"end_of_log": True}
+
+ @pytest.mark.parametrize(
+ ("ti_state", "ti_try_number", "expected_metadata"),
+ [
+ pytest.param(
+ TaskInstanceState.RUNNING,
+ 1,
+ {"end_of_log": False, "log_pos": 3},
+ id="running-current-try",
+ ),
+ pytest.param(
+ TaskInstanceState.DEFERRED,
+ 1,
+ {"end_of_log": False, "log_pos": 3},
+ id="deferred-current-try",
+ ),
+ pytest.param(TaskInstanceState.RUNNING, 2, {"end_of_log": True},
id="running-older-try"),
+ pytest.param(TaskInstanceState.SUCCESS, 1, {"end_of_log": True},
id="success"),
+ pytest.param(TaskInstanceState.FAILED, 1, {"end_of_log": True},
id="failed"),
+ ],
+ )
+ def test__read_accumulates_only_unfinished_reads(
+ self, ti_state, ti_try_number, expected_metadata, create_task_instance
+ ):
+ ti = create_task_instance(
+ dag_id="dag_for_testing_local_log_read",
+ task_id="task_for_testing_local_log_read",
+ run_type=DagRunType.SCHEDULED,
+ logical_date=DEFAULT_DATE,
+ state=ti_state,
+ )
+ ti.try_number = ti_try_number
+ source, pulled = _tracking_stream(["line 1", "line 2", "line 3"])
+ fth = FileTaskHandler("")
+
+ with (
+ patch.object(fth, "_read_from_local", return_value=(["the
messages"], [source])),
+ patch.object(fth, "_read_from_logs_server", return_value=([], [])),
+ patch.object(fth, "_get_executor") as get_executor,
+ patch(
+ "airflow.utils.log.file_task_handler.LogStreamAccumulator",
wraps=LogStreamAccumulator
+ ) as accumulator,
+ ):
+ get_executor.return_value.get_streaming_task_log.return_value =
([], [])
+ log_handler_output_stream, metadata = fth._read(ti=ti,
try_number=1)
+
+ accumulated = not expected_metadata["end_of_log"]
+ assert len(pulled) == (3 if accumulated else 0)
+ assert accumulator.called is accumulated
+ assert extract_events(log_handler_output_stream) == ["line 1",
"line 2", "line 3"]
+ assert metadata == expected_metadata
+ @pytest.mark.parametrize(
+ ("ti_state", "expected_metadata"),
+ [
+ pytest.param(TaskInstanceState.SUCCESS, {"end_of_log": True},
id="finished"),
+ pytest.param(
+ TaskInstanceState.RUNNING,
+ {"end_of_log": False, "log_pos": 3},
+ id="running-follow-up-poll",
+ ),
+ ],
+ )
@patch("airflow.utils.log.file_task_handler.FileTaskHandler._read_from_local")
- def test_read_respects_log_pos_metadata(self, mock_read_local,
create_task_instance):
+ def test_read_respects_log_pos_metadata(
+ self, mock_read_local, ti_state, expected_metadata,
create_task_instance
+ ):
"""The public `read()` wrapper must accept the `islice` stream
`_read()` returns for log_pos reads."""
mock_read_local.return_value = (
["the messages"],
@@ -546,18 +626,25 @@ class TestFileTaskLogHandler:
task_id="task_for_testing_local_log_read",
run_type=DagRunType.SCHEDULED,
logical_date=DEFAULT_DATE,
+ state=ti_state,
)
+ local_log_file_read.try_number = 1
fth = FileTaskHandler("")
- log_handler_output_stream, metadata = fth.read(
- local_log_file_read,
- try_number=1,
- metadata={"log_pos": 2},
- )
+ with (
+ patch.object(fth, "_read_from_logs_server", return_value=([], [])),
+ patch.object(fth, "_get_executor") as get_executor,
+ ):
+ get_executor.return_value.get_streaming_task_log.return_value =
([], [])
+ log_handler_output_stream, metadata = fth.read(
+ local_log_file_read,
+ try_number=1,
+ metadata={"log_pos": 2},
+ )
# Should resume from the third line only.
assert extract_events(log_handler_output_stream) == ["line 3"]
- assert metadata == {"end_of_log": True, "log_pos": 3}
+ assert metadata == expected_metadata
def test__read_from_local(self, tmp_path):
"""Tests the behavior of method _read_from_local"""
@@ -652,7 +739,7 @@ class TestFileTaskLogHandler:
else:
fth._read_from_logs_server.assert_not_called()
assert extract_events(logs, False) == expected_logs
- assert metadata == {"end_of_log": True, "log_pos": 3}
+ assert metadata == {"end_of_log": True}
@pytest.mark.parametrize("is_historical", [False, True])
def test_read_served_logs(self, is_historical, create_task_instance,
session):
diff --git a/providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py
b/providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py
index 06144952072..85ef2325e64 100644
--- a/providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py
+++ b/providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py
@@ -467,7 +467,7 @@ class TestS3TaskHandler:
assert log[5].event == "Line 2"
assert log[6].event == "Log line 3"
assert log[7].event == "Line 4"
- assert metadata == {"end_of_log": True, "log_pos": 4}
+ assert metadata["end_of_log"] is True
elif AIRFLOW_V_3_0_PLUS:
log = list(log)
assert log[0].event == "::group::Log message source details"
@@ -492,7 +492,7 @@ class TestS3TaskHandler:
if AIRFLOW_V_3_0_PLUS:
log = list(log)
assert len(log) == 2
- assert metadata == {"end_of_log": True, "log_pos": 0}
+ assert metadata["end_of_log"] is True
else:
assert len(log) == 1
assert len(log) == len(metadata)
diff --git
a/providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py
b/providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py
index 9cbbc19b74b..78fc5f809a0 100644
--- a/providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py
+++ b/providers/google/tests/unit/google/cloud/log/test_gcs_task_handler.py
@@ -525,7 +525,7 @@ class TestGCSTaskHandler:
assert logs[1].event == expected_gs_uri
assert logs[2].event == "::endgroup::"
assert logs[3].event == "CONTENT"
- assert metadata == {"end_of_log": True, "log_pos": 1}
+ assert metadata["end_of_log"] is True
elif AIRFLOW_V_3_0_PLUS:
logs = list(logs)
assert logs[0].event == "::group::Log message source details"
@@ -563,7 +563,7 @@ class TestGCSTaskHandler:
assert log[1].event == expected_gs_uri
assert log[2].event == f"{self.gcs_task_handler.local_base}/1.log"
assert log[3].event == "::endgroup::"
- assert metadata == {"end_of_log": True, "log_pos": 0}
+ assert metadata["end_of_log"] is True
elif AIRFLOW_V_3_0_PLUS:
log = list(log)
assert log[0].event == "::group::Log message source details"
diff --git
a/providers/microsoft/azure/tests/unit/microsoft/azure/log/test_wasb_task_handler.py
b/providers/microsoft/azure/tests/unit/microsoft/azure/log/test_wasb_task_handler.py
index f4834e2612b..d4c9eab9272 100644
---
a/providers/microsoft/azure/tests/unit/microsoft/azure/log/test_wasb_task_handler.py
+++
b/providers/microsoft/azure/tests/unit/microsoft/azure/log/test_wasb_task_handler.py
@@ -210,7 +210,7 @@ class TestWasbTaskHandler:
)
assert logs[2].event == "::endgroup::"
assert logs[3].event == "Log line"
- assert metadata == {"end_of_log": True, "log_pos": 1}
+ assert metadata["end_of_log"] is True
elif AIRFLOW_V_3_0_PLUS:
logs = list(logs)
assert logs[0].event == "::group::Log message source details"
