Fix: add retry logic in 4.15

ocp_resources/constants.py

@@ -22,7 +22,10 @@
 PROTOCOL_ERROR_EXCEPTION_DICT = {ProtocolError: []}
 NOT_FOUND_ERROR_EXCEPTION_DICT = {NotFoundError: []}
 
+TIMEOUT_1SEC = 1
+TIMEOUT_5SEC = 5
 TIMEOUT_10SEC = 10
+TIMEOUT_30SEC = 30
 TIMEOUT_1MINUTE = 60
 TIMEOUT_2MINUTES = 2 * 60
 TIMEOUT_4MINUTES = 4 * 60

ocp_resources/resource.py

@@ -1,11 +1,15 @@
+from __future__ import annotations
+
 import contextlib
 import copy
 import json
 import os
 import re
 import sys
+from collections.abc import Callable
 from io import StringIO
 from signal import SIGINT, signal
+from typing import Any, Dict, List
 
 import kubernetes
 import yaml
@@ -26,8 +30,11 @@
     NOT_FOUND_ERROR_EXCEPTION_DICT,
     PROTOCOL_ERROR_EXCEPTION_DICT,
     TIMEOUT_1MINUTE,
+    TIMEOUT_1SEC,
     TIMEOUT_4MINUTES,
+    TIMEOUT_5SEC,
     TIMEOUT_10SEC,
+    TIMEOUT_30SEC,
 )
 from ocp_resources.event import Event
 from timeout_sampler import (
@@ -767,11 +774,17 @@ def update_replace(self, resource_dict):
         self.api.replace(body=resource_dict, name=self.name, namespace=self.namespace)
 
     @staticmethod
-    def retry_cluster_exceptions(func, exceptions_dict=DEFAULT_CLUSTER_RETRY_EXCEPTIONS, **kwargs):
+    def retry_cluster_exceptions(
+        func: Callable,
+        exceptions_dict: Dict[type[Exception], List[str]] = DEFAULT_CLUSTER_RETRY_EXCEPTIONS,
+        timeout: int = TIMEOUT_10SEC,
+        sleep_time: int = 1,
+        **kwargs: Any,
+    ) -> Any:
         try:
             sampler = TimeoutSampler(
-                wait_timeout=TIMEOUT_10SEC,
-                sleep=1,
+                wait_timeout=timeout,
+                sleep=sleep_time,
                 func=func,
                 print_log=False,
                 exceptions_dict=exceptions_dict,
@@ -902,26 +915,42 @@ def wait_for_condition(self, condition, status, timeout=300):
                     if cond["type"] == condition and cond["status"] == status:
                         return
 
-    def api_request(self, method, action, url, **params):
+    def api_request(
+        self, method: str, action: str, url: str, retry_params: dict[str, int] | None = None, **params: Any
+    ) -> dict[str, Any]:
         """
         Handle API requests to resource.
 
         Args:
             method (str): Request method (GET/PUT etc.).
             action (str): Action to perform (stop/start/guestosinfo etc.).
             url (str): URL of resource.
+            retry_params (dict): dict of timeout and sleep_time values for retrying the api request call
 
         Returns:
            data(dict): response data
 
         """
         client = self.privileged_client or self.client
-        response = client.client.request(
-            method=method,
-            url=f"{url}/{action}",
-            headers=self.client.configuration.api_key,
-            **params,
-        )
+
+        api_request_params = {
+            "url": f"{url}/{action}",
+            "method": method,
+            "headers": self.client.configuration.api_key,
+        }
+        if retry_params:
+            response = self.retry_cluster_exceptions(
+                func=client.client.request,
+                timeout=retry_params.get("timeout", TIMEOUT_10SEC),
+                sleep_time=retry_params.get("sleep_time", TIMEOUT_1SEC),
+                **api_request_params,
+                **params,
+            )
+        else:
+            response = client.client.request(
+                **api_request_params,
+                **params,
+            )
 
         try:
             return json.loads(response.data)
@@ -931,7 +960,7 @@ def api_request(self, method, action, url, **params):
     def wait_for_conditions(self):
         timeout_watcher = TimeoutWatch(timeout=30)
         for sample in TimeoutSampler(
-            wait_timeout=30,
+            wait_timeout=TIMEOUT_30SEC,
             sleep=1,
             func=lambda: self.exists,
         ):
@@ -1377,4 +1406,6 @@ def _apply_patches_sampler(self, patches, action_text, action):
             patches=patches,
             action_text=action_text,
             action=action,
+            timeout=TIMEOUT_30SEC,
+            sleep_time=TIMEOUT_5SEC,
         )

ocp_resources/virtual_machine.py

@@ -1,10 +1,13 @@
-# -*- coding: utf-8 -*-
+from __future__ import annotations
 
+from typing import Any
 
 from ocp_resources.constants import (
     DEFAULT_CLUSTER_RETRY_EXCEPTIONS,
     PROTOCOL_ERROR_EXCEPTION_DICT,
     TIMEOUT_4MINUTES,
+    TIMEOUT_5SEC,
+    TIMEOUT_30SEC,
 )
 from ocp_resources.resource import NamespacedResource
 from timeout_sampler import TimeoutSampler
@@ -72,8 +75,17 @@ def _subresource_api_url(self):
             f"namespaces/{self.namespace}/virtualmachines/{self.name}"
         )
 
