diff --git a/cloud_pipelines_backend/launchers/kubernetes_launchers.py b/cloud_pipelines_backend/launchers/kubernetes_launchers.py index 65634d5b..1bbea0a7 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, @@ -931,7 +969,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__, @@ -1009,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) @@ -1577,12 +1623,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() } @@ -1749,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. {_extract_api_exception_body(ex)}" + ) + return None raise def _get_all_attempt_pods(self) -> dict[str, list[k8s_client_lib.V1Pod]]: @@ -1859,7 +1919,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 @@ -2145,6 +2210,15 @@ def _serialize_kubernetes_object_to_compact_dict(obj) -> dict[str, Any]: 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 @@ -2191,6 +2265,57 @@ 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 _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 + `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 = _extract_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/launchers/kubernetes/test_kubernetes_command_line_removal.py b/tests/launchers/kubernetes/test_kubernetes_command_line_removal.py new file mode 100644 index 00000000..ffe64dcc --- /dev/null +++ b/tests/launchers/kubernetes/test_kubernetes_command_line_removal.py @@ -0,0 +1,170 @@ +"""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_and_censor_it( + obj=_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_and_censor_it( + obj=_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_and_censor_it( + obj=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_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() + + kubernetes_launchers._serialize_kubernetes_object_to_compact_dict_and_censor_it( + obj=pod + ) + + assert pod.spec.containers[0].args == ["--token", _SECRET_VALUE] diff --git a/tests/test_kubernetes_launcher_error_classification.py b/tests/test_kubernetes_launcher_error_classification.py index 39d0a3d2..7f1b40bc 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 `_extract_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") 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