fix(orchestrator): losing pod logs must not fail an execution

cloud_pipelines_backend/launchers/kubernetes_launchers.py

@@ -1563,6 +1563,15 @@ def _get_log_by_pod_key(self, pod_name: str) -> str | None:
                 # Kubernetes client raises kubernetes.client.exceptions.ApiException when Pod is still in PodInitializing phase
                 # 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.
+                _logger.warning(
+                    f"Pod {pod_name} no longer exists; its logs are unrecoverable."
+                )
+                return None
             raise
 
     def _get_all_logs(self) -> dict[str, str]:

cloud_pipelines_backend/orchestrator_sql.py

@@ -1065,7 +1065,16 @@ def _maybe_preload_value(
                     message=orchestration_error_message,
                 )
 
-            _retry(reloaded_launched_container.upload_log)
+            # A log-upload failure must not change a terminal status.
+            # Logs are best-effort; losing them is never a reason to fail an
+            # execution. The launcher decides what a lost log means; the
+            # orchestrator only records that the upload did not succeed.
+            try:
+                _retry(reloaded_launched_container.upload_log)
+            except Exception as ex:
+                _logger.exception(
+                    f"! Error during `LaunchedContainer.upload_log` call: {ex}."
+                )
             # Skip downstream executions
             for execution_node in execution_nodes:
                 execution_node.container_execution_status = (

tests/test_kubernetes_launcher_error_classification.py

@@ -1,6 +1,9 @@
 """Tests for translating Kubernetes API errors into launcher errors."""
 
+from unittest import mock
+
 import kubernetes.client.exceptions
+import pytest
 
 from cloud_pipelines_backend.launchers import interfaces
 from cloud_pipelines_backend.launchers import kubernetes_launchers
@@ -36,3 +39,53 @@ def test_client_error_is_not_retriable(self) -> None:
             _api_exception(403), message="Failed to refresh pod status"
         )
         assert not error.is_retriable
