diff --git a/cloud_pipelines_backend/launchers/kubernetes_launchers.py b/cloud_pipelines_backend/launchers/kubernetes_launchers.py index ffa3561..9b1a661 100644 --- a/cloud_pipelines_backend/launchers/kubernetes_launchers.py +++ b/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]: diff --git a/cloud_pipelines_backend/orchestrator_sql.py b/cloud_pipelines_backend/orchestrator_sql.py index 8203029..8dfe13b 100644 --- a/cloud_pipelines_backend/orchestrator_sql.py +++ b/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 = ( diff --git a/tests/test_kubernetes_launcher_error_classification.py b/tests/test_kubernetes_launcher_error_classification.py index cd812f2..39d0a3d 100644 --- a/tests/test_kubernetes_launcher_error_classification.py +++ b/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") diff --git a/tests/test_orchestrator_failed_log_upload.py b/tests/test_orchestrator_failed_log_upload.py new file mode 100644 index 0000000..6353516 --- /dev/null +++ b/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"