rest: apply client-go retry generation patches

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

Author: sttts

kubernetes/aio/client/_retry.py

@@ -0,0 +1,61 @@
+# Copyright 2026 The Kubernetes Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import asyncio
+import random
+from typing import Awaitable, Callable, TypeVar
+
+from ._retry_base import (
+    Backoff,
+    _delay,
+    is_retry_after_response,
+    retry_after_backoff,
+    retry_after_seconds,
+)
+
+
+T = TypeVar("T")
+
+
+async def on_retry_after_error(
+    backoff: Backoff,
+    retriable: Callable[[Exception], bool],
+    fn: Callable[[], Awaitable[T]],
+    sleep_func: Callable[[float], Awaitable[None]] = asyncio.sleep,
+    random_func: Callable[[], float] = random.random,
+) -> T:
+    """Async implementation of client-go REST Retry-After sleep semantics."""
+
+    steps = backoff.steps
+    duration = backoff.duration
+    last_error = None
+    while steps > 0:
+        try:
+            return await fn()
+        except Exception as error:
+            if not retriable(error):
+                raise
+            last_error = error
+
+            if steps == 1:
+                break
+
+            delay, duration, steps = _delay(
+                steps, duration, backoff, random_func)
+            retry_after = retry_after_seconds(error)
+            if retry_after is not None and retry_after > delay:
+                delay = retry_after
+            await sleep_func(delay)
+
+    raise last_error

kubernetes/aio/client/_retry_base.py

@@ -0,0 +1 @@
+../../base/retry.py

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.client._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/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/_retry.py

@@ -0,0 +1,61 @@
+# Copyright 2026 The Kubernetes Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import random
+import time
+from typing import Callable, TypeVar
+
+from ._retry_base import (
+    Backoff,
+    _delay,
+    is_retry_after_response,
+    retry_after_backoff,
+    retry_after_seconds,
+)
+
+
+T = TypeVar("T")
+
+
+def on_retry_after_error(
+    backoff: Backoff,
+    retriable: Callable[[Exception], bool],
+    fn: Callable[[], T],
+    sleep_func: Callable[[float], None] = time.sleep,
+    random_func: Callable[[], float] = random.random,
+) -> T:
+    """Run ``fn`` with client-go REST Retry-After sleep semantics."""
+
+    steps = backoff.steps
+    duration = backoff.duration
+    last_error = None
+    while steps > 0:
+        try:
+            return fn()
+        except Exception as error:
+            if not retriable(error):
+                raise
+            last_error = error
+
+            if steps == 1:
+                break
+
+            delay, duration, steps = _delay(
+                steps, duration, backoff, random_func)
+            retry_after = retry_after_seconds(error)
+            if retry_after is not None and retry_after > delay:
+                delay = retry_after
+            sleep_func(delay)
+
+    raise last_error

kubernetes/client/_retry_base.py

@@ -0,0 +1 @@
+../base/retry.py

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