aeroyorch commented on code in PR #73044:
URL: https://github.com/apache/airflow/pull/73044#discussion_r4246573969
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/pod_manager.py:
##########
@@ -1234,29 +1234,34 @@ async def fetch_container_logs_before_current_sec(
"""
Asynchronously read the log file of the specified pod.
- This method streams logs from the base container, skipping log lines
from the current second to prevent duplicate entries on subsequent reads. It is
designed to handle long-running containers and gracefully suppresses transient
interruptions.
+ Lines are consumed one at a time as they arrive, so a large backlog is
never held in memory.
+ Log lines from the current second are skipped to prevent duplicate
entries on the next read.
:param pod: The pod specification to monitor.
:param container_name: The name of the container within the pod.
:param since_time: The timestamp from which to start reading logs.
- :return: The timestamp to use for the next log read, representing the
start of the current second. Returns None if an exception occurred.
+ :return: The timestamp to use for the next log read, representing the
start of the current second.
"""
now = pendulum.now()
- logs = await self._hook.read_logs(
+ log_lines = self._hook.stream_logs(
name=pod.metadata.name,
namespace=pod.metadata.namespace,
container_name=container_name,
since_seconds=(math.ceil((now - since_time).total_seconds()) if
since_time else None),
)
- # CPU-bound per-line parse/emit, offloaded so it can't block the
triggerer event loop.
- await asyncio.to_thread(self._emit_container_logs, logs, now,
container_name)
+ # ``aclosing`` so the early break below finalises the generator, and
with it the HTTP
+ # response, deterministically rather than leaving it to the loop's
asyncgen finaliser.
+ async with aclosing(log_lines) as lines:
Review Comment:
Good point. No, the benchmark measured only the parsing time, not log
emission through the triggerer. I'll extend it to use the real handler and
analyze the contribution in more detail. Depending on the results, we may need
to move log emission to a thread as well or batch the lines and send them to a
thread.
--
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]
