fixup! 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/rest.py

@@ -21,12 +21,12 @@
 import aiohttp
 import aiohttp_retry
 
-from kubernetes.aio.client.exceptions import ApiException, ApiValueError
-from kubernetes.aio.utils.retry import (
+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
 
@@ -289,7 +289,16 @@ 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,
@@ -303,10 +312,7 @@ async def read_request(check_retry_status=False):
                 self._raise_retry_after_response(response)
             return response
 
-        if (
-            method in ['GET', 'HEAD']
-            and getattr(self.configuration, 'client_go_retries', False)
-        ):
+        if client_go_read_retries:
             backoff = retry_after_backoff(
                 getattr(self.configuration, 'retries', None),
                 getattr(self.configuration, 'client_go_retry_backoff', None),

kubernetes/aio/test/test_generated_api.py

@@ -147,6 +147,7 @@ async def test_client_go_retry_retries_get_retry_after_response(self):
 
         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/rest.py

@@ -22,12 +22,12 @@
 import urllib3
 from urllib3.util.retry import Retry
 
-from kubernetes.client.exceptions import ApiException, ApiValueError
-from kubernetes.utils.retry import (
+from kubernetes.client._retry import (
     is_retry_after_response,
     on_retry_after_error,
     retry_after_backoff,
 )
+from kubernetes.client.exceptions import ApiException, ApiValueError
 
 SUPPORTED_SOCKS_PROXIES = {"socks5", "socks5h", "socks4", "socks4a"}
 RESTResponseType = urllib3.HTTPResponse
@@ -225,25 +225,27 @@ def request(
                     read=_request_timeout[1]
                 )
 
-        client_go_retries = getattr(self.configuration, 'client_go_retries', False)
-        request_retries = None
-        if client_go_retries:
-            request_retries = self._urllib3_retries_without_status(
+        client_go_read_retries = (
+            method in ['GET', 'HEAD']
+            and getattr(self.configuration, 'client_go_retries', False)
+        )
+        read_retries = None
+        if client_go_read_retries:
+            read_retries = self._urllib3_retries_without_status(
                 getattr(self.configuration, 'retries', None))
 
-        def pool_request(*args, **kwargs):
-            if request_retries is not None:
-                kwargs['retries'] = request_retries
-            return self.pool_manager.request(*args, **kwargs)
-
         def read_request(check_retry_status=False):
-            response = pool_request(
+            kwargs = {}
+            if read_retries is not None:
+                kwargs['retries'] = read_retries
+            response = self.pool_manager.request(
                 method,
                 url,
                 fields={},
                 timeout=timeout,
                 headers=headers,
-                preload_content=False
+                preload_content=False,
+                **kwargs
             )
             if check_retry_status:
                 self._raise_retry_after_response(response)
@@ -279,7 +281,7 @@ def read_request(check_retry_status=False):
                     request_body = None
                     if body is not None:
                         request_body = json.dumps(body)
-                    r = pool_request(
+                    r = self.pool_manager.request(
                         method,
                         url,
                         body=request_body,
@@ -288,7 +290,7 @@ def read_request(check_retry_status=False):
                         preload_content=False
                     )
                 elif content_type == 'application/x-www-form-urlencoded':
-                    r = pool_request(
+                    r = self.pool_manager.request(
                         method,
                         url,
                         fields=post_params,
@@ -304,7 +306,7 @@ def read_request(check_retry_status=False):
                     del headers['Content-Type']
                     # Ensures that dict objects are serialized
                     post_params = [(a, json.dumps(b)) if isinstance(b, dict) else (a,b) for a, b in post_params]
-                    r = pool_request(
+                    r = self.pool_manager.request(
                         method,
                         url,
                         fields=post_params,
@@ -317,7 +319,7 @@ def read_request(check_retry_status=False):
                 # other content types than JSON when `body` argument is
                 # provided in serialized form.
                 elif isinstance(body, str) or isinstance(body, bytes):
-                    r = pool_request(
+                    r = self.pool_manager.request(
                         method,
                         url,
                         body=body,
@@ -327,7 +329,7 @@ def read_request(check_retry_status=False):
                     )
                 elif headers['Content-Type'].startswith('text/') and isinstance(body, bool):
                     request_body = "true" if body else "false"
-                    r = pool_request(
+                    r = self.pool_manager.request(
                         method,
                         url,
                         body=request_body,
@@ -342,7 +344,7 @@ def read_request(check_retry_status=False):
                     raise ApiException(status=0, reason=msg)
             # For `GET`, `HEAD`
             else:
-                if client_go_retries:
+                if client_go_read_retries:
                     backoff = retry_after_backoff(
                         getattr(self.configuration, 'retries', None),
                         getattr(self.configuration, 'client_go_retry_backoff', None),
@@ -373,12 +375,22 @@ def _raise_retry_after_response(cls, response):
         error = cls._retry_after_error(response)
         if not is_retry_after_response(error):
             return
-        try:
-            response.drain_conn()
-        except Exception:
-            response.release_conn()
+        cls._read_and_close_retry_response(response)
         raise error
 
+    @staticmethod
+    def _read_and_close_retry_response(response):
+        try:
+            try:
+                content_length = int(response.getheaders().get(
+                    'Content-Length', '-1'))
+            except (TypeError, ValueError):
+                content_length = -1
+            if content_length <= 2 << 10:
+                response.read(2 << 10)
+        finally:
+            response.close()
+
     @staticmethod
     def _urllib3_retries_without_status(retries):
         if retries is False: