feat(k8s-launcher): explain empty logs when a pod was deleted before capture

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

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"