From f2470fcea838b7482dab2c1b14d6f2133b2ffcf8 Mon Sep 17 00:00:00 2001 From: Yue Chao Qin Date: Thu, 29 Oct 2026 13:09:36 -0700 Subject: [PATCH 1/6] fix(kubernetes): Do not store container command lines in the serialized launcher data --- .../launchers/kubernetes_launchers.py | 15 +++ .../test_kubernetes_command_line_removal.py | 116 ++++++++++++++++++ 2 files changed, 131 insertions(+) create mode 100644 tests/launchers/kubernetes/test_kubernetes_command_line_removal.py diff --git a/cloud_pipelines_backend/launchers/kubernetes_launchers.py b/cloud_pipelines_backend/launchers/kubernetes_launchers.py index 65634d5b..23f882bb 100644 --- a/cloud_pipelines_backend/launchers/kubernetes_launchers.py +++ b/cloud_pipelines_backend/launchers/kubernetes_launchers.py @@ -2139,6 +2139,9 @@ def _serialize_kubernetes_object_to_compact_dict(obj) -> dict[str, Any]: obj_dict = _kubernetes_serialize(obj) # Removing trash _remove_keys_with_none_values(obj_dict) + # A container that consumes a secret as a value receives that value on its command line, + # so a stored command line can hold a secret in plaintext. + _remove_command_lines(obj_dict) obj_metadata: dict | None = obj_dict.get("metadata") if obj_metadata: obj_metadata.pop("managedFields", None) @@ -2191,6 +2194,18 @@ def _remove_keys_with_none_values(d: dict): _remove_keys_with_none_values(v) +def _remove_command_lines(d: dict): + for k, v in list(d.items()): + if k in ("command", "args"): + del d[k] + elif isinstance(v, dict): + _remove_command_lines(v) + elif isinstance(v, list): + for item in v: + if isinstance(item, dict): + _remove_command_lines(item) + + def _launcher_error_from_api_exception( exception: kubernetes.client.exceptions.ApiException, *, diff --git a/tests/launchers/kubernetes/test_kubernetes_command_line_removal.py b/tests/launchers/kubernetes/test_kubernetes_command_line_removal.py new file mode 100644 index 00000000..72c78f0b --- /dev/null +++ b/tests/launchers/kubernetes/test_kubernetes_command_line_removal.py @@ -0,0 +1,116 @@ +"""Tests that serialized (stored) Kubernetes objects carry no container command lines. + +A container that consumes a secret as a value receives that value on its command line, so a +stored command line can hold a secret in plaintext. +""" + +from __future__ import annotations + +import json +from typing import Final + +from kubernetes import client as k8s_client_lib + +from cloud_pipelines_backend.launchers import kubernetes_launchers + +_SECRET_VALUE: Final[str] = "plaintext-secret-value" + + +def _make_container() -> k8s_client_lib.V1Container: + return k8s_client_lib.V1Container( + name="main", + image="alpine", + command=["python3", "-c", "print(1)"], + args=["--token", _SECRET_VALUE], + ) + + +def _make_pod() -> k8s_client_lib.V1Pod: + return k8s_client_lib.V1Pod( + metadata=k8s_client_lib.V1ObjectMeta(name="test-pod", namespace="test"), + spec=k8s_client_lib.V1PodSpec(containers=[_make_container()]), + ) + + +def _make_job() -> k8s_client_lib.V1Job: + return k8s_client_lib.V1Job( + metadata=k8s_client_lib.V1ObjectMeta(name="test-job", namespace="test"), + spec=k8s_client_lib.V1JobSpec( + template=k8s_client_lib.V1PodTemplateSpec( + spec=k8s_client_lib.V1PodSpec(containers=[_make_container()]), + ), + ), + ) + + +def test_serialized_pod_has_no_command_line() -> None: + pod_dict = kubernetes_launchers._serialize_kubernetes_object_to_compact_dict( + _make_pod() + ) + + container = pod_dict["spec"]["containers"][0] + assert "command" not in container + assert "args" not in container + assert container["image"] == "alpine" + assert _SECRET_VALUE not in json.dumps(pod_dict) + + +def test_serialized_job_has_no_command_line_on_its_pod_template() -> None: + job_dict = kubernetes_launchers._serialize_kubernetes_object_to_compact_dict( + _make_job() + ) + + container = job_dict["spec"]["template"]["spec"]["containers"][0] + assert "command" not in container + assert "args" not in container + assert _SECRET_VALUE not in json.dumps(job_dict) + + +def test_serialized_pod_has_no_command_line_anywhere() -> None: + """The removal is not selective: every command line goes, whatever holds it. + + An init container, a sidecar and a probe are all stored, and any of them can be the one + carrying a secret, so none is left behind. + """ + pod = k8s_client_lib.V1Pod( + metadata=k8s_client_lib.V1ObjectMeta(name="test-pod", namespace="test"), + spec=k8s_client_lib.V1PodSpec( + init_containers=[ + k8s_client_lib.V1Container( + name="init", + image="alpine", + command=["sh", "-c", _SECRET_VALUE], + ), + ], + containers=[ + _make_container(), + k8s_client_lib.V1Container( + name="sidecar", + image="alpine", + args=["--token", _SECRET_VALUE], + readiness_probe=k8s_client_lib.V1Probe( + _exec=k8s_client_lib.V1ExecAction( + command=["check", _SECRET_VALUE], + ), + ), + ), + ], + ), + ) + + pod_dict = kubernetes_launchers._serialize_kubernetes_object_to_compact_dict(pod) + + assert "command" not in pod_dict["spec"]["initContainers"][0] + sidecar = pod_dict["spec"]["containers"][1] + assert "args" not in sidecar + assert "command" not in sidecar["readinessProbe"]["exec"] + assert _SECRET_VALUE not in json.dumps(pod_dict) + + +def test_launched_pod_keeps_its_command_line() -> None: + """The object sent to the cluster must be untouched: it is how the container is started.""" + pod = _make_pod() + + kubernetes_launchers._serialize_kubernetes_object_to_compact_dict(pod) + + assert pod.spec.containers[0].args == ["--token", _SECRET_VALUE] From b709e6b53bb54bf26b068f53aca5d9bd5805154c Mon Sep 17 00:00:00 2001 From: Yue Chao Qin Date: Thu, 29 Oct 2026 14:39:42 -0700 Subject: [PATCH 2/6] test(kubernetes): Cover the stored payload of both launched-container classes --- .../test_kubernetes_command_line_removal.py | 44 +++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/tests/launchers/kubernetes/test_kubernetes_command_line_removal.py b/tests/launchers/kubernetes/test_kubernetes_command_line_removal.py index 72c78f0b..9d89ec3b 100644 --- a/tests/launchers/kubernetes/test_kubernetes_command_line_removal.py +++ b/tests/launchers/kubernetes/test_kubernetes_command_line_removal.py @@ -107,6 +107,50 @@ def test_serialized_pod_has_no_command_line_anywhere() -> None: assert _SECRET_VALUE not in json.dumps(pod_dict) +def test_launched_container_to_dict_has_no_command_line() -> None: + """The Pod launcher's stored payload: what the orchestrator writes to `launcher_data`.""" + launched = kubernetes_launchers.LaunchedKubernetesContainer( + pod_name="test-pod", + namespace="test", + output_uris={}, + log_uri="gs://bucket/log", + debug_pod=_make_pod(), + ) + + launcher_data = launched.to_dict() + + stored_pod = launcher_data["kubernetes"]["debug_pod"] + assert "command" not in stored_pod["spec"]["containers"][0] + assert "args" not in stored_pod["spec"]["containers"][0] + assert _SECRET_VALUE not in json.dumps(launcher_data) + + +def test_launched_job_to_dict_has_no_command_line_in_any_copy() -> None: + """The Job launcher stores the Pod template and one copy per Pod; all of them are cleaned.""" + launched = kubernetes_launchers.LaunchedKubernetesJob( + job_name="test-job", + namespace="test", + output_uris={}, + log_uri="gs://bucket/log", + debug_job=_make_job(), + debug_pods={"test-job-0-aaa": _make_pod(), "test-job-0-bbb": _make_pod()}, + ) + + launcher_data = launched.to_dict() + + stored = launcher_data["kubernetes_job"] + template_container = stored["debug_job"]["spec"]["template"]["spec"]["containers"][ + 0 + ] + assert "command" not in template_container + assert "args" not in template_container + assert len(stored["debug_pods"]) == 2 + for stored_pod in stored["debug_pods"].values(): + assert "command" not in stored_pod["spec"]["containers"][0] + assert "args" not in stored_pod["spec"]["containers"][0] + assert _SECRET_VALUE not in json.dumps(launcher_data) + + def test_launched_pod_keeps_its_command_line() -> None: """The object sent to the cluster must be untouched: it is how the container is started.""" pod = _make_pod() From 9c6394376e033615b6fe378c7c4a6ebf55af7b5a Mon Sep 17 00:00:00 2001 From: Yue Chao Qin Date: Wed, 4 Nov 2026 21:33:03 -0800 Subject: [PATCH 3/6] refactor(kubernetes): Censor command lines in a named wrapper rather than in the shared serializer _serialize_kubernetes_object_to_compact_dict no longer removes command lines; a _serialize_kubernetes_object_to_compact_dict_and_censor_it wrapper does, and the three call sites that persist a launched object use it. --- .../launchers/kubernetes_launchers.py | 24 ++++++++++++++----- .../test_kubernetes_command_line_removal.py | 22 ++++++++++++----- 2 files changed, 34 insertions(+), 12 deletions(-) diff --git a/cloud_pipelines_backend/launchers/kubernetes_launchers.py b/cloud_pipelines_backend/launchers/kubernetes_launchers.py index 23f882bb..4e3e201c 100644 --- a/cloud_pipelines_backend/launchers/kubernetes_launchers.py +++ b/cloud_pipelines_backend/launchers/kubernetes_launchers.py @@ -931,7 +931,9 @@ def launcher_error_message(self) -> str | None: return launcher_error_message def to_dict(self) -> dict[str, Any]: - pod_dict = _serialize_kubernetes_object_to_compact_dict(self._debug_pod) + pod_dict = _serialize_kubernetes_object_to_compact_dict_and_censor_it( + obj=self._debug_pod + ) result = dict( kubernetes=dict( # launched_container_class_name=self.__class__.__name__, @@ -1577,12 +1579,16 @@ def launcher_error_message(self) -> str | None: SERIALIZATION_ROOT_KEY = "kubernetes_job" def to_dict(self) -> dict[str, Any]: - job_dict = _serialize_kubernetes_object_to_compact_dict(self._debug_job) + job_dict = _serialize_kubernetes_object_to_compact_dict_and_censor_it( + obj=self._debug_job + ) pod_dicts = None if self._debug_pods is not None: pod_dicts = { pod_name: ( - _serialize_kubernetes_object_to_compact_dict(pod) if pod else None + _serialize_kubernetes_object_to_compact_dict_and_censor_it(obj=pod) + if pod + else None ) for pod_name, pod in self._debug_pods.items() } @@ -2139,15 +2145,21 @@ def _serialize_kubernetes_object_to_compact_dict(obj) -> dict[str, Any]: obj_dict = _kubernetes_serialize(obj) # Removing trash _remove_keys_with_none_values(obj_dict) - # A container that consumes a secret as a value receives that value on its command line, - # so a stored command line can hold a secret in plaintext. - _remove_command_lines(obj_dict) obj_metadata: dict | None = obj_dict.get("metadata") if obj_metadata: obj_metadata.pop("managedFields", None) return obj_dict +def _serialize_kubernetes_object_to_compact_dict_and_censor_it( + *, obj: k8s_client_lib.V1Job | k8s_client_lib.V1Pod +) -> dict[str, Any]: + """Serializes a Kubernetes object without the command lines, which can hold a secret that the container consumes as a value.""" + obj_dict = _serialize_kubernetes_object_to_compact_dict(obj) + _remove_command_lines(obj_dict) + return obj_dict + + def windows_path_to_docker_path(path: str) -> str: if os.name != "nt": return path diff --git a/tests/launchers/kubernetes/test_kubernetes_command_line_removal.py b/tests/launchers/kubernetes/test_kubernetes_command_line_removal.py index 9d89ec3b..ffe64dcc 100644 --- a/tests/launchers/kubernetes/test_kubernetes_command_line_removal.py +++ b/tests/launchers/kubernetes/test_kubernetes_command_line_removal.py @@ -44,8 +44,10 @@ def _make_job() -> k8s_client_lib.V1Job: def test_serialized_pod_has_no_command_line() -> None: - pod_dict = kubernetes_launchers._serialize_kubernetes_object_to_compact_dict( - _make_pod() + pod_dict = ( + kubernetes_launchers._serialize_kubernetes_object_to_compact_dict_and_censor_it( + obj=_make_pod() + ) ) container = pod_dict["spec"]["containers"][0] @@ -56,8 +58,10 @@ def test_serialized_pod_has_no_command_line() -> None: def test_serialized_job_has_no_command_line_on_its_pod_template() -> None: - job_dict = kubernetes_launchers._serialize_kubernetes_object_to_compact_dict( - _make_job() + job_dict = ( + kubernetes_launchers._serialize_kubernetes_object_to_compact_dict_and_censor_it( + obj=_make_job() + ) ) container = job_dict["spec"]["template"]["spec"]["containers"][0] @@ -98,7 +102,11 @@ def test_serialized_pod_has_no_command_line_anywhere() -> None: ), ) - pod_dict = kubernetes_launchers._serialize_kubernetes_object_to_compact_dict(pod) + pod_dict = ( + kubernetes_launchers._serialize_kubernetes_object_to_compact_dict_and_censor_it( + obj=pod + ) + ) assert "command" not in pod_dict["spec"]["initContainers"][0] sidecar = pod_dict["spec"]["containers"][1] @@ -155,6 +163,8 @@ def test_launched_pod_keeps_its_command_line() -> None: """The object sent to the cluster must be untouched: it is how the container is started.""" pod = _make_pod() - kubernetes_launchers._serialize_kubernetes_object_to_compact_dict(pod) + kubernetes_launchers._serialize_kubernetes_object_to_compact_dict_and_censor_it( + obj=pod + ) assert pod.spec.containers[0].args == ["--token", _SECRET_VALUE] From eabb056b77b9b26d908c8c35d3466d497a90acde Mon Sep 17 00:00:00 2001 From: Yue Chao Qin Date: Tue, 3 Nov 2026 16:42:12 -0800 Subject: [PATCH 4/6] fix: Kubernetes launcher - Read a container log within a budget, and give up quietly Reading a container log goes through the API server's proxy to the node's kubelet, and a read that fails has almost always lost the log for good. The caller retries `upload_log` five times, so each lost log cost its single loop thread five requests and five waits for a log that would never arrive. Both Kubernetes launchers now read within their own budget and return quietly when it runs out. The budget wraps the read only: a failed write to storage still raises and still earns the caller's retries. --- .../launchers/kubernetes_launchers.py | 55 +++- tests/test_kubernetes_log_upload_budget.py | 273 ++++++++++++++++++ 2 files changed, 325 insertions(+), 3 deletions(-) create mode 100644 tests/test_kubernetes_log_upload_budget.py diff --git a/cloud_pipelines_backend/launchers/kubernetes_launchers.py b/cloud_pipelines_backend/launchers/kubernetes_launchers.py index 4e3e201c..ea9382f0 100644 --- a/cloud_pipelines_backend/launchers/kubernetes_launchers.py +++ b/cloud_pipelines_backend/launchers/kubernetes_launchers.py @@ -9,8 +9,9 @@ import os import pathlib import threading +import time import typing -from typing import Any, Optional +from typing import Any, Final, Optional import kubernetes.client.exceptions from kubernetes import client as k8s_client_lib @@ -796,6 +797,43 @@ class KubernetesWithGcsFuseContainerLauncher(GoogleKubernetesEngineLauncher): pass +# A failed log read means the node is gone, so retrying only blocks the caller. +_LOG_READ_MAX_RETRIES: Final[int] = 1 +_LOG_READ_WAIT_SECONDS: Final[float] = 1.0 +_LOG_READ_MAX_TOTAL_SECONDS: Final[float | None] = 10.0 + + +def _read_log_within_budget( + read_log: typing.Callable[[], _T], + *, + max_retries: int = _LOG_READ_MAX_RETRIES, + wait_seconds: float = _LOG_READ_WAIT_SECONDS, + max_total_seconds: float | None = _LOG_READ_MAX_TOTAL_SECONDS, +) -> _T: + """Calls `read_log`, retrying within a budget. Raises if the budget runs out. + + A retry never starts once the deadline has passed. The final attempt can + still overrun it by one request timeout, which `request_timeout` bounds. + """ + start_seconds = time.monotonic() + for attempt in range(max_retries - 1): + try: + return read_log() + except Exception: + _logger.exception( + f"Exception reading a container log on attempt {attempt + 1} of " + f"{max_retries}. Retrying." + ) + if ( + max_total_seconds is not None + and time.monotonic() - start_seconds + wait_seconds > max_total_seconds + ): + raise + time.sleep(wait_seconds) + # The last attempt is not retried, so its exception is the caller's. + return read_log() + + class LaunchedKubernetesContainer(interfaces.LaunchedContainer): def __init__( self, @@ -1011,8 +1049,14 @@ def get_log(self) -> str: response.release_conn() def upload_log(self): + try: + log = _read_log_within_budget(self.get_log) + except Exception: + # Not raising: the caller retries `upload_log`, so raising costs it + # this budget five times over. Writing nothing keeps an earlier log. + _logger.exception(f"Giving up on the log of Pod {self._pod_name}.") + return launcher = self._get_launcher() - log = self.get_log() uri_writer = launcher._storage_provider.make_uri(self._log_uri).get_writer() uri_writer.upload_from_text(log) @@ -1865,7 +1909,12 @@ def get_log(self) -> str: return merged_log def upload_log(self): - all_logs = self._get_all_logs() + try: + all_logs = _read_log_within_budget(self._get_all_logs) + except Exception: + # See the note on `LaunchedKubernetesContainer.upload_log`. + _logger.exception(f"Giving up on the logs of Job {self._job_name}.") + return merged_log = self._merge_logs(all_logs) # Uploading the merged log diff --git a/tests/test_kubernetes_log_upload_budget.py b/tests/test_kubernetes_log_upload_budget.py new file mode 100644 index 00000000..c4615bf0 --- /dev/null +++ b/tests/test_kubernetes_log_upload_budget.py @@ -0,0 +1,273 @@ +"""Tests for the Kubernetes launchers' log-upload budget. + +The orchestrator calls ``upload_log`` inside its own retry, so these tests pin +the two halves of the contract that keeps a dead node from holding its loop: the +log *read* happens at most once, and giving up does not raise. +""" + +from unittest import mock + +import kubernetes.client.exceptions +import pytest + +from cloud_pipelines_backend import orchestrator_sql +from cloud_pipelines_backend.launchers import kubernetes_launchers + +_LAUNCHED_CONTAINER_CLASSES = [ + kubernetes_launchers.LaunchedKubernetesContainer, + kubernetes_launchers.LaunchedKubernetesJob, +] +_IDS = ["pod", "job"] + + +class TestBudget: + """Both launchers call the helper with no arguments, so its defaults are the budget.""" + + def test_the_defaults_are_one_attempt_within_ten_seconds(self) -> None: + assert kubernetes_launchers._LOG_READ_MAX_RETRIES == 1 + assert kubernetes_launchers._LOG_READ_WAIT_SECONDS == 1.0 + assert kubernetes_launchers._LOG_READ_MAX_TOTAL_SECONDS == 10.0 + + def test_calling_with_no_arguments_uses_them(self) -> None: + read_log = mock.MagicMock(side_effect=RuntimeError("kubelet unreachable")) + with mock.patch("time.sleep") as sleep: + with pytest.raises(RuntimeError): + kubernetes_launchers._read_log_within_budget(read_log) + assert read_log.call_count == 1 + assert sleep.call_count == 0 + + +class TestReadLogWithinBudget: + """The helper itself, independent of either launcher.""" + + def test_returns_the_log(self) -> None: + assert ( + kubernetes_launchers._read_log_within_budget( + lambda: "the log", + max_retries=1, + wait_seconds=1.0, + max_total_seconds=10.0, + ) + == "the log" + ) + + def test_a_single_attempt_does_not_sleep(self) -> None: + read_log = mock.MagicMock(side_effect=RuntimeError("kubelet unreachable")) + with mock.patch("time.sleep") as sleep: + with pytest.raises(RuntimeError): + kubernetes_launchers._read_log_within_budget( + read_log, max_retries=1, wait_seconds=1.0, max_total_seconds=10.0 + ) + assert read_log.call_count == 1 + assert sleep.call_count == 0 + + def test_retries_without_a_trailing_sleep(self) -> None: + """The sleep after the last failure only delays the caller.""" + read_log = mock.MagicMock(side_effect=RuntimeError("kubelet unreachable")) + with mock.patch("time.sleep") as sleep: + with pytest.raises(RuntimeError): + kubernetes_launchers._read_log_within_budget( + read_log, max_retries=3, wait_seconds=1.0, max_total_seconds=None + ) + assert read_log.call_count == 3 + assert sleep.call_count == 2 + + def test_the_deadline_stops_retrying_early(self) -> None: + read_log = mock.MagicMock(side_effect=RuntimeError("kubelet unreachable")) + # Each attempt costs 1s of wall time, on top of the 1s wait. + clock = iter([0.0, 1.0, 3.0, 5.0, 7.0, 9.0, 11.0]) + with mock.patch("time.monotonic", lambda: next(clock)): + with mock.patch("time.sleep") as sleep: + with pytest.raises(RuntimeError): + kubernetes_launchers._read_log_within_budget( + read_log, + max_retries=99, + wait_seconds=1.0, + max_total_seconds=3.0, + ) + assert read_log.call_count == 2 + assert sleep.call_count == 1 + + +def _reader_of(container) -> mock.MagicMock: + """The one call that reaches the kubelet, for whichever class this is.""" + if isinstance(container, kubernetes_launchers.LaunchedKubernetesContainer): + return container.get_log + return container._get_all_logs + + +def _container_that_cannot_read_its_log(launched_container_class: type): + """A real launched container, uninitialised, whose log read always fails. + + Built from the real class rather than a `spec` mock, so the budget read by + `upload_log` is the class's own and a renamed attribute fails here. + """ + container = object.__new__(launched_container_class) + container._pod_name = "test-pod" + container._job_name = "test-job" + container._log_uri = "memory://log" + container._get_launcher = mock.MagicMock() + failure = RuntimeError("kubelet unreachable") + if launched_container_class is kubernetes_launchers.LaunchedKubernetesContainer: + container.get_log = mock.MagicMock(side_effect=failure) + else: + container._get_all_logs = mock.MagicMock(side_effect=failure) + return container + + +class TestUploadLogGivesUp: + @pytest.mark.parametrize( + "launched_container_class", _LAUNCHED_CONTAINER_CLASSES, ids=_IDS + ) + def test_a_lost_log_is_read_once_and_does_not_raise( + self, launched_container_class: type + ) -> None: + """Raising here would cost the caller its own five attempts.""" + container = _container_that_cannot_read_its_log(launched_container_class) + with mock.patch("time.sleep") as sleep: + container.upload_log() + assert _reader_of(container).call_count == 1 + assert sleep.call_count == 0 + + @pytest.mark.parametrize( + "launched_container_class", _LAUNCHED_CONTAINER_CLASSES, ids=_IDS + ) + def test_nothing_is_written_when_the_log_is_lost( + self, launched_container_class: type + ) -> None: + """An empty upload would overwrite a log an earlier attempt stored.""" + container = _container_that_cannot_read_its_log(launched_container_class) + with mock.patch("time.sleep"): + container.upload_log() + container._get_launcher.assert_not_called() + + +class TestStorageWriteStillRaises: + """A failed write to storage is worth the caller's retries; a dead node is not.""" + + def test_pod_upload_propagates_a_storage_failure(self) -> None: + container = object.__new__(kubernetes_launchers.LaunchedKubernetesContainer) + container._pod_name = "test-pod" + container._log_uri = "memory://log" + container.get_log = mock.MagicMock(return_value="the log") + launcher = mock.MagicMock() + container._get_launcher = mock.MagicMock(return_value=launcher) + writer = ( + launcher._storage_provider.make_uri.return_value.get_writer.return_value + ) + writer.upload_from_text.side_effect = RuntimeError("storage unavailable") + with pytest.raises(RuntimeError, match="storage unavailable"): + container.upload_log() + + +# The body a real unreachable kubelet returns, with the cluster's own identifiers +# removed. Observed in production on the container-log endpoint. +_KUBELET_UNREACHABLE_BODY = ( + '{"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":' + '"Get \\"https://:10250/containerLogs///main' + '?timestamps=true\\": dial tcp :10250: i/o timeout","code":500}' +) + + +class _FakeResponse: + """Enough of an HTTP response for `ApiException`, across client versions. + + Older clients read the headers with `getheaders()`, newer ones with + `getheader()`, and the body is stored undecoded. + """ + + def __init__(self, *, status: int, reason: str, data: bytes) -> None: + self.status = status + self.reason = reason + self.data = data + self.headers: dict[str, str] = {} + + def getheaders(self) -> dict[str, str]: + return self.headers + + def getheader(self, name: str, default: str | None = None) -> str | None: + return self.headers.get(name, default) + + +def _kubelet_unreachable() -> kubernetes.client.exceptions.ApiException: + """The exception a dead node actually produces, not a stand-in.""" + return kubernetes.client.exceptions.ApiException( + http_resp=_FakeResponse( + status=500, + reason="Internal Server Error", + data=_KUBELET_UNREACHABLE_BODY.encode(), + ) + ) + + +class TestARealKubeletFailureIsSwallowed: + """Closes the gap between "the loop works" and "it catches what is thrown". + + These build the launched containers through their real constructors and let + the real read path run, so the exception reaching the budget is one the + Kubernetes client raised, not a `RuntimeError` a test invented. + """ + + def test_pod_log_read(self) -> None: + container = kubernetes_launchers.LaunchedKubernetesContainer( + pod_name="test-pod", + namespace="test-ns", + output_uris={}, + log_uri="file:///dev/null", + debug_pod=mock.MagicMock(), + launcher=mock.MagicMock(_request_timeout=10), + ) + core_api = mock.MagicMock() + core_api.read_namespaced_pod_log.side_effect = _kubelet_unreachable() + with mock.patch.object( + kubernetes_launchers.k8s_client_lib, "CoreV1Api", return_value=core_api + ): + with mock.patch("time.sleep") as sleep: + container.upload_log() + assert core_api.read_namespaced_pod_log.call_count == 1 + assert sleep.call_count == 0 + + def test_job_log_read(self) -> None: + job = kubernetes_launchers.LaunchedKubernetesJob( + job_name="test-job", + namespace="test-ns", + output_uris={}, + log_uri="file:///dev/null", + debug_job=mock.MagicMock(), + launcher=mock.MagicMock(_request_timeout=10), + ) + pod = mock.MagicMock() + pod.metadata.name = "test-pod" + core_api = mock.MagicMock() + core_api.read_namespaced_pod_log.side_effect = _kubelet_unreachable() + # Only the Pod listing is faked; the read below it is the real one. + with mock.patch.object(job, "_get_all_attempt_pods", return_value={"0": [pod]}): + with mock.patch.object( + kubernetes_launchers.k8s_client_lib, "CoreV1Api", return_value=core_api + ): + with mock.patch("time.sleep") as sleep: + job.upload_log() + assert core_api.read_namespaced_pod_log.call_count == 1 + assert sleep.call_count == 0 + + +class TestCostToTheCaller: + """What the orchestrator's own retry does with the result, using the real `_retry`.""" + + def test_a_lost_log_costs_one_request_instead_of_five(self) -> None: + container = _container_that_cannot_read_its_log( + kubernetes_launchers.LaunchedKubernetesJob + ) + with mock.patch("time.sleep") as sleep: + orchestrator_sql._retry(container.upload_log) + assert container._get_all_logs.call_count == 1 + assert sleep.call_count == 0 + + def test_the_same_failure_raised_would_have_cost_five(self) -> None: + """Why the give-up returns: this is the behaviour it replaces.""" + raising_upload = mock.MagicMock(side_effect=RuntimeError("kubelet unreachable")) + with mock.patch("time.sleep") as sleep: + with pytest.raises(RuntimeError): + orchestrator_sql._retry(raising_upload) + assert raising_upload.call_count == 5 + assert sleep.call_count == 5 From f15728db5285f30f376b38f3dbc33ae666f2ae26 Mon Sep 17 00:00:00 2001 From: Yue Chao Qin Date: Fri, 30 Oct 2026 04:10:14 -0700 Subject: [PATCH 5/6] fix: Kubernetes launcher - Treat an unreachable kubelet as a lost log A container log is read through the API server's proxy to the node's kubelet. When the cluster-autoscaler reclaims the node, that proxy fails with a 500 whose body reports a failed dial of port 10250. The log died with the node, so retrying cannot recover it; the read is a lost cause from the first attempt. `_get_log_by_pod_key` now returns None for that error, beside the existing branches for a deleted Pod (404) and a container that has not started (400), and logs a warning carrying the full response body. The match is deliberately narrow: a 500 whose body names the container-log endpoint and a failed dial. Any other 500 still raises, so a reworded or unrelated error surfaces instead of being swallowed. --- .../launchers/kubernetes_launchers.py | 47 ++++++++++++ ...ubernetes_launcher_error_classification.py | 76 +++++++++++++++++++ 2 files changed, 123 insertions(+) diff --git a/cloud_pipelines_backend/launchers/kubernetes_launchers.py b/cloud_pipelines_backend/launchers/kubernetes_launchers.py index ea9382f0..0c20734c 100644 --- a/cloud_pipelines_backend/launchers/kubernetes_launchers.py +++ b/cloud_pipelines_backend/launchers/kubernetes_launchers.py @@ -1799,6 +1799,16 @@ def _get_log_by_pod_key(self, pod_name: str) -> str | None: f"Pod {pod_name} no longer exists; its logs are unrecoverable." ) return None + if _is_kubelet_unreachable_error(ex): + # The API server could not reach the kubelet of the Pod's node. The + # cluster-autoscaler removes a node soon after the last Pod on it + # completes, and the logs lived on that node, so they are gone. Retrying + # cannot recover them and only blocks the caller. + _logger.warning( + f"Kubelet of Pod {pod_name} is unreachable; its logs are " + f"unrecoverable. {_api_exception_body(ex)}" + ) + return None raise def _get_all_attempt_pods(self) -> dict[str, list[k8s_client_lib.V1Pod]]: @@ -2267,6 +2277,43 @@ def _remove_command_lines(d: dict): _remove_command_lines(item) +def _api_exception_body(exception: kubernetes.client.exceptions.ApiException) -> str: + """The response body as text, whichever way the installed client stored it. + + Older Kubernetes clients assign the raw `bytes`; newer ones decode to `str` in + `ApiException.__init__` and leave the body `None` if decoding fails. + """ + body = exception.body + if isinstance(body, bytes): + return body.decode("utf-8", errors="replace") + return body or "" + + +def _is_kubelet_unreachable_error( + exception: kubernetes.client.exceptions.ApiException, +) -> bool: + """Whether the API server failed to dial the kubelet while proxying a log read. + + Raised as `ServiceException`, a subclass of `ApiException`. An observed response + body, with cluster identifiers replaced: + + {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure", + "message":"Get \\"https://:10250/containerLogs///main?timestamps=true\\": + dial tcp :10250: i/o timeout","code":500} + + The wording comes from the API server's log proxy and is not a documented + contract, so treat a non-match as "some other error" rather than as proof the + kubelet was reachable. Callers that must bound their wall time regardless set + the log-upload retry budget on the launched container instead. + """ + body = _api_exception_body(exception) + return ( + exception.status == http.HTTPStatus.INTERNAL_SERVER_ERROR + and "/containerLogs/" in body + and "dial tcp" in body + ) + + def _launcher_error_from_api_exception( exception: kubernetes.client.exceptions.ApiException, *, diff --git a/tests/test_kubernetes_launcher_error_classification.py b/tests/test_kubernetes_launcher_error_classification.py index 39d0a3d2..f471eb48 100644 --- a/tests/test_kubernetes_launcher_error_classification.py +++ b/tests/test_kubernetes_launcher_error_classification.py @@ -13,6 +13,43 @@ def _api_exception(status: int) -> kubernetes.client.exceptions.ApiException: return kubernetes.client.exceptions.ApiException(status=status, reason="test") +# Verbatim shape of a production 500, with cluster identifiers replaced. The client +# decodes `http_resp.data` itself, so handing it bytes is what exercises the real path. +_KUBELET_DIAL_TIMEOUT_BODY = ( + '{"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure",' + '"message":"Get \\"https://10.0.0.1:10250/containerLogs/jobs-namespace/' + 'job-pod-0/main?timestamps=true\\": dial tcp 10.0.0.1:10250: i/o timeout",' + '"code":500}' +) + + +class _FakeResponse: + """Enough of a urllib3 response for the Kubernetes client's exception classes. + + Both header shapes are provided because the client changed how it reads them: + older versions call `getheaders()`, newer ones read `.headers`. Older versions + also assign the body without decoding, which is why `_api_exception_body` has to + cope with `bytes` as well as `str`. + """ + + status = 500 + reason = "Internal Server Error" + headers: dict[str, str] = {} + + def __init__(self, data: bytes) -> None: + self.data = data + + def getheaders(self) -> dict[str, str]: + return self.headers + + +def _service_exception(body: str) -> kubernetes.client.exceptions.ServiceException: + """The exception the Kubernetes client actually raises for a 500, built as it builds it.""" + return kubernetes.client.exceptions.ServiceException( + http_resp=_FakeResponse(body.encode()) + ) + + class TestLauncherErrorFromApiException: def test_server_error_is_retriable(self) -> None: error = kubernetes_launchers._launcher_error_from_api_exception( @@ -89,3 +126,42 @@ def test_server_error_reraises(self) -> None: ): with pytest.raises(kubernetes.client.exceptions.ApiException): job._get_log_by_pod_key("pod-0") + + def test_unreachable_kubelet_returns_none(self) -> None: + """A node removed by the cluster-autoscaler takes the Pod's logs with it.""" + job, core_api = _make_job_with_pod_log_error( + _service_exception(_KUBELET_DIAL_TIMEOUT_BODY) + ) + with mock.patch.object( + kubernetes_launchers.k8s_client_lib, "CoreV1Api", return_value=core_api + ): + assert job._get_log_by_pod_key("pod-0") is None + + def test_differently_worded_kubelet_failure_still_reraises(self) -> None: + """The API server's wording is not a contract; a non-match must not be swallowed.""" + job, core_api = _make_job_with_pod_log_error( + _service_exception( + '{"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure",' + '"message":"Get \\"https://10.0.0.1:10250/containerLogs/ns/pod-0/main\\": ' + 'remote error: tls: internal error","code":500}' + ) + ) + with mock.patch.object( + kubernetes_launchers.k8s_client_lib, "CoreV1Api", return_value=core_api + ): + with pytest.raises(kubernetes.client.exceptions.ApiException): + job._get_log_by_pod_key("pod-0") + + def test_unrelated_server_error_body_still_reraises(self) -> None: + """The match is narrow: only a failed kubelet dial on a log read.""" + job, core_api = _make_job_with_pod_log_error( + _service_exception( + '{"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure",' + '"message":"etcdserver: request timed out","code":500}' + ) + ) + with mock.patch.object( + kubernetes_launchers.k8s_client_lib, "CoreV1Api", return_value=core_api + ): + with pytest.raises(kubernetes.client.exceptions.ApiException): + job._get_log_by_pod_key("pod-0") From ab15742e6d3e0f8f3cbbcf14b20e2082ffd1133a Mon Sep 17 00:00:00 2001 From: Yue Chao Qin Date: Wed, 4 Nov 2026 21:48:33 -0800 Subject: [PATCH 6/6] refactor(kubernetes): Rename _api_exception_body to _extract_api_exception_body --- cloud_pipelines_backend/launchers/kubernetes_launchers.py | 8 +++++--- tests/test_kubernetes_launcher_error_classification.py | 2 +- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/cloud_pipelines_backend/launchers/kubernetes_launchers.py b/cloud_pipelines_backend/launchers/kubernetes_launchers.py index 0c20734c..1bbea0a7 100644 --- a/cloud_pipelines_backend/launchers/kubernetes_launchers.py +++ b/cloud_pipelines_backend/launchers/kubernetes_launchers.py @@ -1806,7 +1806,7 @@ def _get_log_by_pod_key(self, pod_name: str) -> str | None: # cannot recover them and only blocks the caller. _logger.warning( f"Kubelet of Pod {pod_name} is unreachable; its logs are " - f"unrecoverable. {_api_exception_body(ex)}" + f"unrecoverable. {_extract_api_exception_body(ex)}" ) return None raise @@ -2277,7 +2277,9 @@ def _remove_command_lines(d: dict): _remove_command_lines(item) -def _api_exception_body(exception: kubernetes.client.exceptions.ApiException) -> str: +def _extract_api_exception_body( + exception: kubernetes.client.exceptions.ApiException, +) -> str: """The response body as text, whichever way the installed client stored it. Older Kubernetes clients assign the raw `bytes`; newer ones decode to `str` in @@ -2306,7 +2308,7 @@ def _is_kubelet_unreachable_error( kubelet was reachable. Callers that must bound their wall time regardless set the log-upload retry budget on the launched container instead. """ - body = _api_exception_body(exception) + body = _extract_api_exception_body(exception) return ( exception.status == http.HTTPStatus.INTERNAL_SERVER_ERROR and "/containerLogs/" in body diff --git a/tests/test_kubernetes_launcher_error_classification.py b/tests/test_kubernetes_launcher_error_classification.py index f471eb48..7f1b40bc 100644 --- a/tests/test_kubernetes_launcher_error_classification.py +++ b/tests/test_kubernetes_launcher_error_classification.py @@ -28,7 +28,7 @@ class _FakeResponse: Both header shapes are provided because the client changed how it reads them: older versions call `getheaders()`, newer ones read `.headers`. Older versions - also assign the body without decoding, which is why `_api_exception_body` has to + also assign the body without decoding, which is why `_extract_api_exception_body` has to cope with `bytes` as well as `str`. """