fix: Kubernetes launcher - Treat an unreachable kubelet as a lost log

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://<node-ip>:10250/containerLogs/<namespace>/<pod>/main?timestamps=true\\":
+                     dial tcp <node-ip>: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,
     *,

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")