[mungiyo]

What

Adds S3RemoteLogIO.stream() so that remote task logs stored in S3 are read as a stream of lines instead of being materialised as one string per object, and lets read() delegate to it.

Why

S3RemoteLogIO.read() loads every remote log object into a single string:

for key in keys:
    logs.append(self.s3_read(key, return_error=True))   # S3Hook.read_key() -> Body.read().decode()

Core then wraps that string in an io.StringIO (_get_compatible_log_stream), so opening a task log in the UI costs the api-server at least twice the size of the log, and a large enough log OOM-kills it.

That failure is worse than a failed log view. When the api-server runs with the default --apps all, the UI and the Task Execution API share a process, so the OOM kill also drops the endpoint that running tasks heartbeat to — viewing a log can take down tasks that had nothing to do with it.

Everything downstream of the provider is already bounded: LogStreamAccumulator spills to disk every HEAP_DUMP_SIZE lines and the K-way merge heap is capped by the same value. The provider read is the only unbounded step left.

Why now

Core has had the streaming path since 3.2.0 — FileTaskHandler._read_remote_logs prefers remote_io.stream() when the provider implements it and falls back to read() otherwise:

if stream_method := getattr(remote_io, "stream", None):
    sources, logs = stream_method(path, ti)
    return sources, logs or []
sources, logs = remote_io.read(path, ti)

gs (google) and cloudwatch (amazon) already implement stream(). S3 was the remaining legacy reader, so it is the one backend where this still costs the whole object in memory.

How

stream() follows the same shape as the GCS and CloudWatch implementations:

• It returns one generator per object. The GetObject is issued when the generator is first consumed, so listing the keys of a task instance no longer implies downloading them.
• _get_log_stream() iterates Body.iter_chunks() and splits on "\n" only, with the trailing newline stripped — the same records file_task_handler._stream_lines_by_chunk produces for local logs, so the log reader sees exactly what it saw before. An incremental UTF-8 decoder keeps multi-byte characters intact across chunk boundaries.
• A failure to read one object is yielded as a log line rather than raised, so one unreadable object does not hide the other objects of the same task instance. This preserves the previous s3_read(..., return_error=True) behaviour.
• read() re-joins the lines with their newline, so the string it returns is byte-identical to the object for callers that cannot consume a stream (Airflow < 3.2, and S3TaskHandler._read_remote_logs, which is left calling io.read() exactly like the GCS and CloudWatch handlers do).

Peak memory for a log read becomes a function of the chunk size instead of the log size.

Tests

providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py — 43 passed (8 new).

The new tests pin the two properties that matter:

• test_read_does_not_load_the_object_as_a_single_string fails if s3_read is ever called again from this path — a regression back to the whole-object read is caught by the suite, not by review.
• test_stream_does_not_fetch_the_object_until_it_is_consumed pins the laziness.

Plus: line splitting without the trailing newline, key ordering, multi-byte characters split across chunk boundaries (with CHUNK_SIZE patched to 3 bytes), the empty case, and the unreadable-object case.

Verification against this patch

Measured on breeze start-airflow --integration localstack --backend postgres from this branch (Airflow 3.4.0.dev0, the default S3 remote logging config, delete_local_logs = True so that the log can only come back from S3). A task printed 2,000,000 lines, each with multi-byte characters, into a 406.3 MiB object. The log was then read through TaskLogReader.read_log_stream, the call behind the application/x-ndjson log endpoint the UI uses. The baseline is the same checkout with only s3_task_handler.py swapped back to main, reading the same object.

	main	this PR
Method core entered	S3RemoteLogIO.read()	S3RemoteLogIO.stream()
Lines returned	2,000,000 / 2,000,000, in order, unmodified	identical
Process max RSS, start → end	266.1 → 2,655.9 MiB	266.5 → 369.8 MiB
Python heap peak (tracemalloc)	2,367.4 MiB	82.1 MiB¹
Wall time, without tracemalloc	65.4 s	57.1 s

¹ The tracemalloc run of this PR stopped at line 886,112. tracemalloc slows the consumer down enough that LocalStack resets the GetObject connection; a plain boto3 reader at 2 MiB/s is reset the same way after 79 s, while a full-speed read of the object takes 1 s. The read surfaced the reset as a Could not read logs from ... line instead of raising, as intended. Real S3 does not do this: reading a 1.4 GB public object at a steady 0.5 MiB/s ran for 10 minutes without a reset. The RSS row comes from complete runs of both versions.

In production

An equivalent implementation (same iter_chunks + incremental-decoder read path, installed via [logging] logging_config_class) has been running on Airflow 3.1.8 with S3 remote logging in our dev, staging and production clusters since 2026-09-10.

• Before, reading a 250 MB / 2,000,000-line log peaked at 1,168.7 MiB of Python heap; after, 29.5 MiB (tracemalloc, same input).
• In production, reading a real task log through TaskLogReader.read_log_stream streams 11,326 lines at a 61.9 MiB heap peak.
• No api-server OOM on log views since.

Notes

• Behaviour for existing configs is unchanged — no config, no key layout and no message formatting changes. The Airflow 2 message format (Found logs in s3: / * <key>) is preserved as-is.
• CHUNK_SIZE is defined in the provider rather than imported from core, so the provider keeps working against Airflow versions that do not expose it.
• For a finished task, core no longer drains the stream into LogStreamAccumulator (Skip log accumulator disk spill when reading finished task logs #70671), so the GetObject is read as fast as the client consumes the NDJSON response and stays open that long. Against real S3, a body read slowly but steadily was not reset (0.5 MiB/s for 10 minutes); one left unread for 400 s was (60 s and 180 s were fine). In that case the rest of that object is replaced by a Could not read logs from ... line. The numbers above were measured before Skip log accumulator disk spill when reading finished task logs #70671, when the accumulator still drained the stream.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[rjgoyln]

LGTM!

[o-nikolas]

Thanks, looks reasonable to me, but there are conflicts that need resolving and I just staretd the tests.

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.