You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "jason810496 (via GitHub)" <gi...@apache.org> on 2025/08/22 01:58:21 UTC
[PR] Add stream method to RemoteIO [airflow]
jason810496 opened a new pull request, #54813:
URL: https://github.com/apache/airflow/pull/54813
related: [Fix CloudwatchTaskHandler display error #54054](https://github.com/apache/airflow/pull/54054)
## Why
While fixing comments in #54054, I realized that we should support `stream` method in `RemoteIO` as https://github.com/apache/airflow/pull/54054#pullrequestreview-3083573995 point out. Starting with Airflow 3, [FileTaskHandler._read_remote_logs](https://github.com/apache/airflow/blob/f7f32544726672ada2766fb76efcabfba4191cb8/airflow-core/src/airflow/utils/log/file_task_handler.py#L908-L909) will utilize `RemoteIO` behind the scenes, rather than relying on `<Provider>TaskHandler._read_remote_logs`.
## What
Add `stream` method to `RemoteIO` Protocol and add corresponding unit tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2329835039
##########
airflow-core/src/airflow/utils/log/file_task_handler.py:
##########
@@ -73,7 +73,7 @@
"""Information _about_ the log fetching process for display to a user"""
RawLogStream: TypeAlias = Generator[str, None, None]
"""Raw log stream, containing unparsed log lines."""
-LegacyLogResponse: TypeAlias = tuple[LogSourceInfo, LogMessages]
+LegacyLogResponse: TypeAlias = tuple[LogSourceInfo, LogMessages | None]
Review Comment:
Agree! I just clarified the type naming by renaming
- `LogResponse` to `StreamingLogResponse`
- `LegacyLogResponse` to `LogResponse`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2321856179
##########
providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py:
##########
@@ -95,6 +95,9 @@ def read(self, relative_path, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages
return messages, logs
return messages, None
+ def stream(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
+ raise NotImplementedError
Review Comment:
Given we now do `getattr(remote_io, "stream")` we don't want this method here anymore do we, as we don't want this to be called (else it would raise NotImplementedError at runtime instead of falling back to the read method)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 merged PR #54813:
URL: https://github.com/apache/airflow/pull/54813
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #54813:
URL: https://github.com/apache/airflow/pull/54813#issuecomment-3252979721
re-triggering tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2348945020
##########
airflow-core/src/airflow/logging/remote.py:
##########
@@ -44,6 +44,15 @@ def upload(self, path: os.PathLike | str, ti: RuntimeTI) -> None:
"""Upload the given log path to the remote storage."""
...
- def read(self, relative_path: str, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages | None]:
+ def read(self, relative_path: str, ti: RuntimeTI) -> LegacyLogResponse:
"""Read logs from the given remote log path."""
...
+
+
+@runtime_checkable
+class RemoteLogStreamIO(Protocol):
Review Comment:
Sorry missed this message.
I think it depends on what the Mypy error message was?
Maybe it needs to be `RemoteLogStreamIO(RemoteLogIO, Protocol):` to keep mypy happy?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2329835716
##########
providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py:
##########
@@ -176,6 +176,9 @@ def read(self, relative_path, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages
return messages, logs
+ def stream(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
+ raise NotImplementedError
+
Review Comment:
Sure, I just removed all of them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2320887347
##########
providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py:
##########
@@ -95,6 +95,9 @@ def read(self, relative_path, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages
return messages, logs
return messages, None
+ def stream(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
+ raise NotImplementedError
Review Comment:
I just added another `RemoteLogStreamIO` Protocal and having `REMOTE_TASK_LOG` annotated as `REMOTE_TASK_LOG: RemoteLogIO | RemoteLogStreamIO | None = None`.
Additionally, mypy is not waring and the CI have passed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2321335974
##########
providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py:
##########
@@ -95,6 +95,9 @@ def read(self, relative_path, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages
return messages, logs
return messages, None
+ def stream(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
+ raise NotImplementedError
Review Comment:
Yes, and the CI also passes after I mark it for review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2321856768
##########
providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py:
##########
@@ -176,6 +176,9 @@ def read(self, relative_path, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages
return messages, logs
+ def stream(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
+ raise NotImplementedError
+
Review Comment:
Ditto here. Remove this etc.
##########
providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py:
##########
@@ -95,6 +95,9 @@ def read(self, relative_path, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages
return messages, logs
return messages, None
+ def stream(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
+ raise NotImplementedError
Review Comment:
Given we now do `getattr(remote_io, "stream")` we don't want this method here anymore do we, as we don't want this to be called.
##########
airflow-core/src/airflow/logging/remote.py:
##########
@@ -44,6 +44,15 @@ def upload(self, path: os.PathLike | str, ti: RuntimeTI) -> None:
"""Upload the given log path to the remote storage."""
...
- def read(self, relative_path: str, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages | None]:
+ def read(self, relative_path: str, ti: RuntimeTI) -> LegacyLogResponse:
"""Read logs from the given remote log path."""
...
+
+
+@runtime_checkable
+class RemoteLogStreamIO(Protocol):
Review Comment:
I think this protocol should be a extension to the non-streaming interface, so:
```suggestion
class RemoteLogStreamIO(RemoteLogIO):
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2321850410
##########
airflow-core/src/airflow/utils/log/file_task_handler.py:
##########
@@ -73,7 +73,7 @@
"""Information _about_ the log fetching process for display to a user"""
RawLogStream: TypeAlias = Generator[str, None, None]
"""Raw log stream, containing unparsed log lines."""
-LegacyLogResponse: TypeAlias = tuple[LogSourceInfo, LogMessages]
+LegacyLogResponse: TypeAlias = tuple[LogSourceInfo, LogMessages | None]
Review Comment:
Rather than "Legacy" I think we should call this something like `LogResponse` and `StreamingLogResponse` -- as unless I'm missing something there could always be log providers that don't implement the stream method?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2349528096
##########
airflow-core/src/airflow/logging/remote.py:
##########
@@ -44,6 +44,15 @@ def upload(self, path: os.PathLike | str, ti: RuntimeTI) -> None:
"""Upload the given log path to the remote storage."""
...
- def read(self, relative_path: str, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages | None]:
+ def read(self, relative_path: str, ti: RuntimeTI) -> LegacyLogResponse:
"""Read logs from the given remote log path."""
...
+
+
+@runtime_checkable
+class RemoteLogStreamIO(Protocol):
Review Comment:
Thanks for the hint! It works well with inheriting both `RemoteLogIO` and `Protocol`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2329847364
##########
airflow-core/src/airflow/logging/remote.py:
##########
@@ -44,6 +44,15 @@ def upload(self, path: os.PathLike | str, ti: RuntimeTI) -> None:
"""Upload the given log path to the remote storage."""
...
- def read(self, relative_path: str, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages | None]:
+ def read(self, relative_path: str, ti: RuntimeTI) -> LegacyLogResponse:
"""Read logs from the given remote log path."""
...
+
+
+@runtime_checkable
+class RemoteLogStreamIO(Protocol):
Review Comment:
Either option works for me, but I’ll wait for your suggestion, Ash.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2321332427
##########
providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py:
##########
@@ -95,6 +95,9 @@ def read(self, relative_path, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages
return messages, logs
return messages, None
+ def stream(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
+ raise NotImplementedError
Review Comment:
Ready for re-review then?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2304157218
##########
providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py:
##########
@@ -95,6 +95,9 @@ def read(self, relative_path, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages
return messages, logs
return messages, None
+ def stream(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
+ raise NotImplementedError
Review Comment:
> Hmmm, it's a bit annoying that we need to define this in all the subclasses (as otherwise it fails the protocol check).
So do I
> I wonder if there's a different option we could take here 🤔
I had though about another approach by replacing the `RemoteLogIO` Protocal with ABC, then the `stream` method will raise `NotImplementedError` by default. No sure if there is any concern of using ABC instead of Protocal ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2304243968
##########
providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py:
##########
@@ -95,6 +95,9 @@ def read(self, relative_path, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages
return messages, logs
return messages, None
+ def stream(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
+ raise NotImplementedError
Review Comment:
Other option would be to have two protocols (one with just read, and one which subclasses it that adds in stream), and then either mark it as "runtime checkable", or simply do something like this:
```python
logs: LogMessages | list[RawLogStream] | None # extra typing to void mypy assignment error
if stream := getattr(remote_io, "stream"):
# Use .stream interface if provider's RemoteIO supports it
sources, logs = stream(path, ti)
return sources, logs or []
else:
# Fallback to .read interface
sources, logs = remote_io.read(path, ti)
return sources, logs or []
```
(This also has the benefit of not masking potential exceptions from inside the stream function itself -- for instance if the stream function was implemented but raised an AttributeError inside it then it would be lost and the read fallback path chosen instead)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb closed pull request #54813: Add stream method to RemoteIO
URL: https://github.com/apache/airflow/pull/54813
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2323921729
##########
airflow-core/src/airflow/logging/remote.py:
##########
@@ -44,6 +44,15 @@ def upload(self, path: os.PathLike | str, ti: RuntimeTI) -> None:
"""Upload the given log path to the remote storage."""
...
- def read(self, relative_path: str, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages | None]:
+ def read(self, relative_path: str, ti: RuntimeTI) -> LegacyLogResponse:
"""Read logs from the given remote log path."""
...
+
+
+@runtime_checkable
+class RemoteLogStreamIO(Protocol):
Review Comment:
I had tried using subclass of `RemoteLogIO` before, but it raise some Mypy error.
If then, do we still want to make it subclass here and add the ignore mypy comment?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #54813:
URL: https://github.com/apache/airflow/pull/54813#issuecomment-3265676225
I added `backport-to-v3-0-test` because, regardless of whether Airflow Core is `3.0.x` or `3.1.x`.
If the provider implements the `.stream` method, the core should use it and fallback to `.read` if not implement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2304121922
##########
providers/alibaba/src/airflow/providers/alibaba/cloud/log/oss_task_handler.py:
##########
@@ -95,6 +95,9 @@ def read(self, relative_path, ti: RuntimeTI) -> tuple[LogSourceInfo, LogMessages
return messages, logs
return messages, None
+ def stream(self, relative_path: str, ti: RuntimeTI) -> LogResponse:
+ raise NotImplementedError
Review Comment:
Hmmm, it's a bit annoying that we need to define this in all the subclasses (as otherwise it fails the protocol check).
I wonder if there's a different option we could take here 🤔
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add stream method to RemoteIO [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #54813:
URL: https://github.com/apache/airflow/pull/54813#discussion_r2485906433
##########
airflow-core/src/airflow/utils/log/file_task_handler.py:
##########
@@ -936,5 +935,10 @@ def _read_remote_logs(self, ti, try_number, metadata=None) -> LegacyLogResponse
# This living here is not really a good plan, but it just about works for now.
# Ideally we move all the read+combine logic in to TaskLogReader and out of the task handler.
path = self._render_filename(ti, try_number)
+ if hasattr(remote_io, "stream") and (stream_method := getattr(remote_io, "stream")):
Review Comment:
```suggestion
if (stream_method := getattr(remote_io, "stream", None)):
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org