retry.py 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154
  1. import logging
  2. import random
  3. import time
  4. from kazoo.exceptions import (
  5. ConnectionClosedError,
  6. ConnectionLoss,
  7. KazooException,
  8. OperationTimeoutError,
  9. SessionExpiredError,
  10. )
  11. log = logging.getLogger(__name__)
  12. class ForceRetryError(Exception):
  13. """Raised when some recipe logic wants to force a retry."""
  14. class RetryFailedError(KazooException):
  15. """Raised when retrying an operation ultimately failed, after
  16. retrying the maximum number of attempts.
  17. """
  18. class InterruptedError(RetryFailedError):
  19. """Raised when the retry is forcibly interrupted by the interrupt
  20. function"""
  21. class KazooRetry(object):
  22. """Helper for retrying a method in the face of retry-able
  23. exceptions"""
  24. RETRY_EXCEPTIONS = (
  25. ConnectionLoss,
  26. OperationTimeoutError,
  27. ForceRetryError
  28. )
  29. EXPIRED_EXCEPTIONS = (
  30. SessionExpiredError,
  31. )
  32. def __init__(self, max_tries=1, delay=0.1, backoff=2, max_jitter=0.4,
  33. max_delay=60.0, ignore_expire=True, sleep_func=time.sleep,
  34. deadline=None, interrupt=None):
  35. """Create a :class:`KazooRetry` instance for retrying function
  36. calls.
  37. :param max_tries: How many times to retry the command. -1 means
  38. infinite tries.
  39. :param delay: Initial delay between retry attempts.
  40. :param backoff: Backoff multiplier between retry attempts.
  41. Defaults to 2 for exponential backoff.
  42. :param max_jitter: Percentage of jitter to apply to each retry's delay
  43. to ensure all clients to do not hammer the server
  44. at the same time. Between 0.0 and 1.0.
  45. :param max_delay: Maximum delay in seconds, regardless of other
  46. backoff settings. Defaults to one minute.
  47. :param ignore_expire:
  48. Whether a session expiration should be ignored and treated
  49. as a retry-able command.
  50. :param interrupt:
  51. Function that will be called with no args that may return
  52. True if the retry should be ceased immediately. This will
  53. be called no more than every 0.1 seconds during a wait
  54. between retries.
  55. """
  56. self.max_tries = max_tries
  57. self.delay = delay
  58. self.backoff = backoff
  59. # Ensure max_jitter is in (0, 1)
  60. self.max_jitter = max(min(max_jitter, 1.0), 0.0)
  61. self.max_delay = float(max_delay)
  62. self._attempts = 0
  63. self._cur_delay = delay
  64. self.deadline = deadline
  65. self._cur_stoptime = None
  66. self.sleep_func = sleep_func
  67. self.retry_exceptions = self.RETRY_EXCEPTIONS
  68. self.interrupt = interrupt
  69. if ignore_expire:
  70. self.retry_exceptions += self.EXPIRED_EXCEPTIONS
  71. def reset(self):
  72. """Reset the attempt counter"""
  73. self._attempts = 0
  74. self._cur_delay = self.delay
  75. self._cur_stoptime = None
  76. def copy(self):
  77. """Return a clone of this retry manager"""
  78. obj = KazooRetry(max_tries=self.max_tries,
  79. delay=self.delay,
  80. backoff=self.backoff,
  81. max_jitter=self.max_jitter,
  82. max_delay=self.max_delay,
  83. sleep_func=self.sleep_func,
  84. deadline=self.deadline,
  85. interrupt=self.interrupt)
  86. obj.retry_exceptions = self.retry_exceptions
  87. return obj
  88. def __call__(self, func, *args, **kwargs):
  89. """Call a function with arguments until it completes without
  90. throwing a Kazoo exception
  91. :param func: Function to call
  92. :param args: Positional arguments to call the function with
  93. :params kwargs: Keyword arguments to call the function with
  94. The function will be called until it doesn't throw one of the
  95. retryable exceptions (ConnectionLoss, OperationTimeout, or
  96. ForceRetryError), and optionally retrying on session
  97. expiration.
  98. """
  99. self.reset()
  100. while True:
  101. try:
  102. if self.deadline is not None and self._cur_stoptime is None:
  103. self._cur_stoptime = time.time() + self.deadline
  104. return func(*args, **kwargs)
  105. except ConnectionClosedError:
  106. raise
  107. except self.retry_exceptions:
  108. # Note: max_tries == -1 means infinite tries.
  109. if self._attempts == self.max_tries:
  110. raise RetryFailedError("Too many retry attempts")
  111. self._attempts += 1
  112. jitter = random.uniform(1.0-self.max_jitter,
  113. 1.0+self.max_jitter)
  114. sleeptime = self._cur_delay * jitter
  115. if self._cur_stoptime is not None and \
  116. time.time() + sleeptime >= self._cur_stoptime:
  117. raise RetryFailedError("Exceeded retry deadline")
  118. if self.interrupt:
  119. remain_time = sleeptime
  120. while remain_time > 0:
  121. # Break the time period down and sleep for no
  122. # longer than 0.1 before calling the interrupt
  123. self.sleep_func(min(0.1, remain_time))
  124. remain_time -= 0.1
  125. if self.interrupt():
  126. raise InterruptedError()
  127. else:
  128. self.sleep_func(sleeptime)
  129. self._cur_delay = min(sleeptime * self.backoff,
  130. self.max_delay)