fix(orchestrator): optionally retry retriable container-refresh failures

cloud_pipelines_backend/launchers/interfaces.py

@@ -26,7 +26,15 @@
 
 
 class LauncherError(RuntimeError):
-    pass
+    """Base class for errors raised by a launcher.
+
+    `is_retriable` marks a failure worth trying again (e.g. the backing platform
+    was unavailable) rather than a definitive error.
+    """
+
+    def __init__(self, *args: object, is_retriable: bool = False) -> None:
+        super().__init__(*args)
+        self.is_retriable = is_retriable
 
 
 @dataclasses.dataclass(kw_only=True)

cloud_pipelines_backend/launchers/kubernetes_launchers.py

@@ -595,7 +595,12 @@ def get_refreshed_launched_container_from_dict(
         launched_container = LaunchedKubernetesContainer.from_dict(
             launched_container_dict, launcher=self
         )
-        return launched_container.get_refreshed()
+        try:
+            return launched_container.get_refreshed()
+        except kubernetes.client.exceptions.ApiException as ex:
+            raise _launcher_error_from_api_exception(
+                ex, message="Failed to refresh pod status"
+            ) from ex
 
     def deserialize_launched_container_from_dict(
         self, launched_container_dict: dict
@@ -1257,7 +1262,12 @@ def get_refreshed_launched_container_from_dict(
         launched_container = LaunchedKubernetesJob.from_dict(
             launched_container_dict, launcher=self
         )
-        return launched_container.get_refreshed()
+        try:
+            return launched_container.get_refreshed()
+        except kubernetes.client.exceptions.ApiException as ex:
+            raise _launcher_error_from_api_exception(
+                ex, message="Failed to refresh job status"
+            ) from ex
 
     def deserialize_launched_container_from_dict(
         self, launched_container_dict: dict
@@ -1848,3 +1858,20 @@ def _remove_keys_with_none_values(d: dict):
             del d[k]
         if isinstance(v, dict):
             _remove_keys_with_none_values(v)
+
+
+def _launcher_error_from_api_exception(
+    exception: kubernetes.client.exceptions.ApiException,
+    *,
+    message: str,
+) -> 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.
+    """
+    status = exception.status
+    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

@@ -2,6 +2,7 @@
 import json
 import datetime
 import logging
+import os
 import time
 import traceback
 import typing
@@ -52,6 +53,7 @@ def __init__(
         output_data_purge_duration: datetime.timedelta = None,
         *,
         # Internal/experimental:
+        _max_container_execution_refresh_error_retries: int = 3,
         _max_queue_batch_size: int = 1,
         _max_queue_batch_duration: datetime.timedelta = datetime.timedelta(),
     ):
@@ -65,6 +67,10 @@ def __init__(
         self._queued_executions_queue_idle = False
         self._running_executions_queue_idle = False
         self._output_data_purge_duration = output_data_purge_duration
+        self._max_container_execution_refresh_error_retries = (
+            _max_container_execution_refresh_error_retries
+        )
+        self._container_execution_refresh_error_counts: dict[Any, int] = {}
 
         self._max_queue_batch_size = _max_queue_batch_size
         self._max_queue_batch_duration = _max_queue_batch_duration
@@ -213,6 +219,38 @@ def internal_process_running_executions_queue(self, session: orm.Session):
                 except Exception as ex:
                     _logger.exception("Error processing running container execution")
                     session.rollback()
+
+                    is_retriable = (
+                        isinstance(ex, launcher_interfaces.LauncherError)
+                        and ex.is_retriable
+                    )
+                    error_count = (
+                        self._container_execution_refresh_error_counts.get(
+                            running_container_execution.id, 0
+                        )
+                        + 1
+                    )
+                    if (
+                        is_retriable
+                        and error_count
+                        < self._max_container_execution_refresh_error_retries
+                    ):
+                        self._container_execution_refresh_error_counts[
+                            running_container_execution.id
+                        ] = error_count
+                        _logger.warning(
+                            "Could not refresh running container execution "
+                            f"({error_count} of "
+                            f"{self._max_container_execution_refresh_error_retries}"
+                            f" consecutive retriable failures): {ex!r}. Leaving "
+                            f"it in {running_container_execution.status} and "
+                            "refreshing it on a later sweep."
+                        )
+                        return True
+
+                    self._container_execution_refresh_error_counts.pop(
+                        running_container_execution.id, None
+                    )
                     running_container_execution.status = (
                         bts.ContainerExecutionStatus.SYSTEM_ERROR
                     )
@@ -238,6 +276,10 @@ def internal_process_running_executions_queue(self, session: orm.Session):
                             session=session, execution=execution_node
                         )
                     session.commit()
+                else:
+                    self._container_execution_refresh_error_counts.pop(
+                        running_container_execution.id, None
+                    )
                 finally:
                     duration_ms = (time.monotonic_ns() - start_timestamp) / 1_000_000
                     _logger.info(