rest: apply client-go retry generation patches

Signed-off-by: Dr. Stefan Schimanski <[email protected]>

Author: sttts

kubernetes/aio/client/configuration.py

@@ -355,6 +355,22 @@ def __init__(
         self.retries = retries
         """Retry configuration
         """
+        self.client_go_retries = False
+        """Enable Kubernetes client-go-compatible retry semantics.
+
+           When enabled, GET and HEAD requests retry Retry-After responses.
+           The retry ceiling is read from ``retries`` when set; otherwise it
+           follows the client-go default of at most 10 retries.
+        """
+        self.client_go_retry_backoff = None
+        """Backoff for Kubernetes client-go-compatible retries.
+
+           If unset, client-go-compatible GET and HEAD retries use the
+           client-go default retry ceiling with no additional client-side
+           delay beyond Retry-After. When set, ``retries`` still overrides
+           the retry ceiling if it is not None.
+        """
+
         self.trace_configs = trace_configs
         """aiohttp.TraceConfig list forwarded to ClientSession for tracing.
         """

kubernetes/aio/client/rest.py

@@ -21,6 +21,11 @@
 import aiohttp
 import aiohttp_retry
 
+from kubernetes.aio.utils.retry import (
+    is_retry_after_response,
+    on_retry_after_error,
+    retry_after_backoff,
+)
 from kubernetes.aio.client.exceptions import ApiException, ApiValueError
 
 RESTResponseType = aiohttp.ClientResponse
@@ -284,14 +289,55 @@ async def request(
             self.pool_manager = self._create_pool_manager()
         pool_manager = self.pool_manager
 
-        if self._effective_retry_options is not None and method in ALLOW_RETRY_METHODS:
+        client_go_read_retries = (
+            method in ['GET', 'HEAD']
+            and getattr(self.configuration, 'client_go_retries', False)
+        )
+
+        if (
+            self._effective_retry_options is not None
+            and method in ALLOW_RETRY_METHODS
+            and not client_go_read_retries
+        ):
             if self.retry_client is None:
                 self.retry_client = aiohttp_retry.RetryClient(
                     client_session=self.pool_manager,
                     retry_options=self._effective_retry_options
                 )
             pool_manager = self.retry_client
 
-        r = await pool_manager.request(**args)
+        async def read_request(check_retry_status=False):
+            response = await self.pool_manager.request(**args)
+            if check_retry_status:
+                self._raise_retry_after_response(response)
+            return response
+
+        if client_go_read_retries:
+            backoff = retry_after_backoff(
+                getattr(self.configuration, 'retries', None),
+                getattr(self.configuration, 'client_go_retry_backoff', None),
+            )
+            r = await on_retry_after_error(
+                backoff, self._is_read_retryable, lambda: read_request(True))
+        else:
+            r = await pool_manager.request(**args)
 
         return RESTResponse(r)
+
+    @classmethod
+    def _is_read_retryable(cls, error):
+        return is_retry_after_response(error)
+
+    @staticmethod
+    def _retry_after_error(response):
+        error = ApiException(status=response.status, reason=response.reason)
+        error.headers = response.headers
+        return error
+
+    @classmethod
+    def _raise_retry_after_response(cls, response):
+        error = cls._retry_after_error(response)
+        if not is_retry_after_response(error):
+            return
+        response.release()
+        raise error

kubernetes/aio/test/test_generated_api.py

@@ -46,6 +46,8 @@ async def asyncSetUp(self):
             'items': [],
         }
         self.response_status = 200
+        self.response_headers = {}
+        self.responses = None
         app = web.Application()
         app.router.add_route('*', '/{path:.*}', self._handle_request)
         self.runner = web.AppRunner(app)
@@ -91,7 +93,17 @@ async def _handle_request(self, request):
             }).encode() + b'\n')
             await response.write_eof()
             return response
-        return web.json_response(self.response, status=self.response_status)
+        if self.responses is not None:
+            index = len(self.requests) - 1
+            response, status, headers = self.responses[
+                min(index, len(self.responses) - 1)
+            ]
+            return web.json_response(response, status=status, headers=headers)
+        return web.json_response(
+            self.response,
+            status=self.response_status,
+            headers=self.response_headers,
+        )
 
     async def test_bearer_alias_supports_synchronous_token_refresh(self):
         self.configuration.api_key['authorization'] = 'expired-token'
@@ -123,6 +135,20 @@ async def refresh(configuration):
             self.requests[-1][0].headers['Authorization'],
         )
 
+    async def test_client_go_retry_retries_get_retry_after_response(self):
+        self.configuration.client_go_retries = True
+        self.configuration.retries = 1
+        self.responses = [
+            ({'message': 'retry later'}, 429, {'Retry-After': '0'}),
+            (self.response, 200, {}),
+        ]
+
+        namespaces = await CoreV1Api(self.api_client).list_namespace()
+
+        self.assertEqual([], namespaces.items)
+        self.assertEqual(2, len(self.requests))
+        self.assertIsNone(self.api_client.rest_client.retry_client)
+
     async def test_delete_job_accepts_job_and_status_responses(self):
         responses = (
             (

kubernetes/aio/utils/__init__.py

@@ -14,13 +14,61 @@
 
 from __future__ import absolute_import
 
-from .create_from_yaml import (
-    FailToCreateError, create_from_dict, create_from_yaml,
-    create_from_yaml_single_item,
-)
+import typing as _typing
+
 from .retry import (Backoff, DEFAULT_BACKOFF, DEFAULT_RETRY,
                     DEFAULT_RETRY_AFTER_BACKOFF, on_error,
                     on_retry_after_error, retry_on_conflict,
                     is_conflict, is_retry_after_response,
                     is_too_many_requests, retry_after_backoff,
                     retry_after_max_retries, retry_after_seconds)
+
+
+__all__ = [
+    "FailToCreateError",
+    "create_from_dict",
+    "create_from_yaml",
+    "create_from_yaml_single_item",
+    "Backoff",
+    "DEFAULT_BACKOFF",
+    "DEFAULT_RETRY",
+    "DEFAULT_RETRY_AFTER_BACKOFF",
+    "on_error",
+    "on_retry_after_error",
+    "retry_on_conflict",
+    "is_conflict",
+    "is_retry_after_response",
+    "is_too_many_requests",
+    "retry_after_backoff",
+    "retry_after_max_retries",
+    "retry_after_seconds",
+]
+
+
+# REST imports retry while client is initializing, so defer client-dependent utilities.
+if _typing.TYPE_CHECKING:
+    from .create_from_yaml import (
+        FailToCreateError, create_from_dict, create_from_yaml,
+        create_from_yaml_single_item,
+    )
+else:
+    from importlib import import_module
+
+    _exports = {
+        "FailToCreateError": ".create_from_yaml",
+        "create_from_dict": ".create_from_yaml",
+        "create_from_yaml": ".create_from_yaml",
+        "create_from_yaml_single_item": ".create_from_yaml",
+    }
+
+    def __getattr__(name: str) -> object:
+        if (module_name := _exports.get(name)) is None:
+            raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+        module = import_module(module_name, __name__)
+        for export, export_module in _exports.items():
+            if export_module == module_name:
+                globals()[export] = getattr(module, export)
+        return globals()[name]
+
+    def __dir__() -> list[str]:
+        return sorted(globals().keys() | _exports.keys())

kubernetes/aio/utils/retry_test.py

@@ -20,6 +20,7 @@
     is_too_many_requests,
     on_error,
     on_retry_after_error,
+    retry_after_max_retries,
     retry_after_seconds,
     retry_on_conflict,
 )
@@ -33,6 +34,12 @@ def __init__(self, status, headers=None):
         self.headers = headers or {}
 
 
+class RetryOptions:
+
+    def __init__(self, attempts):
+        self.attempts = attempts
+
+
 class AioRetryTest(unittest.IsolatedAsyncioTestCase):
 
     def test_default_retry_matches_client_go(self):
@@ -41,6 +48,10 @@ def test_default_retry_matches_client_go(self):
             Backoff(steps=5, duration=0.01, factor=1.0, jitter=0.1),
         )
 
+    def test_retry_after_backoff_uses_aio_retry_attempts(self):
+        self.assertEqual(retry_after_max_retries(RetryOptions(attempts=3)), 2)
+        self.assertEqual(retry_after_max_retries(RetryOptions(attempts=1)), 0)
+
     def test_retry_after_seconds_parses_delay_seconds(self):
         error = FakeError(429, {"Retry-After": "7"})
 

kubernetes/base/retry.py

@@ -77,7 +77,8 @@ def retry_after_backoff(
     ``None``, it is interpreted as the retry ceiling and overrides
     ``backoff.steps``. ``False`` and ``0`` disable retries by returning a
     single-attempt backoff. Integer values and urllib3-style Retry objects with
-    a ``total`` value are treated as the retry ceiling.
+    a ``total`` value are treated as retry ceilings. aiohttp-retry-style
+    options with an ``attempts`` value are treated as request attempt ceilings.
     """
 
     if backoff is None:
@@ -106,13 +107,22 @@ def retry_after_max_retries(retries: Any = None) -> int:
     if isinstance(retries, int):
         return max(0, retries)
 
-    total = getattr(retries, "total", None)
-    if total is False:
+    if hasattr(retries, "total"):
+        total = getattr(retries, "total", None)
+        if total is False:
+            return 0
+        if total is True or total is None:
+            return 10
+        if isinstance(total, int):
+            return max(0, total)
+
+    attempts = getattr(retries, "attempts", None)
+    if attempts is False:
         return 0
-    if total is True or total is None:
+    if attempts is True:
         return 10
-    if isinstance(total, int):
-        return max(0, total)
+    if isinstance(attempts, int):
+        return max(0, attempts - 1)
     return 10
 
 

kubernetes/client/configuration.py

@@ -357,6 +357,22 @@ def __init__(
         self.retries = retries
         """Retry configuration
         """
+        self.client_go_retries = False
+        """Enable Kubernetes client-go-compatible retry semantics.
+
+           When enabled, GET and HEAD requests retry Retry-After responses.
+           The retry ceiling is read from ``retries`` when set; otherwise it
+           follows the client-go default of at most 10 retries.
+        """
+        self.client_go_retry_backoff = None
+        """Backoff for Kubernetes client-go-compatible retries.
+
+           If unset, client-go-compatible GET and HEAD retries use the
+           client-go default retry ceiling with no additional client-side
+           delay beyond Retry-After. When set, ``retries`` still overrides
+           the retry ceiling if it is not None.
+        """
+
         # Enable client side validation
         self.client_side_validation = client_side_validation