jason810496 commented on code in PR #70671:
URL: https://github.com/apache/airflow/pull/70671#discussion_r4408883072
##########
airflow-core/src/airflow/utils/log/file_task_handler.py:
##########
@@ -647,21 +645,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
+ # lines drains the stream eagerly, so finished reads stream
straight through.
+ with LogStreamAccumulator(out_stream, HEAP_DUMP_SIZE) as
stream_accumulator:
+ out_metadata["log_pos"] = stream_accumulator.total_lines
+ out_stream = stream_accumulator.stream
Review Comment:
Agreed, no more inference. I updated issue #75211 to state the mechanism:
the caller sets an explicit flag in metadata, and only the JSON branch of
get_log asks for the count.
##########
airflow-core/tests/unit/utils/test_log_handlers.py:
##########
@@ -516,18 +597,26 @@ def test_read_respects_log_pos_metadata(self,
mock_read_local, create_task_insta
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}
+ # log_pos counts every record read so far, not only the ones returned
in this call.
+ assert metadata == expected_metadata
Review Comment:
Dropped the note in 4c91c71a0dc.
##########
airflow-core/tests/unit/utils/test_log_handlers.py:
##########
@@ -502,10 +518,75 @@ 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", "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"]
+ # An unfinished read drains the source up front to count lines; a
finished read stays lazy.
Review Comment:
Dropped the comment in 28a2b9c2ab5.
##########
providers/amazon/tests/unit/amazon/aws/log/test_s3_task_handler.py:
##########
@@ -492,7 +499,10 @@ def test_read_when_s3_log_missing(self):
if AIRFLOW_V_3_0_PLUS:
log = list(log)
assert len(log) == 2
- assert metadata == {"end_of_log": True, "log_pos": 0}
+ if AIRFLOW_V_3_4_PLUS:
+ assert metadata == {"end_of_log": True}
+ else:
+ assert metadata == {"end_of_log": True, "log_pos": 0}
Review Comment:
Addressed in 589d2015687, these tests now assert only end_of_log and the
version gate is gone.
--
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]
