diff --git a/ocp_resources/constants.py b/ocp_resources/constants.py index b08c8d07a1..7c80dca161 100644 --- a/ocp_resources/constants.py +++ b/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 diff --git a/ocp_resources/resource.py b/ocp_resources/resource.py index 7601256bd8..8a5bca32df 100644 --- a/ocp_resources/resource.py +++ b/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,7 +915,9 @@ 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. @@ -910,18 +925,32 @@ def api_request(self, method, action, url, **params): 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, ) diff --git a/ocp_resources/virtual_machine.py b/ocp_resources/virtual_machine.py index 0e5d70907a..8b5e12d130 100644 --- a/ocp_resources/virtual_machine.py +++ b/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() diff --git a/ocp_resources/virtual_machine_instance.py b/ocp_resources/virtual_machine_instance.py index bf215f986d..51ad709754 100644 --- a/ocp_resources/virtual_machine_instance.py +++ b/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") diff --git a/tests/unittests/test_resource_retry.py b/tests/unittests/test_resource_retry.py new file mode 100644 index 0000000000..2980bf2b27 --- /dev/null +++ b/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}