feat: add retryer

Author: yuguo.dtpe

volcenginesdkcore/api_client.py

@@ -3,7 +3,9 @@
 from __future__ import absolute_import
 
 import datetime
+import logging
 from multiprocessing.pool import ThreadPool
+from time import sleep
 
 # python 2 and python 3 compatibility library
 import six
@@ -15,7 +17,9 @@
 from volcenginesdkcore.interceptor import DeserializedResponseInterceptor
 from volcenginesdkcore.interceptor import InterceptorChain, InterceptorContext
 from volcenginesdkcore.interceptor import Request, Response
+from volcenginesdkcore.retryer.retryer import DEFAULT_RETRYER
 
+logger = logging.getLogger(__name__)
 
 class ApiClient(object):
     """Generic API client for Swagger client library builds.
@@ -122,14 +126,35 @@ def __call_api(
 
         interceptor_context = self.interceptor_chain.execute_request(interceptor_context)
 
-        # perform request and return response
-        response_data = self.request(
-            method, url=interceptor_context.request.url, query_params=interceptor_context.request.query_params,
-            headers=interceptor_context.request.header_params,
-            post_params=interceptor_context.request.post_params, body=interceptor_context.request.body,
-            _preload_content=interceptor_context.request.preload_content,
-            _request_timeout=interceptor_context.request.request_timeout)
-        self.last_response = response_data
+        retry_count = 0
+        response_data = None
+        retry_err = None
+
+        auto_retry = interceptor_context.request.auto_retry
+        retryer = interceptor_context.request.retryer
+        num_max_retries = retryer.num_max_retries
+
+        while True:
+            # perform request and return response
+            try:
+                response_data = self.request(
+                    method, url=interceptor_context.request.url, query_params=interceptor_context.request.query_params,
+                    headers=interceptor_context.request.header_params,
+                    post_params=interceptor_context.request.post_params, body=interceptor_context.request.body,
+                    _preload_content=interceptor_context.request.preload_content,
+                    _request_timeout=interceptor_context.request.request_timeout)
+                self.last_response = response_data
+            except Exception as e:
+                logger.warning("request error: {}".format(e))
+                retry_err = e
+                if retry_count >= num_max_retries:
+                    raise e
+
+            if not auto_retry or not self.request_retry(response_data, retry_count, retry_err, retryer):
+                if retry_err is not None:
+                    raise retry_err
+                break
+            retry_count += 1
 
         interceptor_context.response = Response(response_data)
         interceptor_context = self.interceptor_chain.execute_response(interceptor_context)
@@ -141,6 +166,16 @@ def __call_api(
             return (response.result, response.http_response.status,
                     response.http_response.getheaders())
 
+    def request_retry(self, response_data, retry_count, retry_err, retryer):
+        if retryer.should_retry(response_data, retry_count, retry_err):
+            delay = retryer.get_backoff_delay(retry_count)
+            sleep(delay / 1000)
+            if self.configuration.debug:
+                logger.debug("retry backoff strategy:%s, retry condition: %s, max retry count:%d, current retry count: %d, retry delay(ms):%f",
+                             type(retryer.backoff_strategy).__name__,type(retryer.retry_condition).__name__, retryer.num_max_retries, retry_count + 1, delay)
+            return True
+        return False
+
     def call_api(self, resource_path, method,
                  path_params=None, query_params=None, header_params=None,
                  body=None, post_params=None, files=None,

volcenginesdkcore/configuration.py

@@ -12,6 +12,7 @@
 from six.moves import http_client as httplib
 
 from volcenginesdkcore.endpoint import DefaultEndpointProvider
+from volcenginesdkcore.retryer.retryer import DEFAULT_RETRYER
 
 
 class TypeWithDefault(type):
@@ -134,6 +135,12 @@ def __init__(self):
 
         self.endpoint_provider = DefaultEndpointProvider()
 
+        self.auto_retry = True
+        self.__retryer = DEFAULT_RETRYER
+        self.__retry_error_codes = None
+        self.__min_retry_delay_ms = None
+        self.__max_retry_delay_ms = None
+
     @property
     def logger_file(self):
         """The logger file.
@@ -259,3 +266,68 @@ def to_debug_report(self):
                "Version of the API: 0.1.0\n" \
                "SDK Package Version: 3.0.15".\
             format(env=sys.platform, pyversion=sys.version)
