This is an automated email from the ASF dual-hosted git repository.
Miretpl 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 7ae9d588dc9 Fix duplicated task logs in the WASB log handler (#70860)
7ae9d588dc9 is described below
commit 7ae9d588dc9ba346692946d884149bc1b65844df
Author: Manan Bhatt <[email protected]>
AuthorDate: Sat Aug 1 05:01:15 2026 +0530
Fix duplicated task logs in the WASB log handler (#70860)
Several task-execution lifecycles can write to one remote log key. With a
reschedule-mode sensor, try_number does not change, so every poke uploads to
the same attempt log. The local log file is opened in append mode and is
never
truncated after a successful upload, so a lifecycle that reuses the same
local
path re-sends the lines the blob already holds and the stored log grows
quadratically. This depends on the local path being reused, which happens
on a
long-lived worker or a shared logs volume, and not when every lifecycle
starts
with a fresh one.
The S3 handler had the same defect, reported as a worker OOM and fixed in
#67144. Emptying the local file on a successful upload is intentionally the
same approach, so the two handlers stay consistent.
---
.../microsoft/azure/log/wasb_task_handler.py | 6 +-
.../microsoft/azure/log/test_wasb_task_handler.py | 70 ++++++++++++++++++++++
2 files changed, 75 insertions(+), 1 deletion(-)
diff --git
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/log/wasb_task_handler.py
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/log/wasb_task_handler.py
index 32cd3ee374e..5c631f4d612 100644
---
a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/log/wasb_task_handler.py
+++
b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/log/wasb_task_handler.py
@@ -66,6 +66,8 @@ class WasbRemoteLogIO(LoggingMixin): # noqa: D101
has_uploaded = self.write(log, remote_loc)
if has_uploaded and self.delete_local_copy:
shutil.rmtree(os.path.dirname(local_loc))
+ elif has_uploaded:
+ local_loc.write_text("")
@cached_property
def hook(self):
@@ -193,7 +195,9 @@ class WasbRemoteLogIO(LoggingMixin): # noqa: D101
"""
if append and self.wasb_log_exists(remote_log_location):
old_log = self.wasb_read(remote_log_location)
- log = f"{old_log}\n{log}" if old_log else log
+ if old_log:
+ sep = "" if old_log.endswith("\n") else "\n"
+ log = f"{old_log}{sep}{log}"
try:
self.hook.load_string(log, self.wasb_container,
remote_log_location, overwrite=True)
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 57d21664c13..349ca250cbd 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
@@ -255,6 +255,76 @@ class TestWasbTaskHandler:
overwrite=True,
)
+ @mock.patch("airflow.providers.microsoft.azure.hooks.wasb.WasbHook")
+ @mock.patch.object(WasbRemoteLogIO, "wasb_read")
+ @mock.patch.object(WasbRemoteLogIO, "wasb_log_exists")
+ def test_write_on_existing_log_already_newline_terminated(
+ self, mock_log_exists, mock_wasb_read, mock_hook
+ ):
+ mock_log_exists.return_value = True
+ mock_wasb_read.return_value = "old log\n"
+ self.wasb_task_handler.io.write("text", self.remote_log_location)
+ mock_hook.return_value.load_string.assert_called_once_with(
+ "old log\ntext",
+ self.container_name,
+ self.remote_log_location,
+ overwrite=True,
+ )
+
+ def test_upload_repeated_appends_no_duplication(self, tmp_path):
+ """Each execution lifecycle of one attempt appends to the same local
log and uploads it."""
+ blobs: dict[str, str] = {}
+
+ class FakeHook:
+ def check_for_blob(self, container, blob_name, **kwargs):
+ return blob_name in blobs
+
+ def read_file(self, container, blob_name, **kwargs):
+ return blobs.get(blob_name, "")
+
+ def load_string(self, string_data, container, blob_name, **kwargs):
+ blobs[blob_name] = string_data
+
+ io = WasbRemoteLogIO(
+ remote_base="remote/log/location",
+ base_log_folder=str(tmp_path),
+ delete_local_copy=False,
+ wasb_container=self.container_name,
+ )
+ io.hook = FakeHook()
+
+ local_log = tmp_path / "attempt=1.log"
+ for cycle in range(1, 4):
+ with open(local_log, "a") as f:
+ f.write(f"cycle {cycle}\n")
+ io.upload("attempt=1.log")
+
+ assert blobs["remote/log/location/attempt=1.log"] == "cycle 1\ncycle
2\ncycle 3\n"
+ assert local_log.read_text() == ""
+
+ @pytest.mark.parametrize(
+ ("has_uploaded", "expected_local_content"),
+ [(True, ""), (False, "some log\n")],
+ )
+ @mock.patch.object(WasbRemoteLogIO, "write")
+ def test_upload_truncates_local_log_only_after_successful_write(
+ self, mock_write, tmp_path, has_uploaded, expected_local_content
+ ):
+ """A failed upload must leave the local log intact so it can be
retried."""
+ mock_write.return_value = has_uploaded
+ io = WasbRemoteLogIO(
+ remote_base="remote/log/location",
+ base_log_folder=str(tmp_path),
+ delete_local_copy=False,
+ wasb_container=self.container_name,
+ )
+ local_log = tmp_path / "attempt=1.log"
+ local_log.write_text("some log\n")
+
+ io.upload("attempt=1.log")
+
+ assert local_log.read_text() == expected_local_content
+
@mock.patch("airflow.providers.microsoft.azure.hooks.wasb.WasbHook")
def test_write_when_append_is_false(self, mock_hook):
self.wasb_task_handler.io.write("text", self.remote_log_location,
False)
