From 90818cdeb7796bc0009d91655d58135317526e0f Mon Sep 17 00:00:00 2001 From: Adam Cinko Date: Tue, 25 Aug 2026 15:44:31 +0200 Subject: [PATCH] Add retry logic for api_request() in v4.15 (backport of #2240, #2217) Backport the api_request() retry mechanism from v4.16 to fix CNV-86678, where VM/VMI subresource actions (start/stop/restart/pause/migrate/ guestosinfo) propagated transient cluster errors (e.g. "Internal error occurred: failed calling webhook") instead of retrying. - constants.py: add TIMEOUT_1SEC, TIMEOUT_5SEC, TIMEOUT_30SEC. - resource.py: add 'from __future__ import annotations'; extend retry_cluster_exceptions() with timeout/sleep_time params; wrap api_request()'s client request in retry_cluster_exceptions via an optional retry_params; pass 30s/5s in _apply_patches_sampler. - virtual_machine.py / virtual_machine_instance.py: override api_request() to default retry_params to 30s timeout / 5s sleep. Co-Authored-By: Claude Opus 4.8 --- ocp_resources/constants.py | 3 + ocp_resources/resource.py | 84 +++++++++++++++-------- ocp_resources/virtual_machine.py | 18 ++++- ocp_resources/virtual_machine_instance.py | 18 ++++- tests/unittests/test_utils.py | 1 - 5 files changed, 89 insertions(+), 35 deletions(-) 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..7098fe75c8 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 import kubernetes import yaml @@ -13,28 +17,31 @@ from kubernetes.dynamic import DynamicClient from kubernetes.dynamic.exceptions import ( ConflictError, + ForbiddenError, MethodNotAllowedError, NotFoundError, - ForbiddenError, ) from kubernetes.dynamic.resource import ResourceField from packaging.version import Version from simple_logger.logger import get_logger +from timeout_sampler import ( + TimeoutExpiredError, + TimeoutSampler, + TimeoutWatch, +) from ocp_resources.constants import ( DEFAULT_CLUSTER_RETRY_EXCEPTIONS, 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 ( - TimeoutExpiredError, - TimeoutSampler, - TimeoutWatch, -) from ocp_resources.utils import skip_existing_resource_creation_teardown LOGGER = get_logger(name=__name__) @@ -124,7 +131,7 @@ def parse(self, vstring): with contextlib.suppress(ValueError): components[idx] = int(obj) - errmsg = f"version '{vstring}' does not conform to kubernetes api versioning" " guidelines" + errmsg = f"version '{vstring}' does not conform to kubernetes api versioning guidelines" if len(components) not in (2, 4) or components[0] != "v" or not isinstance(components[1], int): raise ValueError(errmsg) @@ -352,9 +359,7 @@ def __init__( """ self.api_group = api_group or self.api_group if not self.api_group and not self.api_version: - raise NotImplementedError( - "Subclasses of Resource require self.api_group or self.api_version to" " be defined" - ) + raise NotImplementedError("Subclasses of Resource require self.api_group or self.api_version to be defined") self.namespace = None self.name = name self.client = client @@ -767,11 +772,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 +913,41 @@ 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): - """ - Handle API requests to resource. + 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. + method: HTTP method (e.g. ``GET``, ``PUT``). + action: Subresource action to perform (e.g. ``start``, ``stop``, ``guestosinfo``). + url: Base URL of the resource. + retry_params: Optional timeout and sleep_time values for retrying the API request call. + **params: Additional keyword arguments forwarded to the underlying HTTP request. Returns: - data(dict): response data - + Parsed JSON response data, or raw response data when JSON decoding fails. """ 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 +957,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 +1403,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_utils.py b/tests/unittests/test_utils.py index 8431d57506..0bc0a527a1 100644 --- a/tests/unittests/test_utils.py +++ b/tests/unittests/test_utils.py @@ -1,7 +1,6 @@ import re import pytest - from timeout_sampler import TimeoutExpiredError, TimeoutSampler