kaxil commented on code in PR #70671:
URL: https://github.com/apache/airflow/pull/70671#discussion_r4323352982
##########
airflow-core/tests/unit/utils/test_log_handlers.py:
##########
@@ -502,7 +502,34 @@ def test__read_when_local_respects_log_pos_metadata(self,
mock_read_local, creat
# 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", [TaskInstanceState.SUCCESS,
TaskInstanceState.FAILED])
Review Comment:
`create_task_instance` leaves `ti.try_number` at 0, so `_read(ti=ti,
try_number=1)` gets `end_of_log=True` from the `ti.try_number != try_number`
half before the state is looked at. SUCCESS and FAILED run the same path here,
and a RUNNING TI would skip the accumulator too. With `ti.try_number = 1` the
state starts to matter, and then a RUNNING case on the current try (accumulator
called, `log_pos` returned) next to a RUNNING TI read at an older try
(accumulator not called) would cover both halves of the `end_of_log` check.
`TestFileTaskHandlerExecutorLogs._running_ti` in `test_file_task_handler.py`
already stubs the executor branch for a running TI.
##########
airflow-core/src/airflow/utils/log/file_task_handler.py:
##########
@@ -647,21 +647,22 @@ def _read(
TaskInstanceState.DEFERRED,
)
- with LogStreamAccumulator(out_stream, HEAP_DUMP_SIZE) as
stream_accumulator:
- 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)
+ out_metadata: LogMetadata = {"end_of_log": end_of_log}
+ if not end_of_log:
Review Comment:
On the NDJSON path, which is the UI's default `accept` in `useLogs.tsx`, the
`log_pos` computed here is never used either: `get_log` builds the
`Airflow-Continuation-Token` header from the inbound metadata before
`read_log_stream` runs, and `read_log_stream` pops any inbound `log_pos`. So a
running task open in the UI still drains its whole log through the accumulator
on every refresh. Would it work to have the caller say whether it needs
`log_pos` (only the JSON branch of `get_log` turns it into a token) instead of
inferring it from `end_of_log`? Fine as a follow-up, since this PR is scoped to
finished reads.
##########
airflow-core/tests/unit/utils/test_log_handlers.py:
##########
@@ -502,7 +502,34 @@ def test__read_when_local_respects_log_pos_metadata(self,
mock_read_local, creat
# 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}
Review Comment:
This test and `test_read_respects_log_pos_metadata` are the only ones that
send an inbound `log_pos`, and before this change both went through the
accumulator and then `islice`. Now both take the finished path, so a running
task's follow-up poll (accumulate, then skip N) has no test left. Could one of
them be parametrized with a running TI on the current try, expecting `["line
3"]` and `{"end_of_log": False, "log_pos": 3}`?
##########
airflow-core/src/airflow/utils/log/file_task_handler.py:
##########
@@ -647,21 +647,22 @@ def _read(
TaskInstanceState.DEFERRED,
)
- with LogStreamAccumulator(out_stream, HEAP_DUMP_SIZE) as
stream_accumulator:
- 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)
+ out_metadata: LogMetadata = {"end_of_log": end_of_log}
+ if not end_of_log:
+ # Only an unfinished read needs log_pos for the continuation
token, and counting
Review Comment:
The `:return:` section of the docstring above still lists `log_pos` as
always present and calls it a char position. After this change it's only
returned when `end_of_log` is False, and it's a count of log records. Could
that be updated here? The providers that call `super()._read` read it as the
contract.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
