Add retry logic for Kubernetes 410 (Gone) errors

Co-authored-by: fabricebrito <1178901+fabricebrito@users.noreply.github.com>

Author: Copilot

calrissian/retry.py

@@ -18,6 +18,18 @@ def _is_4xx(exc) -> bool:
         return False
 
 
+def _is_retryable_4xx(exc) -> bool:
+    """
+    Check if a 4xx error is retryable.
+    410 (Gone) errors are retryable because they indicate the watch resource version is too old.
+    """
+    status = getattr(exc, "status", None)
+    try:
+        return int(status) == 410
+    except (TypeError, ValueError):
+        return False
+
+
 def retry_exponential_if_exception_type(exc_type, logger):
     """
     Decorator function that returns the tenacity @retry decorator with our commonly-used config
@@ -27,8 +39,9 @@ def retry_exponential_if_exception_type(exc_type, logger):
     """
     retry_on_type = retry_if_exception_type(exc_type)
     retry_not_4xx = retry_if_exception(lambda e: not _is_4xx(e))
+    retry_on_410 = retry_if_exception(_is_retryable_4xx)
 
-    return retry(retry=retry_on_type & retry_not_4xx,
+    return retry(retry=retry_on_type & (retry_not_4xx | retry_on_410),
             wait=wait_exponential(multiplier=RetryParameters.MULTIPLIER, min=RetryParameters.MIN, max=RetryParameters.MAX),
             stop=stop_after_attempt(RetryParameters.ATTEMPTS),
             before_sleep=before_sleep_log(logger, logging.DEBUG),

tests/test_k8s.py

@@ -196,6 +196,29 @@ def test_wait_raises_exception_when_state_is_unexpected(self, mock_watch, mock_g
         with self.assertRaisesRegex(CalrissianJobException, 'Unexpected pod container status'):
             kc.wait_for_completion()
 
+    @patch('calrissian.k8s.watch', autospec=True)
+    @patch('calrissian.k8s.KubernetesClient._extract_cpu_memory_requests')
+    def test_wait_raises_410_error_when_retries_disabled(self, mock_cpu_memory, mock_watch, mock_get_namespace, mock_client):
+        """Test that 410 errors are raised when retries are disabled (RETRY_ATTEMPTS=0)"""
+        mock_cpu_memory.return_value = ('1', '1Mi')
+        
+        mock_pod = create_autospec(V1Pod)
+        mock_pod.status.container_statuses[0].state = Mock(running=None, waiting=None, terminated=Mock(exit_code=0))
+        
+        # Stream raises 410
+        mock_watch.Watch.return_value.stream.side_effect = ApiException(status=410, reason="Expired: too old resource version")
+        mock_watch.Watch.return_value.stop = Mock()
+        
+        kc = KubernetesClient()
+        kc._set_pod(mock_pod)
+        
+        # With RETRY_ATTEMPTS=0, the 410 should be raised immediately
+        with self.assertRaises(ApiException) as context:
+            kc.wait_for_completion()
+        
+        # The 410 error should be raised (retries disabled in test env)
+        self.assertEqual(context.exception.status, 410)
+
     def test_raises_on_set_second_pod(self, mock_get_namespace, mock_client):
         kc = KubernetesClient()
         kc._set_pod(Mock())

tests/test_retry.py

@@ -252,3 +252,39 @@ def wrapped():
             wrapped()
 
         self.assertEqual(self.mock.call_count, mock_retry_parameters.ATTEMPTS)
+
+
+    @patch("calrissian.retry.RetryParameters")
+    def test_retry_on_410_gone_error(self, mock_retry_parameters):
+        """410 (Gone) IS 4xx but SHOULD retry due to special handling."""
+        self.setup_mock_retry_parameters(mock_retry_parameters)
+        self.mock.side_effect = FakeApiException(410, "Expired: too old resource version")
+
+        @retry_exponential_if_exception_type(FakeApiException, self.logger)
+        def wrapped():
+            return self.mock()
+
+        with self.assertRaisesRegex(FakeApiException, "Expired"):
+            wrapped()
+
+        # Should retry because 410 is a retryable 4xx
+        self.assertEqual(self.mock.call_count, mock_retry_parameters.ATTEMPTS)
+
+
+    @patch("calrissian.retry.RetryParameters")
+    def test_retry_on_410_eventually_succeeds(self, mock_retry_parameters):
+        """410 should retry and eventually succeed."""
+        self.setup_mock_retry_parameters(mock_retry_parameters)
+        self.mock.side_effect = [
+            FakeApiException(410, "Expired: too old resource version"),
+            FakeApiException(410, "Expired: too old resource version"),
+            "ok",
+        ]
+
+        @retry_exponential_if_exception_type(FakeApiException, self.logger)
+        def wrapped():
+            return self.mock()
+
+        result = wrapped()
+        self.assertEqual(result, "ok")
+        self.assertEqual(self.mock.call_count, 3)