+
+
+def _make_job_with_pod_log_error(
+    api_exception: kubernetes.client.exceptions.ApiException,
+) -> tuple[kubernetes_launchers.LaunchedKubernetesJob, mock.MagicMock]:
+    """A ``LaunchedKubernetesJob`` whose pod-log reads raise ``api_exception``."""
+    job = kubernetes_launchers.LaunchedKubernetesJob(
+        job_name="job",
+        namespace="ns",
+        output_uris={},
+        log_uri="file:///tmp/log",
+        debug_job=mock.MagicMock(),
+        launcher=mock.MagicMock(_request_timeout=10),
+    )
+    core_api = mock.MagicMock()
+    core_api.read_namespaced_pod_log.side_effect = api_exception
+    return job, core_api
+
+
+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.
+    """
+
+    def test_not_found_returns_none(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
+
+    def test_forbidden_reraises(self) -> None:
+        job, core_api = _make_job_with_pod_log_error(_api_exception(403))
+        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_server_error_reraises(self) -> None:
+        job, core_api = _make_job_with_pod_log_error(_api_exception(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")

tests/test_orchestrator_failed_log_upload.py

@@ -0,0 +1,164 @@
+"""A log-upload failure on the FAILED path must not escalate to SYSTEM_ERROR.
+
+Production scenario: the cluster-autoscaler deletes a Job's pod mid-run. The Job
+still reports Failed, so the orchestrator has everything it needs to record a
+clean FAILED. But ``upload_log`` then reads logs for the remembered (now deleted)
+pod, the API 404s, and -- if that call is unguarded -- the exception propagates
+into the outer handler, which overwrites FAILED with SYSTEM_ERROR and orphans the
+downstream subtree. Losing logs must never change a terminal status.
+"""
+
+import datetime
+from typing import Callable
+from unittest import mock
+
+from sqlalchemy import orm
+from sqlalchemy import sql
+
+from cloud_pipelines_backend import api_server_sql
+from cloud_pipelines_backend import backend_types_sql as bts
+from cloud_pipelines_backend import component_structures as structures
+from cloud_pipelines_backend import database_ops
+from cloud_pipelines_backend import orchestrator_sql
+from cloud_pipelines_backend.launchers import interfaces as launcher_interfaces
+
+_TS = datetime.datetime(2026, 1, 1, 12, 0, 0, tzinfo=datetime.timezone.utc)
+_LAUNCHER_DATA = {"pod": "pod-0"}
+
+
+def _create_session_factory() -> Callable[[], orm.Session]:
+    db_engine = database_ops.create_db_engine_and_migrate_db(database_uri="sqlite://")
+    return lambda: orm.Session(bind=db_engine)
+
+
+def _container_component(
+    *, inputs: list[str] | None = None, outputs: list[str] | None = None
+) -> structures.ComponentSpec:
+    return structures.ComponentSpec(
+        inputs=[structures.InputSpec(name=name) for name in (inputs or [])],
+        outputs=[structures.OutputSpec(name=name) for name in (outputs or [])],
+        implementation=structures.ContainerImplementation(
+            container=structures.ContainerSpec(image="python")
+        ),
+    )
+
+
+def _upstream_downstream_pipeline() -> structures.TaskSpec:
+    """``upstream`` (produces ``out``) feeding ``downstream`` (consumes ``in``)."""
+    pipeline_spec = structures.ComponentSpec(
+        implementation=structures.GraphImplementation(
+            graph=structures.GraphSpec(
+                tasks={
+                    "upstream": structures.TaskSpec(
+                        component_ref=structures.ComponentReference(
+                            spec=_container_component(outputs=["out"])
+                        ),
+                    ),
+                    "downstream": structures.TaskSpec(
+                        component_ref=structures.ComponentReference(
+                            spec=_container_component(inputs=["in"])
+                        ),
+                        arguments={
+                            "in": structures.TaskOutputArgument(
+                                task_output=structures.TaskOutputReference(
+                                    task_id="upstream", output_name="out"
+                                )
+                            )
+                        },
+                    ),
+                }
+            )
+        ),
+    )
+    return structures.TaskSpec(
+        component_ref=structures.ComponentReference(spec=pipeline_spec)
+    )
+
+
+def _get_execution_node(session: orm.Session, task_id: str) -> bts.ExecutionNode:
+    node = session.scalar(
+        sql.select(bts.ExecutionNode).where(
+            bts.ExecutionNode.task_id_in_parent_execution == task_id
+        )
+    )
+    assert node is not None, f"No execution node found for task_id={task_id!r}"
+    return node
+
+
+def _pending_launched_container() -> mock.MagicMock:
+    return mock.MagicMock(
+        status=launcher_interfaces.ContainerStatus.PENDING,
+        to_dict=lambda: dict(_LAUNCHER_DATA),
+    )
+
+
+def _failed_launched_container_that_loses_logs() -> mock.MagicMock:
+    """A Job that Failed cleanly but whose ``upload_log`` 404s on a deleted pod."""
+    return mock.MagicMock(
+        status=launcher_interfaces.ContainerStatus.FAILED,
+        exit_code=1,
+        started_at=_TS,
+        ended_at=_TS,
+        launcher_error_message=None,
+        to_dict=lambda: dict(_LAUNCHER_DATA),
+        upload_log=mock.MagicMock(
+            side_effect=RuntimeError("read_namespaced_pod_log 404: pod deleted")
+        ),
+    )
+
+
+def test_failed_execution_survives_log_upload_failure() -> None:
+    session_factory = _create_session_factory()
+    api_server_sql.PipelineRunsApiService_Sql().create(
+        session=session_factory(),
+        root_task=_upstream_downstream_pipeline(),
+        created_by="user1",
+    )
+
+    launcher = mock.MagicMock()
+    launcher.launch_container_task.side_effect = (
+        lambda *a, **kw: _pending_launched_container()
+    )
+    launcher.deserialize_launched_container_from_dict.side_effect = (
+        lambda data: _pending_launched_container()
+    )
+    failed_container = _failed_launched_container_that_loses_logs()
+    launcher.get_refreshed_launched_container_from_dict.side_effect = (
+        lambda data: failed_container
+    )
+
+    orchestrator = orchestrator_sql.OrchestratorService_Sql(
+        session_factory=session_factory,
+        launcher=launcher,
+        storage_provider=mock.MagicMock(),
+        data_root_uri="file:///tmp/artifacts",
+        logs_root_uri="file:///tmp/logs",
+    )
+
+    # Launch the queued executions: `upstream` becomes PENDING, `downstream` stays
+    # WAITING_FOR_UPSTREAM (it depends on upstream's output).
+    session = session_factory()
+    for _ in range(20):
+        if not orchestrator.internal_process_queued_executions_queue(session=session):
+            break
+
+    # Refresh the running execution. `upstream` comes back FAILED, and its log
+    # upload raises -- `_retry` burns its attempts and re-raises. `time.sleep` is
+    # patched out so the retries do not slow the test down.
+    with mock.patch.object(orchestrator_sql.time, "sleep"):
+        orchestrator.internal_process_running_executions_queue(
+            session=session_factory()
+        )
+
+    # The unguarded call would have re-raised into the outer handler and marked
+    # upstream SYSTEM_ERROR; the guard keeps it FAILED with normal skipping.
+    assert failed_container.upload_log.called
+    check_session = session_factory()
+    upstream = _get_execution_node(check_session, "upstream")
+    downstream = _get_execution_node(check_session, "downstream")
+    assert (
+        upstream.container_execution_status == bts.ContainerExecutionStatus.FAILED
+    ), "a lost log must not turn FAILED into SYSTEM_ERROR"
+    assert (
+        downstream.container_execution_status == bts.ContainerExecutionStatus.SKIPPED
+    ), "downstream of a FAILED node must still be skipped"