retryhandler.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359
  1. # Copyright (c) 2012-2013 Mitch Garnaat http://garnaat.org/
  2. # Copyright 2012-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  3. #
  4. # Licensed under the Apache License, Version 2.0 (the "License"). You
  5. # may not use this file except in compliance with the License. A copy of
  6. # the License is located at
  7. #
  8. # http://aws.amazon.com/apache2.0/
  9. #
  10. # or in the "license" file accompanying this file. This file is
  11. # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
  12. # ANY KIND, either express or implied. See the License for the specific
  13. # language governing permissions and limitations under the License.
  14. import random
  15. import functools
  16. import logging
  17. from binascii import crc32
  18. from botocore.exceptions import (
  19. ChecksumError, EndpointConnectionError, ReadTimeoutError,
  20. ConnectionError, ConnectionClosedError,
  21. )
  22. logger = logging.getLogger(__name__)
  23. # The only supported error for now is GENERAL_CONNECTION_ERROR
  24. # which maps to requests generic ConnectionError. If we're able
  25. # to get more specific exceptions from requests we can update
  26. # this mapping with more specific exceptions.
  27. EXCEPTION_MAP = {
  28. 'GENERAL_CONNECTION_ERROR': [
  29. ConnectionError, ConnectionClosedError, ReadTimeoutError,
  30. EndpointConnectionError
  31. ],
  32. }
  33. def delay_exponential(base, growth_factor, attempts):
  34. """Calculate time to sleep based on exponential function.
  35. The format is::
  36. base * growth_factor ^ (attempts - 1)
  37. If ``base`` is set to 'rand' then a random number between
  38. 0 and 1 will be used as the base.
  39. Base must be greater than 0, otherwise a ValueError will be
  40. raised.
  41. """
  42. if base == 'rand':
  43. base = random.random()
  44. elif base <= 0:
  45. raise ValueError("The 'base' param must be greater than 0, "
  46. "got: %s" % base)
  47. time_to_sleep = base * (growth_factor ** (attempts - 1))
  48. return time_to_sleep
  49. def create_exponential_delay_function(base, growth_factor):
  50. """Create an exponential delay function based on the attempts.
  51. This is used so that you only have to pass it the attempts
  52. parameter to calculate the delay.
  53. """
  54. return functools.partial(
  55. delay_exponential, base=base, growth_factor=growth_factor)
  56. def create_retry_handler(config, operation_name=None):
  57. checker = create_checker_from_retry_config(
  58. config, operation_name=operation_name)
  59. action = create_retry_action_from_config(
  60. config, operation_name=operation_name)
  61. return RetryHandler(checker=checker, action=action)
  62. def create_retry_action_from_config(config, operation_name=None):
  63. # The spec has the possibility of supporting per policy
  64. # actions, but right now, we assume this comes from the
  65. # default section, which means that delay functions apply
  66. # for every policy in the retry config (per service).
  67. delay_config = config['__default__']['delay']
  68. if delay_config['type'] == 'exponential':
  69. return create_exponential_delay_function(
  70. base=delay_config['base'],
  71. growth_factor=delay_config['growth_factor'])
  72. def create_checker_from_retry_config(config, operation_name=None):
  73. checkers = []
  74. max_attempts = None
  75. retryable_exceptions = []
  76. if '__default__' in config:
  77. policies = config['__default__'].get('policies', [])
  78. max_attempts = config['__default__']['max_attempts']
  79. for key in policies:
  80. current_config = policies[key]
  81. checkers.append(_create_single_checker(current_config))
  82. retry_exception = _extract_retryable_exception(current_config)
  83. if retry_exception is not None:
  84. retryable_exceptions.extend(retry_exception)
  85. if operation_name is not None and config.get(operation_name) is not None:
  86. operation_policies = config[operation_name]['policies']
  87. for key in operation_policies:
  88. checkers.append(_create_single_checker(operation_policies[key]))
  89. retry_exception = _extract_retryable_exception(
  90. operation_policies[key])
  91. if retry_exception is not None:
  92. retryable_exceptions.extend(retry_exception)
  93. if len(checkers) == 1:
  94. # Don't need to use a MultiChecker
  95. return MaxAttemptsDecorator(checkers[0], max_attempts=max_attempts)
  96. else:
  97. multi_checker = MultiChecker(checkers)
  98. return MaxAttemptsDecorator(
  99. multi_checker, max_attempts=max_attempts,
  100. retryable_exceptions=tuple(retryable_exceptions))
  101. def _create_single_checker(config):
  102. if 'response' in config['applies_when']:
  103. return _create_single_response_checker(
  104. config['applies_when']['response'])
  105. elif 'socket_errors' in config['applies_when']:
  106. return ExceptionRaiser()
  107. def _create_single_response_checker(response):
  108. if 'service_error_code' in response:
  109. checker = ServiceErrorCodeChecker(
  110. status_code=response['http_status_code'],
  111. error_code=response['service_error_code'])
  112. elif 'http_status_code' in response:
  113. checker = HTTPStatusCodeChecker(
  114. status_code=response['http_status_code'])
  115. elif 'crc32body' in response:
  116. checker = CRC32Checker(header=response['crc32body'])
  117. else:
  118. # TODO: send a signal.
  119. raise ValueError("Unknown retry policy: %s" % config)
  120. return checker
  121. def _extract_retryable_exception(config):
  122. applies_when = config['applies_when']
  123. if 'crc32body' in applies_when.get('response', {}):
  124. return [ChecksumError]
  125. elif 'socket_errors' in applies_when:
  126. exceptions = []
  127. for name in applies_when['socket_errors']:
  128. exceptions.extend(EXCEPTION_MAP[name])
  129. return exceptions
  130. class RetryHandler(object):
  131. """Retry handler.
  132. The retry handler takes two params, ``checker`` object
  133. and an ``action`` object.
  134. The ``checker`` object must be a callable object and based on a response
  135. and an attempt number, determines whether or not sufficient criteria for
  136. a retry has been met. If this is the case then the ``action`` object
  137. (which also is a callable) determines what needs to happen in the event
  138. of a retry.
  139. """
  140. def __init__(self, checker, action):
  141. self._checker = checker
  142. self._action = action
  143. def __call__(self, attempts, response, caught_exception, **kwargs):
  144. """Handler for a retry.
  145. Intended to be hooked up to an event handler (hence the **kwargs),
  146. this will process retries appropriately.
  147. """
  148. if self._checker(attempts, response, caught_exception):
  149. result = self._action(attempts=attempts)
  150. logger.debug("Retry needed, action of: %s", result)
  151. return result
  152. logger.debug("No retry needed.")
  153. class BaseChecker(object):
  154. """Base class for retry checkers.
  155. Each class is responsible for checking a single criteria that determines
  156. whether or not a retry should not happen.
  157. """
  158. def __call__(self, attempt_number, response, caught_exception):
  159. """Determine if retry criteria matches.
  160. Note that either ``response`` is not None and ``caught_exception`` is
  161. None or ``response`` is None and ``caught_exception`` is not None.
  162. :type attempt_number: int
  163. :param attempt_number: The total number of times we've attempted
  164. to send the request.
  165. :param response: The HTTP response (if one was received).
  166. :type caught_exception: Exception
  167. :param caught_exception: Any exception that was caught while trying to
  168. send the HTTP response.
  169. :return: True, if the retry criteria matches (and therefore a retry
  170. should occur. False if the criteria does not match.
  171. """
  172. # The default implementation allows subclasses to not have to check
  173. # whether or not response is None or not.
  174. if response is not None:
  175. return self._check_response(attempt_number, response)
  176. elif caught_exception is not None:
  177. return self._check_caught_exception(
  178. attempt_number, caught_exception)
  179. else:
  180. raise ValueError("Both response and caught_exception are None.")
  181. def _check_response(self, attempt_number, response):
  182. pass
  183. def _check_caught_exception(self, attempt_number, caught_exception):
  184. pass
  185. class MaxAttemptsDecorator(BaseChecker):
  186. """Allow retries up to a maximum number of attempts.
  187. This will pass through calls to the decorated retry checker, provided
  188. that the number of attempts does not exceed max_attempts. It will
  189. also catch any retryable_exceptions passed in. Once max_attempts has
  190. been exceeded, then False will be returned or the retryable_exceptions
  191. that was previously being caught will be raised.
  192. """
  193. def __init__(self, checker, max_attempts, retryable_exceptions=None):
  194. self._checker = checker
  195. self._max_attempts = max_attempts
  196. self._retryable_exceptions = retryable_exceptions
  197. def __call__(self, attempt_number, response, caught_exception):
  198. should_retry = self._should_retry(attempt_number, response,
  199. caught_exception)
  200. if should_retry:
  201. if attempt_number >= self._max_attempts:
  202. # explicitly set MaxAttemptsReached
  203. if response is not None and 'ResponseMetadata' in response[1]:
  204. response[1]['ResponseMetadata']['MaxAttemptsReached'] = True
  205. logger.debug("Reached the maximum number of retry "
  206. "attempts: %s", attempt_number)
  207. return False
  208. else:
  209. return should_retry
  210. else:
  211. return False
  212. def _should_retry(self, attempt_number, response, caught_exception):
  213. if self._retryable_exceptions and \
  214. attempt_number < self._max_attempts:
  215. try:
  216. return self._checker(attempt_number, response, caught_exception)
  217. except self._retryable_exceptions as e:
  218. logger.debug("retry needed, retryable exception caught: %s",
  219. e, exc_info=True)
  220. return True
  221. else:
  222. # If we've exceeded the max attempts we just let the exception
  223. # propogate if one has occurred.
  224. return self._checker(attempt_number, response, caught_exception)
  225. class HTTPStatusCodeChecker(BaseChecker):
  226. def __init__(self, status_code):
  227. self._status_code = status_code
  228. def _check_response(self, attempt_number, response):
  229. if response[0].status_code == self._status_code:
  230. logger.debug(
  231. "retry needed: retryable HTTP status code received: %s",
  232. self._status_code)
  233. return True
  234. else:
  235. return False
  236. class ServiceErrorCodeChecker(BaseChecker):
  237. def __init__(self, status_code, error_code):
  238. self._status_code = status_code
  239. self._error_code = error_code
  240. def _check_response(self, attempt_number, response):
  241. if response[0].status_code == self._status_code:
  242. actual_error_code = response[1].get('Error', {}).get('Code')
  243. if actual_error_code == self._error_code:
  244. logger.debug(
  245. "retry needed: matching HTTP status and error code seen: "
  246. "%s, %s", self._status_code, self._error_code)
  247. return True
  248. return False
  249. class MultiChecker(BaseChecker):
  250. def __init__(self, checkers):
  251. self._checkers = checkers
  252. def __call__(self, attempt_number, response, caught_exception):
  253. for checker in self._checkers:
  254. checker_response = checker(attempt_number, response,
  255. caught_exception)
  256. if checker_response:
  257. return checker_response
  258. return False
  259. class CRC32Checker(BaseChecker):
  260. def __init__(self, header):
  261. # The header where the expected crc32 is located.
  262. self._header_name = header
  263. def _check_response(self, attempt_number, response):
  264. http_response = response[0]
  265. expected_crc = http_response.headers.get(self._header_name)
  266. if expected_crc is None:
  267. logger.debug("crc32 check skipped, the %s header is not "
  268. "in the http response.", self._header_name)
  269. else:
  270. actual_crc32 = crc32(response[0].content) & 0xffffffff
  271. if not actual_crc32 == int(expected_crc):
  272. logger.debug(
  273. "retry needed: crc32 check failed, expected != actual: "
  274. "%s != %s", int(expected_crc), actual_crc32)
  275. raise ChecksumError(checksum_type='crc32',
  276. expected_checksum=int(expected_crc),
  277. actual_checksum=actual_crc32)
  278. class ExceptionRaiser(BaseChecker):
  279. """Raise any caught exceptions.
  280. This class will raise any non None ``caught_exception``.
  281. """
  282. def _check_caught_exception(self, attempt_number, caught_exception):
  283. # This is implementation specific, but this class is useful by
  284. # coordinating with the MaxAttemptsDecorator.
  285. # The MaxAttemptsDecorator has a list of exceptions it should catch
  286. # and retry, but something needs to come along and actually raise the
  287. # caught_exception. That's what this class is being used for. If
  288. # the MaxAttemptsDecorator is not interested in retrying the exception
  289. # then this exception just propogates out past the retry code.
  290. raise caught_exception