You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "jedcunningham (via GitHub)" <gi...@apache.org> on 2025/05/17 03:41:30 UTC
[PR] Stop streaming task logs if end of log mark is missing [airflow]
jedcunningham opened a new pull request, #50715:
URL: https://github.com/apache/airflow/pull/50715
Sometimes, somehow, the end of log mark can be missing, and when that happens the streaming log reader enters an infinite loop. Instead, if the task is in a non-running state and we stop receiving log lines but never get the end of log mark, we assume we won't and stop trying. We do tell emit that we are stopping though.
(Leaving this as draft until it can be properly tested)
Closes #39323
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2101658101
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,15 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if not logs:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
Review Comment:
I think this scenario occurs when the current `TaskInstance` is still running, but it's a long-running task.
1. The current `end_of_log` will still be `true` if the `TaskInstance` is still `running` or `deferred`:
https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/utils/log/file_task_handler.py#L446C9-L449
2. `FileTaskHandler` will skip the same logs (based on `log_pos` in metadata), so if the `logs` returned from `read_log_chunks` remain empty for several iterations, we can infer that the `TaskInstance` is still running but not producing new logs (still polling or computing...).
https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/utils/log/file_task_handler.py#L451-L452
This behavior causes the following error:
https://github.com/apache/airflow/issues/39323#issuecomment-2850639129
Therefore, we need to check for multiple consecutive empty iterations.
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2127081579
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,17 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if logs:
+ empty_iterations = 0
+ else:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
+ if empty_iterations >= self.STREAM_LOOP_STOP_AFTER_EMPTY_ITERATIONS:
+ # we have not received any logs for a while, so we stop the stream
+ yield "(Log stream stopped - End of log marker not found, so logs may be incomplete)\n"
Review Comment:
you don't actually check if there is an "end of log marker" here. and, not all task log handlers even use them. so the messaging seems weird.
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2127156044
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,17 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if logs:
+ empty_iterations = 0
+ else:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
+ if empty_iterations >= self.STREAM_LOOP_STOP_AFTER_EMPTY_ITERATIONS:
+ # we have not received any logs for a while, so we stop the stream
+ yield "(Log stream stopped - End of log marker not found, so logs may be incomplete)\n"
Review Comment:
```suggestion
yield "Log stream stopped but no end of log marker not found; logs may be incomplete."
```
Is it supposed to include a newline as you have done?
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2103906545
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,17 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
Review Comment:
**The `ti` object passed into `read_log_stream` will always reflect the same state**, because **SQLAlchemy doesn’t automatically reflect changes made to the underlying database by other processes**.
Based on the previous scenario (see: https://github.com/apache/airflow/pull/50715/files#r2101658101 ), if the `ti` is in the `RUNNING` state when it is initially passed into `read_log_stream`, it will remain `RUNNING` even if another process has updated its state.
We should consider adding `session.refresh(ti)` to ensure we get the latest (and correct) state of the `ti` object, especially in cases where it may have transitioned to `SUCCESS` or `FAILED` after several calls to `read_log_chunks`.
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2103350513
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,15 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if not logs:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
Review Comment:
We will never hit this block if the task running or deferred. Honestly, bailing after multiple empty iterations was just me being a bit defensive - bailing after the first instance would probably be fine too.
This is hit when we will never get the end of log mark - imagine that line is simply missing for a finished task from ES. You can look and look and look, but you will never get another line. In theory, each log handler could handle this scenario gracefully, but having something generic here as a fail safe felt like a good idea.
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2127156441
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,17 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if logs:
+ empty_iterations = 0
+ else:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
+ if empty_iterations >= self.STREAM_LOOP_STOP_AFTER_EMPTY_ITERATIONS:
+ # we have not received any logs for a while, so we stop the stream
+ yield "(Log stream stopped - End of log marker not found, so logs may be incomplete)\n"
Review Comment:
(seems odd if so)
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2128954683
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,17 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
Review Comment:
I see, I overlooked the _not_ in `ti.state not in` statement, thanks for clarifying.
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2101063623
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,15 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if not logs:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
Review Comment:
Maybe something like the after at line 97:
```
if logs:
empty_iterations = 0
```
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2127093493
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,17 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if logs:
+ empty_iterations = 0
+ else:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
+ if empty_iterations >= self.STREAM_LOOP_STOP_AFTER_EMPTY_ITERATIONS:
+ # we have not received any logs for a while, so we stop the stream
+ yield "(Log stream stopped - End of log marker not found, so logs may be incomplete)\n"
+ break
Review Comment:
might as well return since, there's nothing after the loop. not sure why the other one is a break. it's just more direct and clear since that's what the effect is anyway.
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2103381601
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,15 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if not logs:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
Review Comment:
I'm not super concerned with accumulating empties, but we can humor copilot :)
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2114615500
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,17 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
Review Comment:
I think it's the opposite actually - if its running, it hits the else block here, then breaks out of the while loop. If the ti is running we _always_ break out and do not do another iteration.
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2101061394
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,15 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if not logs:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
Review Comment:
Right now, it accumulates empties across the entire stream, even if some non-empty batches slip in. For example, if it ever gets “hello”, then 2 empties, then another log line, you’ll still stop after the 5th empty overall. It might be good to consider resetting `empty_iterations = 0` whenever it actually receives a non-empty batch, to make it a counter of consecutive empty polls.
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,15 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if not logs:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
Review Comment:
Maybe something like the following at line 97:
```
if logs:
empty_iterations = 0
```
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #50715:
URL: https://github.com/apache/airflow/pull/50715#issuecomment-2949534342
### Backport successfully created: v3-0-test
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>✅</td>
<td>v3-0-test</td>
<td><a href="https://github.com/apache/airflow/pull/51482"><img src="https://img.shields.io/badge/PR-51482-blue" alt="PR Link"></a></td>
</tr>
</table>
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2101350746
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,15 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if not logs:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
Review Comment:
Why do we want to check for multiple empty iterations? Is it to add resilience to potential network hiccups?
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2127093493
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,17 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if logs:
+ empty_iterations = 0
+ else:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
+ if empty_iterations >= self.STREAM_LOOP_STOP_AFTER_EMPTY_ITERATIONS:
+ # we have not received any logs for a while, so we stop the stream
+ yield "(Log stream stopped - End of log marker not found, so logs may be incomplete)\n"
+ break
Review Comment:
might as well return since, there's nothing after the loop. not sure why the other one is a break.
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham merged PR #50715:
URL: https://github.com/apache/airflow/pull/50715
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2127592177
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,17 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if logs:
+ empty_iterations = 0
+ else:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
+ if empty_iterations >= self.STREAM_LOOP_STOP_AFTER_EMPTY_ITERATIONS:
+ # we have not received any logs for a while, so we stop the stream
+ yield "(Log stream stopped - End of log marker not found, so logs may be incomplete)\n"
+ break
Review Comment:
Done.
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,17 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if logs:
+ empty_iterations = 0
+ else:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
+ if empty_iterations >= self.STREAM_LOOP_STOP_AFTER_EMPTY_ITERATIONS:
+ # we have not received any logs for a while, so we stop the stream
+ yield "(Log stream stopped - End of log marker not found, so logs may be incomplete)\n"
Review Comment:
We add a newline onto each other line, including the last one, so kept it that way for consistency.
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2127608403
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,17 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if logs:
+ empty_iterations = 0
+ else:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
+ if empty_iterations >= self.STREAM_LOOP_STOP_AFTER_EMPTY_ITERATIONS:
+ # we have not received any logs for a while, so we stop the stream
+ yield "(Log stream stopped - End of log marker not found, so logs may be incomplete)\n"
Review Comment:
I didn't take your message changes as-is, but tried to find a decent middle ground. Take a peek.
--
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] Stop streaming task logs if end of log mark is missing [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50715:
URL: https://github.com/apache/airflow/pull/50715#discussion_r2127081579
##########
airflow-core/src/airflow/utils/log/log_reader.py:
##########
@@ -96,10 +99,17 @@ def read_log_stream(self, ti: TaskInstance, try_number: int | None, metadata: di
TaskInstanceState.RUNNING,
TaskInstanceState.DEFERRED,
):
- if not logs[0]:
+ if logs:
+ empty_iterations = 0
+ else:
# we did not receive any logs in this loop
# sleeping to conserve resources / limit requests on external services
time.sleep(self.STREAM_LOOP_SLEEP_SECONDS)
+ empty_iterations += 1
+ if empty_iterations >= self.STREAM_LOOP_STOP_AFTER_EMPTY_ITERATIONS:
+ # we have not received any logs for a while, so we stop the stream
+ yield "(Log stream stopped - End of log marker not found, so logs may be incomplete)\n"
Review Comment:
not all task log handlers even use them. so the messaging seems weird.
--
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