diff --git a/cloud_pipelines_backend/launchers/kubernetes_launchers.py b/cloud_pipelines_backend/launchers/kubernetes_launchers.py index 25db318..b4fe7f8 100644 --- a/cloud_pipelines_backend/launchers/kubernetes_launchers.py +++ b/cloud_pipelines_backend/launchers/kubernetes_launchers.py @@ -1748,6 +1748,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]]: @@ -2190,6 +2200,43 @@ def _remove_keys_with_none_values(d: dict): _remove_keys_with_none_values(v) +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 39d0a3d..f471eb4 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")