-    def api_request(self, method, action, **params):
-        return super().api_request(method=method, action=action, url=self._subresource_api_url, **params)
+    def api_request(
+        self, method: str, action: str, url: str = "", retry_params: dict[str, int] | None = None, **params: Any
+    ) -> dict[str, Any]:
+        default_vm_api_request_retry_params: dict[str, int] = {"timeout": TIMEOUT_30SEC, "sleep_time": TIMEOUT_5SEC}
+        return super().api_request(
+            method=method,
+            action=action,
+            url=url or self._subresource_api_url,
+            retry_params=retry_params or default_vm_api_request_retry_params,
+            **params,
+        )
 
     def to_dict(self):
         super().to_dict()

ocp_resources/virtual_machine_instance.py

@@ -1,9 +1,12 @@
+from __future__ import annotations
+
 import shlex
+from typing import Any
 
 import xmltodict
 from kubernetes.dynamic.exceptions import ResourceNotFoundError
 
-from ocp_resources.constants import PROTOCOL_ERROR_EXCEPTION_DICT, TIMEOUT_4MINUTES
+from ocp_resources.constants import PROTOCOL_ERROR_EXCEPTION_DICT, TIMEOUT_4MINUTES, TIMEOUT_5SEC, TIMEOUT_30SEC
 from ocp_resources.node import Node
 from ocp_resources.pod import Pod
 from ocp_resources.resource import NamespacedResource
@@ -48,8 +51,17 @@ def _subresource_api_url(self):
             f"namespaces/{self.namespace}/virtualmachineinstances/{self.name}"
         )
 
-    def api_request(self, method, action, **params):
-        return super().api_request(method=method, action=action, url=self._subresource_api_url, **params)
+    def api_request(
+        self, method: str, action: str, url: str = "", retry_params: dict[str, int] | None = None, **params: Any
+    ) -> dict[str, Any]:
+        default_vmi_api_request_retry_params: dict[str, int] = {"timeout": TIMEOUT_30SEC, "sleep_time": TIMEOUT_5SEC}
+        return super().api_request(
+            method=method,
+            action=action,
+            url=url or self._subresource_api_url,
+            retry_params=retry_params or default_vmi_api_request_retry_params,
+            **params,
+        )
 
     def pause(self, timeout=TIMEOUT_4MINUTES, wait=False):
         self.api_request(method="PUT", action="pause")

tests/unittests/test_resource_retry.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+from ocp_resources.constants import TIMEOUT_5SEC, TIMEOUT_30SEC
+from ocp_resources.resource import Resource
+from ocp_resources.virtual_machine import VirtualMachine
+from ocp_resources.virtual_machine_instance import VirtualMachineInstance
+
+
+class TestApiRequestRetry:
+    @staticmethod
+    def _resource_with_mocked_client(
+        request_side_effect: list[Any],
+    ) -> tuple[Resource, MagicMock]:
+        # Build a Resource without running __init__ and wire up a mocked dynamic client.
+        resource = object.__new__(Resource)
+        client = MagicMock()
+        client.configuration.api_key = {}
+        client.client.request.side_effect = request_side_effect
+        resource.client = client
+        resource.privileged_client = None
+        return resource, client
+
+    def test_api_request_retries_transient_cluster_exception(self) -> None:
+        response = MagicMock()
+        response.data = '{"status": "ok"}'
+        # First call raises a transient cluster error (in DEFAULT_CLUSTER_RETRY_EXCEPTIONS),
+        # the retry then succeeds.
+        resource, client = self._resource_with_mocked_client(
+            request_side_effect=[ConnectionResetError("transient"), response]
+        )
+
+        result = resource.api_request(
+            method="PUT",
+            action="start",
+            url="https://api.example/vm",
+            retry_params={"timeout": TIMEOUT_5SEC, "sleep_time": 1},
+        )
+
+        assert result == {"status": "ok"}
+        assert client.client.request.call_count == 2
+
+    def test_api_request_no_retry_by_default(self) -> None:
+        response = MagicMock()
+        response.data = '{"status": "ok"}'
+        resource, client = self._resource_with_mocked_client(request_side_effect=[response])
+
+        result = resource.api_request(method="GET", action="guestosinfo", url="https://api.example/vm")
+
+        assert result == {"status": "ok"}
+        assert client.client.request.call_count == 1
+
+
+class TestVirtualMachineDefaultRetryParams:
+    def test_vm_api_request_passes_default_retry_params(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        captured: dict[str, Any] = {}
+
+        def fake_api_request(
+            self: Any, method: str, action: str, url: str, retry_params: dict[str, int] | None = None, **params: Any
+        ) -> dict[str, Any]:
+            captured["retry_params"] = retry_params
+            return {}
+
+        monkeypatch.setattr(target=Resource, name="api_request", value=fake_api_request)
+        vm = object.__new__(VirtualMachine)
+        vm.api_request(method="PUT", action="start", url="https://api.example/vm")
+
+        assert captured["retry_params"] == {"timeout": TIMEOUT_30SEC, "sleep_time": TIMEOUT_5SEC}
+
+    def test_vmi_api_request_passes_default_retry_params(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        captured: dict[str, Any] = {}
+
+        def fake_api_request(
+            self: Any, method: str, action: str, url: str, retry_params: dict[str, int] | None = None, **params: Any
+        ) -> dict[str, Any]:
+            captured["retry_params"] = retry_params
+            return {}
+
+        monkeypatch.setattr(target=Resource, name="api_request", value=fake_api_request)
+        vmi = object.__new__(VirtualMachineInstance)
+        vmi.api_request(method="PUT", action="pause", url="https://api.example/vmi")
+
+        assert captured["retry_params"] == {"timeout": TIMEOUT_30SEC, "sleep_time": TIMEOUT_5SEC}