retry.py 5.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150
  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.8,
  33. max_delay=3600, 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: Additional max jitter period to wait between
  43. retry attempts to avoid slamming the server.
  44. :param max_delay: Maximum delay in seconds, regardless of other
  45. backoff settings. Defaults to one hour.
  46. :param ignore_expire:
  47. Whether a session expiration should be ignored and treated
  48. as a retry-able command.
  49. :param interrupt:
  50. Function that will be called with no args that may return
  51. True if the retry should be ceased immediately. This will
  52. be called no more than every 0.1 seconds during a wait
  53. between retries.
  54. """
  55. self.max_tries = max_tries
  56. self.delay = delay
  57. self.backoff = backoff
  58. self.max_jitter = int(max_jitter * 100)
  59. self.max_delay = float(max_delay)
  60. self._attempts = 0
  61. self._cur_delay = delay
  62. self.deadline = deadline
  63. self._cur_stoptime = None
  64. self.sleep_func = sleep_func
  65. self.retry_exceptions = self.RETRY_EXCEPTIONS
  66. self.interrupt = interrupt
  67. if ignore_expire:
  68. self.retry_exceptions += self.EXPIRED_EXCEPTIONS
  69. def reset(self):
  70. """Reset the attempt counter"""
  71. self._attempts = 0
  72. self._cur_delay = self.delay
  73. self._cur_stoptime = None
  74. def copy(self):
  75. """Return a clone of this retry manager"""
  76. obj = KazooRetry(max_tries=self.max_tries,
  77. delay=self.delay,
  78. backoff=self.backoff,
  79. max_jitter=self.max_jitter / 100.0,
  80. max_delay=self.max_delay,
  81. sleep_func=self.sleep_func,
  82. deadline=self.deadline,
  83. interrupt=self.interrupt)
  84. obj.retry_exceptions = self.retry_exceptions
  85. return obj
  86. def __call__(self, func, *args, **kwargs):
  87. """Call a function with arguments until it completes without
  88. throwing a Kazoo exception
  89. :param func: Function to call
  90. :param args: Positional arguments to call the function with
  91. :params kwargs: Keyword arguments to call the function with
  92. The function will be called until it doesn't throw one of the
  93. retryable exceptions (ConnectionLoss, OperationTimeout, or
  94. ForceRetryError), and optionally retrying on session
  95. expiration.
  96. """
  97. self.reset()
  98. while True:
  99. try:
  100. if self.deadline is not None and self._cur_stoptime is None:
  101. self._cur_stoptime = time.time() + self.deadline
  102. return func(*args, **kwargs)
  103. except ConnectionClosedError:
  104. raise
  105. except self.retry_exceptions:
  106. # Note: max_tries == -1 means infinite tries.
  107. if self._attempts == self.max_tries:
  108. raise RetryFailedError("Too many retry attempts")
  109. self._attempts += 1
  110. sleeptime = self._cur_delay + (random.randint(0, self.max_jitter) / 100.0)
  111. if self._cur_stoptime is not None and time.time() + sleeptime >= self._cur_stoptime:
  112. raise RetryFailedError("Exceeded retry deadline")
  113. if self.interrupt:
  114. while sleeptime > 0:
  115. # Break the time period down and sleep for no longer than
  116. # 0.1 before calling the interrupt
  117. if sleeptime < 0.1:
  118. self.sleep_func(sleeptime)
  119. sleeptime -= sleeptime
  120. else:
  121. self.sleep_func(0.1)
  122. sleeptime -= 0.1
  123. if self.interrupt():
  124. raise InterruptedError()
  125. else:
  126. self.sleep_func(sleeptime)
  127. self._cur_delay = min(self._cur_delay * self.backoff, self.max_delay)