Vamsi-klu commented on code in PR #73044:
URL: https://github.com/apache/airflow/pull/73044#discussion_r4002486812
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -1088,35 +1087,78 @@ async def read_logs(
method is used for async output of the logs only in the pod
failed it execution or the task was cancelled by the user.
+ :param name: Name of the pod.
+ :param namespace: Name of the pod's namespace.
+ :param container_name: Name of the container inside the pod.
+ :param since_seconds: Only return logs newer than a relative duration
in seconds.
+ """
+ return [
+ line
+ async for line in self.stream_logs(
+ name=name, namespace=namespace, container_name=container_name,
since_seconds=since_seconds
+ )
+ ]
+
+ async def stream_logs(
Review Comment:
stream_logs never looks at raw_resp.status. With _preload_content=False,
kubernetes_asyncio hands back the ClientResponse and does not raise on
403/404/5xx. The error body is split on newlines, yielded as pod logs, and
last_log_time advances. Permanent log loss, and the new retry path in the
trigger never runs for status errors.
Check raw_resp.status before iterating and raise KubernetesApiError (or the
existing permission error) on non-2xx. The old read() path had the same hole.
Streaming makes it worse because the body now looks like successful log lines.
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/pod.py:
##########
@@ -407,10 +411,25 @@ async def _wait_for_container_completion(self) ->
TriggerEvent:
now = datetime.datetime.now(tz=datetime.timezone.utc)
if time_get_more_logs and now >= time_get_more_logs:
if self.get_logs and self.logging_interval:
- self.last_log_time = await
self.pod_manager.fetch_container_logs_before_current_sec(
- pod, container_name=self.base_container_name,
since_time=self.last_log_time
- )
+ # Advance before fetching so a failed read waits a full
interval rather than
+ # being retried on the next poll.
time_get_more_logs = now +
datetime.timedelta(seconds=self.logging_interval)
+ try:
+ self.last_log_time = await
self.pod_manager.fetch_container_logs_before_current_sec(
+ pod,
+ container_name=self.base_container_name,
+ since_time=self.last_log_time,
+ )
+ except (ClientError, asyncio.TimeoutError,
KubernetesApiError) as e:
Review Comment:
This is the behavior change the PR sells: swallow the read, wait a full
logging_interval, do not re-defer. There is no test.
Parameterize ClientError, TimeoutError, and KubernetesApiError. Assert
last_log_time is unchanged, the warning fires, and the next fetch waits
logging_interval.
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:
##########
@@ -1088,35 +1087,78 @@ async def read_logs(
method is used for async output of the logs only in the pod
failed it execution or the task was cancelled by the user.
+ :param name: Name of the pod.
+ :param namespace: Name of the pod's namespace.
+ :param container_name: Name of the container inside the pod.
+ :param since_seconds: Only return logs newer than a relative duration
in seconds.
+ """
+ return [
+ line
+ async for line in self.stream_logs(
+ name=name, namespace=namespace, container_name=container_name,
since_seconds=since_seconds
+ )
+ ]
+
+ async def stream_logs(
+ self, name: str, namespace: str, container_name: str | None = None,
since_seconds: int | None = None
+ ) -> AsyncGenerator[str, None]:
+ """
+ Yield pod log lines one at a time, without buffering the whole window.
+
:param name: Name of the pod.
:param namespace: Name of the pod's namespace.
:param container_name: Name of the container inside the pod.
:param since_seconds: Only return logs newer than a relative duration
in seconds.
"""
async with self.get_conn() as connection:
- try:
- v1_api = async_client.CoreV1Api(connection)
- # Always retrieve raw bytes and decode with 'replace' to avoid
- # UnicodeDecodeError when pod output contains non-UTF-8 bytes
- # (e.g. binary data, truncated multi-byte sequences).
- # kubernetes_asyncio's default decoding uses strict UTF-8 which
- # crashes the task in those cases.
- kwargs: dict[str, Any] = {
- "name": name,
- "namespace": namespace,
- "follow": False,
- "timestamps": True,
- "_preload_content": False,
- }
- if container_name is not None:
- kwargs["container"] = container_name
- if since_seconds is not None:
- kwargs["since_seconds"] = since_seconds
+ v1_api = async_client.CoreV1Api(connection)
+ kwargs: dict[str, Any] = {
+ "name": name,
+ "namespace": namespace,
+ "follow": False,
+ "timestamps": True,
+ "_preload_content": False,
+ }
+ if container_name is not None:
+ kwargs["container"] = container_name
+ if since_seconds is not None:
+ kwargs["since_seconds"] = since_seconds
+ try:
raw_resp: ClientResponse = await
v1_api.read_namespaced_pod_log(**kwargs) # type: ignore #
_preload_content=False makes returning ClientResponse instead of str!
- raw_bytes = await raw_resp.read()
- # CPU-bound decode/split, offloaded so it can't block the
triggerer event loop.
- return await asyncio.to_thread(_split_log_bytes, raw_bytes)
+ try:
+ pending: list[bytes] = []
+ lines_since_yield = 0
+ # Not ``async for line in raw_resp.content``:
``readline()`` raises
+ # ``LineTooLong`` past the 4 MiB high-water mark.
+ # Decoding below uses 'replace' because kubernetes_asyncio
decodes with strict
+ # UTF-8, which raises UnicodeDecodeError and crashes the
task when pod output
+ # contains binary data or truncated multi-byte sequences.
+ async for chunk in
raw_resp.content.iter_chunked(_LOG_STREAM_CHUNK_SIZE):
+ if b"\n" in chunk:
+ *lines, tail = chunk.split(b"\n")
+ if pending:
+ pending.append(lines[0])
+ lines[0] = b"".join(pending)
+ pending.clear()
+ for line in lines:
+ # Safe per line: ``\n`` never appears inside a
multi-byte sequence.
+ yield line.removesuffix(b"\r").decode("utf-8",
errors="replace")
+ lines_since_yield += 1
+ if lines_since_yield >=
_LOG_STREAM_YIELD_EVERY_LINES:
+ lines_since_yield = 0
+ await asyncio.sleep(0)
+ if tail:
+ pending.append(tail)
+ else:
+ # Appended whole: concatenating per chunk would
recopy the line so far
+ # every time, which is quadratic when a container
emits no newline.
+ pending.append(chunk)
+ await asyncio.sleep(0)
+ if pending:
+ yield
b"".join(pending).removesuffix(b"\r").decode("utf-8", errors="replace")
+ finally:
+ raw_resp.close()
except HTTPError as e:
Review Comment:
except HTTPError is urllib3.exceptions.HTTPError. This is the async client.
Failures here are aiohttp / ApiException, so this handler is dead for the new
path unless something still raises urllib3 through this stack. Catch what the
async client actually raises, or delete it.
--
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]
