feat(orchestrator): count vanished workloads

cloud_pipelines_backend/instrumentation/metrics.py

@@ -55,6 +55,15 @@ class MetricUnit(str, enum.Enum):
     unit=MetricUnit.ERRORS,
 )
 
+execution_missing_workloads = orchestrator_meter.create_counter(
+    name="execution.missing_workloads",
+    description=(
+        "Number of execution nodes that failed because their workload"
+        " disappeared before it completed (e.g. the pod was deleted)"
+    ),
+    unit=MetricUnit.ERRORS,
+)
+
 execution_status_transition_duration = orchestrator_meter.create_histogram(
     name="execution.status_transition.duration",
     description="Duration an execution spent in a status before transitioning to the next status",

cloud_pipelines_backend/launchers/interfaces.py

@@ -37,6 +37,16 @@ def __init__(self, *args: object, is_retriable: bool = False) -> None:
         self.is_retriable = is_retriable
 
 
+class LaunchedContainerNotFoundError(LauncherError):
+    """The launched container no longer exists (e.g. its pod was deleted).
+
+    A definitive, non-retriable failure: re-reading cannot bring it back. The
+    launcher raises this so the orchestrator can tell "the workload is gone"
+    apart from "we could not reach the platform" without inspecting platform
+    error codes itself.
+    """
+
+
 @dataclasses.dataclass(kw_only=True)
 class InputArgument:
     total_size: int

cloud_pipelines_backend/launchers/kubernetes_launchers.py

@@ -2,6 +2,7 @@
 
 import copy
 import datetime
+import http
 import json
 import logging
 import os
@@ -39,10 +40,12 @@ def _launcher_error_from_api_exception(
 ) -> interfaces.LauncherError:
     """Translate a Kubernetes API error into a launcher error.
 
-    A 5xx is retriable (the API server broke or shed load); any other status is
-    a definitive failure.
+    A 404 means the workload is gone; a 5xx is retriable (the API server broke
+    or shed load); any other status is a definitive failure.
     """
     status = exception.status
+    if status == http.HTTPStatus.NOT_FOUND:
+        return interfaces.LaunchedContainerNotFoundError(f"{message}: {exception!r}")
     is_retriable = isinstance(status, int) and 500 <= status < 600
     return interfaces.LauncherError(
         f"{message}: {exception!r}", is_retriable=is_retriable

cloud_pipelines_backend/orchestrator_sql.py

@@ -222,6 +222,16 @@ def internal_process_running_executions_queue(self, session: orm.Session):
                         isinstance(ex, launcher_interfaces.LauncherError)
                         and ex.is_retriable
                     )
+                    if isinstance(
+                        ex, launcher_interfaces.LaunchedContainerNotFoundError
+                    ):
+                        app_metrics.execution_missing_workloads.add(
+                            1,
+                            attributes={
+                                "status": running_container_execution.status.value
+                            },
+                        )
+
                     error_count = (
                         self._container_execution_refresh_error_counts.get(
                             running_container_execution.id, 0

tests/test_container_execution_refresh_retries.py

@@ -28,6 +28,10 @@ def _non_retriable_error() -> launcher_interfaces.LauncherError:
     return launcher_interfaces.LauncherError("Something went definitively wrong")
 
 
+def _missing_workload_error() -> launcher_interfaces.LaunchedContainerNotFoundError:
+    return launcher_interfaces.LaunchedContainerNotFoundError("The workload is gone")
+
+
 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)
@@ -179,6 +183,30 @@ def test_disabled_flag_terminalizes_retriable_error_immediately(self) -> None:
             bts.ContainerExecutionStatus.SYSTEM_ERROR
         )
 
+    def test_missing_workload_is_counted_and_terminalizes(self) -> None:
+        session_factory = _create_launched_container_executions()
+        orchestrator = _make_orchestrator(
+            session_factory,
+            mock.MagicMock(side_effect=_missing_workload_error()),
+            max_failures=3,
+        )
+
+        with mock.patch.object(
+            orchestrator_sql.app_metrics, "execution_missing_workloads"
+        ) as missing_workloads:
+            orchestrator.internal_process_running_executions_queue(
+                session=session_factory()
+            )
+
+        missing_workloads.add.assert_called_once()
+        args, kwargs = missing_workloads.add.call_args
+        assert args == (1,)
+        assert isinstance(kwargs["attributes"]["status"], str)
+        assert kwargs["attributes"]["status"]
+        assert _only_status(session_factory) == (
+            bts.ContainerExecutionStatus.SYSTEM_ERROR
+        )
+
     def test_retriable_errors_below_budget_leave_execution_running(self) -> None:
         session_factory = _create_launched_container_executions()
         orchestrator = _make_orchestrator(

tests/test_kubernetes_launcher_error_classification.py

@@ -24,10 +24,11 @@ def test_service_unavailable_is_retriable(self) -> None:
         )
         assert error.is_retriable
 
-    def test_not_found_is_not_retriable(self) -> None:
+    def test_not_found_is_a_missing_workload_error(self) -> None:
         error = kubernetes_launchers._launcher_error_from_api_exception(
             _api_exception(404), message="Failed to refresh pod status"
         )
+        assert isinstance(error, interfaces.LaunchedContainerNotFoundError)
         assert not error.is_retriable
 
     def test_client_error_is_not_retriable(self) -> None: