From a6b2d5bf1db71cee26d0a7b01350e048734763c7 Mon Sep 17 00:00:00 2001 From: Morgan Wowk Date: Wed, 12 Aug 2026 13:23:28 -0700 Subject: [PATCH] feat(k8s-launcher): explain empty logs when a pod was deleted before capture Signed-off-by: Morgan Wowk --- .../launchers/kubernetes_launchers.py | 71 ++++++++--- ...ubernetes_launcher_error_classification.py | 115 +++++++++++++++++- 2 files changed, 165 insertions(+), 21 deletions(-) diff --git a/cloud_pipelines_backend/launchers/kubernetes_launchers.py b/cloud_pipelines_backend/launchers/kubernetes_launchers.py index 9b1a661..9bbd458 100644 --- a/cloud_pipelines_backend/launchers/kubernetes_launchers.py +++ b/cloud_pipelines_backend/launchers/kubernetes_launchers.py @@ -34,6 +34,28 @@ _MAIN_CONTAINER_NAME = "main" +class _PodDeleted: + """Sentinel: a Pod no longer exists, so its logs can never be retrieved.""" + + +# Returned by `LaunchedKubernetesJob._get_log_by_pod_key` when a Pod read 404s. +# Distinct from `None`, which means the Pod exists but has no logs available yet +# (HTTP 400, still initializing). +_POD_DELETED = _PodDeleted() + +# Persisted in place of an empty log when the only reason no logs were captured +# is that the Pod(s) were deleted before we could read them -- e.g. the +# cluster-autoscaler evicting the node, or Kubernetes garbage-collecting a +# finished Pod. Without this the UI shows a blank pane indistinguishable from a +# task that simply produced no output. +_LOGS_UNAVAILABLE_POD_DELETED_MESSAGE = ( + "Logs are unavailable: the pod was deleted before its logs could be " + "captured (for example, autoscaler eviction or Kubernetes garbage " + "collection). The task's final status still reflects what happened; the " + "logs themselves could not be captured." +) + + # Kubernetes annotation keys. (Has strict naming policy. Single slash only etc.) _CLOUD_PIPELINES_KUBERNETES_ANNOTATION_KEY = "cloud-pipelines.net" _KUBERNETES_LAUNCHER_ANNOTATION_KEY = "cloud-pipelines.net/launchers.kubernetes" @@ -1534,7 +1556,7 @@ def get_refreshed(self) -> "LaunchedKubernetesJob": new_launched_container._debug_pods = pod_map return new_launched_container - def _get_log_by_pod_key(self, pod_name: str) -> str | None: + def _get_log_by_pod_key(self, pod_name: str) -> str | None | _PodDeleted: launcher = self._get_launcher() core_api_client = k8s_client_lib.CoreV1Api(api_client=launcher._api_client) try: @@ -1564,23 +1586,34 @@ def _get_log_by_pod_key(self, pod_name: str) -> str | None: # See https://github.com/TangleML/tangle/issues/139 return None if ex.status == http.HTTPStatus.NOT_FOUND: - # The Pod is gone (e.g. deleted by the cluster-autoscaler mid-run). - # `_debug_pods` deliberately retains vanished Pods, so this key would - # 404 on every subsequent read. A deleted Pod means "no logs", not an - # error, so return None instead of re-raising. + # The Pod is gone (e.g. evicted by the cluster-autoscaler or + # garbage-collected mid-run). `_debug_pods` deliberately retains + # vanished Pods, so this key would 404 on every subsequent read. A + # deleted Pod means "no logs", not an error, so report it as such + # instead of re-raising. _logger.warning( f"Pod {pod_name} no longer exists; its logs are unrecoverable." ) - return None + return _POD_DELETED raise - def _get_all_logs(self) -> dict[str, str]: - logs = {} + def _get_all_logs(self) -> tuple[dict[str, str], list[str]]: + """Logs by pod key, plus the keys of pods that no longer exist. + + A Pod can vanish before its logs are captured -- the cluster-autoscaler + evicting its node, or Kubernetes garbage-collecting a finished Pod. Such + pods are reported separately so callers can tell "the pod is gone" apart + from "the pod produced no output". + """ + logs: dict[str, str] = {} + deleted_pod_keys: list[str] = [] for pod_key, pod in self._debug_pods.items(): log = self._get_log_by_pod_key(pod.metadata.name) - if log: + if log is _POD_DELETED: + deleted_pod_keys.append(pod_key) + elif log: logs[pod_key] = log - return logs + return logs, deleted_pod_keys def _merge_logs(self, logs: dict[str, str | None]) -> str: if not logs: @@ -1605,13 +1638,21 @@ def _merge_logs(self, logs: dict[str, str | None]) -> str: return "\n".join(all_log_lines) + "\n" def get_log(self) -> str: - all_logs = self._get_all_logs() - merged_log = self._merge_logs(all_logs) + logs, deleted_pod_keys = self._get_all_logs() + merged_log = self._merge_logs(logs) + # Only substitute the notice when we recovered nothing *and* a Pod was + # deleted -- an empty log from a Pod that still exists is a real result. + if not merged_log and deleted_pod_keys: + return _LOGS_UNAVAILABLE_POD_DELETED_MESSAGE return merged_log def upload_log(self): - all_logs = self._get_all_logs() - merged_log = self._merge_logs(all_logs) + logs, deleted_pod_keys = self._get_all_logs() + merged_log = self._merge_logs(logs) + # Only substitute the notice when we recovered nothing *and* a Pod was + # deleted -- an empty log from a Pod that still exists is a real result. + if not merged_log and deleted_pod_keys: + merged_log = _LOGS_UNAVAILABLE_POD_DELETED_MESSAGE # Uploading the merged log launcher = self._get_launcher() @@ -1620,7 +1661,7 @@ def upload_log(self): # Uploading per-pod logs. # It's not ideal to construct new URIs ourselves. But Orchestrator only supports single log per container execution. - for pod_key, log in all_logs.items(): + for pod_key, log in logs.items(): uri_writer = launcher._storage_provider.make_uri( self._log_uri + f".{pod_key}" ).get_writer() diff --git a/tests/test_kubernetes_launcher_error_classification.py b/tests/test_kubernetes_launcher_error_classification.py index 39d0a3d..2af6d8e 100644 --- a/tests/test_kubernetes_launcher_error_classification.py +++ b/tests/test_kubernetes_launcher_error_classification.py @@ -61,18 +61,18 @@ def _make_job_with_pod_log_error( class TestGetLogByPodKey: """``_get_log_by_pod_key`` maps a vanished Pod (404) to "no logs", not an error. - A Pod deleted mid-run (e.g. by the cluster-autoscaler) is retained in - ``_debug_pods``, so every later read 404s. That must return ``None`` rather - than propagate -- otherwise it fails the whole execution. Genuine client and - server errors must still propagate. + A Pod deleted mid-run (e.g. by the cluster-autoscaler or garbage collection) + is retained in ``_debug_pods``, so every later read 404s. That must report + the Pod as deleted rather than propagate -- otherwise it fails the whole + execution. Genuine client and server errors must still propagate. """ - def test_not_found_returns_none(self) -> None: + def test_not_found_returns_pod_deleted_sentinel(self) -> None: job, core_api = _make_job_with_pod_log_error(_api_exception(404)) 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 + assert job._get_log_by_pod_key("pod-0") is kubernetes_launchers._POD_DELETED def test_forbidden_reraises(self) -> None: job, core_api = _make_job_with_pod_log_error(_api_exception(403)) @@ -89,3 +89,106 @@ def test_server_error_reraises(self) -> None: ): with pytest.raises(kubernetes.client.exceptions.ApiException): job._get_log_by_pod_key("pod-0") + + +def _pod(name: str) -> mock.MagicMock: + pod = mock.MagicMock() + pod.metadata.name = name + return pod + + +def _log_response(text: str) -> mock.MagicMock: + """A ``read_namespaced_pod_log`` response as used with ``_preload_content=False``.""" + response = mock.MagicMock() + response.data = text.encode("utf-8") + return response + + +def _make_job_reading_pods( + debug_pods: dict[str, mock.MagicMock], +) -> tuple[kubernetes_launchers.LaunchedKubernetesJob, dict[str, str], mock.MagicMock]: + """A Job over ``debug_pods`` plus a dict capturing everything ``upload_log`` writes.""" + uploads: dict[str, str] = {} + + def _make_uri(uri: str) -> mock.MagicMock: + writer = mock.MagicMock() + writer.upload_from_text.side_effect = lambda text, u=uri: uploads.__setitem__( + u, text + ) + accessor = mock.MagicMock() + accessor.get_writer.return_value = writer + return accessor + + launcher = mock.MagicMock(_request_timeout=10) + launcher._storage_provider.make_uri.side_effect = _make_uri + job = kubernetes_launchers.LaunchedKubernetesJob( + job_name="job", + namespace="ns", + output_uris={}, + log_uri="file:///tmp/log", + debug_job=mock.MagicMock(), + debug_pods=debug_pods, + launcher=launcher, + ) + return job, uploads, launcher + + +class TestVanishedPodLogPlaceholder: + """When a Pod is deleted before its logs are captured, the persisted log is a + human-readable notice rather than a blank -- but only when that is the *sole* + reason there are no logs. A Pod that exists and printed nothing stays empty, + and any recovered logs are persisted verbatim. + """ + + def test_deleted_pod_persists_notice(self) -> None: + job, uploads, _ = _make_job_reading_pods({"0": _pod("pod-0")}) + core_api = mock.MagicMock() + core_api.read_namespaced_pod_log.side_effect = _api_exception(404) + with mock.patch.object( + kubernetes_launchers.k8s_client_lib, "CoreV1Api", return_value=core_api + ): + job.upload_log() + assert job.get_log() == ( + kubernetes_launchers._LOGS_UNAVAILABLE_POD_DELETED_MESSAGE + ) + assert ( + uploads["file:///tmp/log"] + == kubernetes_launchers._LOGS_UNAVAILABLE_POD_DELETED_MESSAGE + ) + # The notice is not written as a per-pod log. + assert list(uploads) == ["file:///tmp/log"] + + def test_existing_pod_with_empty_output_stays_empty(self) -> None: + job, uploads, _ = _make_job_reading_pods({"0": _pod("pod-0")}) + core_api = mock.MagicMock() + core_api.read_namespaced_pod_log.return_value = _log_response("") + with mock.patch.object( + kubernetes_launchers.k8s_client_lib, "CoreV1Api", return_value=core_api + ): + job.upload_log() + assert job.get_log() == "" + assert uploads["file:///tmp/log"] == "" + + def test_recovered_logs_are_not_replaced(self) -> None: + job, uploads, _ = _make_job_reading_pods( + {"0": _pod("pod-0"), "1": _pod("pod-1")} + ) + core_api = mock.MagicMock() + + def _read(name: str, **kwargs: object) -> mock.MagicMock: + if name == "pod-1": # This Pod is gone... + raise _api_exception(404) + return _log_response("2026-01-01T00:00:00Z hello\n") # ...but pod-0 logged. + + core_api.read_namespaced_pod_log.side_effect = _read + with mock.patch.object( + kubernetes_launchers.k8s_client_lib, "CoreV1Api", return_value=core_api + ): + job.upload_log() + # Partial logs win over the notice: we show what we have. + assert "hello" in uploads["file:///tmp/log"] + assert ( + kubernetes_launchers._LOGS_UNAVAILABLE_POD_DELETED_MESSAGE + not in uploads["file:///tmp/log"] + ) + assert uploads["file:///tmp/log.0"] == "2026-01-01T00:00:00Z hello\n"