+
+    @property
+    def num_max_retries(self):
+        return self.__retryer.num_max_retries
+
+    @num_max_retries.setter
+    def num_max_retries(self, num_max_retries):
+        if num_max_retries is None:
+            raise ValueError("num_max_retries cannot be None")
+        if num_max_retries < 0:
+            raise ValueError("num_max_retries must be greater than or equal to 0")
+        self.__retryer.num_max_retries = num_max_retries
+
+    @property
+    def backoff_strategy(self):
+        return self.__retryer.backoff_strategy
+
+    @backoff_strategy.setter
+    def backoff_strategy(self, value):
+        self.__retryer.backoff_strategy = value
+        if self.min_retry_delay_ms is not None:
+            self.__retryer.backoff_strategy.min_retry_delay_ms = self.min_retry_delay_ms
+        if self.max_retry_delay_ms is not None:
+            self.__retryer.backoff_strategy.max_retry_delay_ms = self.max_retry_delay_ms
+
+    @property
+    def retry_condition(self):
+        return self.__retryer.retry_condition
+
+    @retry_condition.setter
+    def retry_condition(self, value):
+        self.__retryer.retry_condition = value
+        if self.retry_error_codes is not None:
+            self.__retryer.retry_condition.retry_error_codes = self.retry_error_codes
+
+    @property
+    def retry_error_codes(self):
+        return self.__retry_error_codes
+
+    @retry_error_codes.setter
+    def retry_error_codes(self, value):
+        self.__retry_error_codes = value
+        self.__retryer.retry_condition.retry_error_codes = value
+
+    @property
+    def min_retry_delay_ms(self):
+        return self.__min_retry_delay_ms
+
+    @min_retry_delay_ms.setter
+    def min_retry_delay_ms(self, value):
+        self.__min_retry_delay_ms = value
+        self.__retryer.backoff_strategy.min_retry_delay_ms = value
+
+    @property
+    def max_retry_delay_ms(self):
+        return self.__max_retry_delay_ms
+
+    @max_retry_delay_ms.setter
+    def max_retry_delay_ms(self, value):
+        self.__max_retry_delay_ms = value
+        self.__retryer.backoff_strategy.max_retry_delay_ms = value
+
+    @property
+    def retryer(self):
+        return self.__retryer

volcenginesdkcore/interceptor/interceptors/build_request_interceptor.py

@@ -1,9 +1,9 @@
 import datetime
 import mimetypes
 import os
-from urllib.parse import quote
 
 import six
+from six.moves.urllib.parse import quote
 
 from .interceptor import RequestInterceptor
 

volcenginesdkcore/interceptor/interceptors/request.py

@@ -1,6 +1,11 @@
+from volcenginesdkcore.retryer.retryer import DefaultRetryCondition, ExponentialWithRandomJitterBackoffStrategy
+
+
 class RuntimeOption(object):
     def __init__(self, client_side_validation, ak=None, sk=None, session_token=None, region=None, scheme=None,
-                 endpoint_provider=None, connect_timeout=None, read_timeout=None):
+                 endpoint_provider=None, connect_timeout=None, read_timeout=None, auto_retry=None, num_max_retries=None,
+                 backoff_strategy=None, retry_condition=None,
+                 retry_error_codes=None, min_retry_delay_ms=None, max_retry_delay_ms=None):
         self.ak = ak
         self.sk = sk
         self.session_token = session_token
@@ -10,6 +15,13 @@ def __init__(self, client_side_validation, ak=None, sk=None, session_token=None,
         self.connect_timeout = connect_timeout
         self.read_timeout = read_timeout
         self.client_side_validation = client_side_validation
+        self.auto_retry = auto_retry
+        self.num_max_retries = num_max_retries
+        self.backoff_strategy = backoff_strategy
+        self.retry_condition = retry_condition
+        self.retry_error_codes = retry_error_codes
+        self.min_retry_delay_ms = min_retry_delay_ms
+        self.max_retry_delay_ms = max_retry_delay_ms
 
 
 class Request(object):
@@ -49,6 +61,10 @@ def __init__(
         self.custom_bootstrap_region = configuration.custom_bootstrap_region
         self.use_dual_stack = configuration.use_dual_stack
 
+        # retryer setting, default use global configration if value not set
+        self.auto_retry = configuration.auto_retry
+        self.retryer = configuration.retryer
+
         self.runtime_options = None
         if hasattr(body, '_configuration') and isinstance(body._configuration, RuntimeOption):
             self.runtime_options = body._configuration

volcenginesdkcore/interceptor/interceptors/runtime_options_interceptor.py

@@ -32,4 +32,25 @@ def intercept(self, context):
             context.request.endpoint_provider = opt.endpoint_provider
             context.request.host = None
 
+        # retryer arguments set
+        RuntimeOptionsInterceptor.__update_retryer(context, opt)
         return context
+
+    @staticmethod
+    def __update_retryer(context, opt):
+        min_retry_delay_ms = opt.min_retry_delay_ms \
+            if opt.min_retry_delay_ms is not None else context.request.retryer.backoff_strategy.min_retry_delay_ms
+        max_retry_delay_ms = opt.max_retry_delay_ms \
+            if opt.max_retry_delay_ms is not None else context.request.retryer.backoff_strategy.max_retry_delay_ms
+        retry_error_codes = opt.retry_error_codes \
+            if opt.retry_error_codes is not None else context.request.retryer.retry_condition.retry_error_codes
+        context.request.auto_retry = opt.auto_retry if opt.auto_retry is not None else context.request.auto_retry
+        context.request.retryer.num_max_retries = opt.num_max_retries \
+            if opt.num_max_retries is not None else context.request.num_max_retries
+        context.request.retryer.backoff_strategy = opt.backoff_strategy \
+            if opt.backoff_strategy is not None else context.request.retryer.backoff_strategy
+        context.request.retryer.backoff_strategy.min_retry_delay_ms = min_retry_delay_ms
+        context.request.retryer.backoff_strategy.max_retry_delay_ms = max_retry_delay_ms
+        context.request.retryer.retry_condition = opt.retry_condition \
+            if opt.retry_condition is not None else context.request.retryer.retry_condition
+        context.request.retryer.retry_condition.retry_error_codes = retry_error_codes

volcenginesdkcore/rest.py

@@ -12,6 +12,7 @@
 # python 2 and python 3 compatibility library
 import six
 from six.moves.urllib.parse import urlencode
+from urllib3 import Retry
 
 try:
     import urllib3
@@ -88,6 +89,7 @@ def __init__(self, configuration, pools_size=4, maxsize=None):
                 key_file=configuration.key_file,
                 proxy_url=configuration.proxy,
                 timeout=timeout,
+                retries=Retry(total=False),
                 **addition_pool_args
             )
         else:
@@ -99,6 +101,7 @@ def __init__(self, configuration, pools_size=4, maxsize=None):
                 cert_file=configuration.cert_file,
                 key_file=configuration.key_file,
                 timeout=timeout,
+                retries=Retry(total=False),
                 **addition_pool_args
             )
 

volcenginesdkcore/retryer/__init__.py

@@ -0,0 +1,97 @@
+# coding: utf-8
+
+import random
+from abc import abstractmethod
+
+from volcenginesdkcore.utils import six_utils
+
+class BackoffStrategy(six_utils.get_abstract_meta_class()):
+    """
+    Abstract base class for all backoff strategies.
+    Defines the interface for computing retry delays.
+    """
+
+    def __init__(self,
+                 min_retry_delay_ms=30,
+                 max_retry_delay_ms=300 * 1000
+                 ):
+        # type:(float, float) -> None
+        """
+        Initializes the BackoffStrategy with a min_retry_delay_ms and max_retry_delay_ms values.
+        Args:
+            :param min_retry_delay_ms: The minimum retry delay in milliseconds.
+            :param max_retry_delay_ms: The maximum retry delay in milliseconds.
+       """
+        self.min_retry_delay_ms = min_retry_delay_ms
+        self.max_retry_delay_ms = max_retry_delay_ms
+
+
+    @abstractmethod
+    def compute_delay(self, retry_count):
+        # type: (int) -> float
+        """
+        Args:
+            :param retry_count: The number of retries.
+        Returns:
+            float: The computed delay in milliseconds.
+        """
+        pass
+
+
+class NoBackoffStrategy(BackoffStrategy):
+    """
+    A backoff strategy that implements no delay between retries.
+    """
+
+    def compute_delay(self, retry_count):
+        # type: (int) -> float
+        """
+        Args:
+            :param retry_count: The number of retries.
+        Returns:
+            float: The computed delay in milliseconds.
+        """
+        return 0.0
+
+
+class ExponentialBackoffStrategy(BackoffStrategy):
+    """
+    A backoff strategy that increases the delay exponentially with each retry.
+    The delay is calculated as: min_retry_delay_ms * (2 ** retry_count),
+    capped by max_retry_delay_ms.
+    """
+
+    def compute_delay(self, retry_count):
+        # type: (int) -> float
+        """
+        Args:
+            :param retry_count: The number of retries.
+        Returns:
+            float: The computed delay in milliseconds.
+        """
+        min_retry_delay_ms = self.min_retry_delay_ms
+        max_retry_delay_ms = self.max_retry_delay_ms
+        delay = min(min_retry_delay_ms * (2 ** retry_count), max_retry_delay_ms)
+        return delay
+
+class ExponentialWithRandomJitterBackoffStrategy(ExponentialBackoffStrategy):
+    """
+    An exponential backoff strategy with random jitter.
+    The delay is calculated as: base + random_uniform(0, base),
+    capped by max_retry_delay_ms.
+    This effectively means the delay is a random value between base and 2*base,
+    then capped by max_retry_delay_ms.
+    """
+
+    def compute_delay(self, retry_count):
+        # type: (int) -> float
+        """
+        Args:
+            :param retry_count: The number of retries.
+        Returns:
+            float: The computed delay in milliseconds.
+        """
+        base = super(ExponentialWithRandomJitterBackoffStrategy, self).compute_delay(retry_count)
+        jitter = base + random.uniform(0, base)
+        max_retry_delay_ms = self.max_retry_delay_ms
+        return min(max_retry_delay_ms, jitter)