Source code for kazoo.retry

import logging
import random
import time

from kazoo.exceptions import (
    ConnectionClosedError,
    ConnectionLoss,
    KazooException,
    OperationTimeoutError,
    SessionExpiredError,
)


log = logging.getLogger(__name__)


[docs] class ForceRetryError(Exception): """Raised when some recipe logic wants to force a retry."""
[docs] class RetryFailedError(KazooException): """Raised when retrying an operation ultimately failed, after retrying the maximum number of attempts. """
[docs] class InterruptedError(RetryFailedError): """Raised when the retry is forcibly interrupted by the interrupt function"""
[docs] class KazooRetry(object): """Helper for retrying a method in the face of retry-able exceptions""" RETRY_EXCEPTIONS = (ConnectionLoss, OperationTimeoutError, ForceRetryError) EXPIRED_EXCEPTIONS = (SessionExpiredError,)
[docs] def __init__( self, max_tries=1, delay=0.1, backoff=2, max_jitter=0.4, max_delay=60.0, ignore_expire=True, sleep_func=time.sleep, deadline=None, interrupt=None, ): """Create a :class:`KazooRetry` instance for retrying function calls. :param max_tries: How many times to retry the command. -1 means infinite tries. :param delay: Initial delay between retry attempts. :param backoff: Backoff multiplier between retry attempts. Defaults to 2 for exponential backoff. :param max_jitter: Percentage of jitter to apply to each retry's delay to ensure all clients to do not hammer the server at the same time. Between 0.0 and 1.0. :param max_delay: Maximum delay in seconds, regardless of other backoff settings. Defaults to one minute. :param ignore_expire: Whether a session expiration should be ignored and treated as a retry-able command. :param interrupt: Function that will be called with no args that may return True if the retry should be ceased immediately. This will be called no more than every 0.1 seconds during a wait between retries. """ self.max_tries = max_tries self.delay = delay self.backoff = backoff # Ensure max_jitter is in (0, 1) self.max_jitter = max(min(max_jitter, 1.0), 0.0) self.max_delay = float(max_delay) self._attempts = 0 self._cur_delay = delay self.deadline = deadline self._cur_stoptime = None self.sleep_func = sleep_func self.retry_exceptions = self.RETRY_EXCEPTIONS self.interrupt = interrupt if ignore_expire: self.retry_exceptions += self.EXPIRED_EXCEPTIONS
[docs] def reset(self): """Reset the attempt counter""" self._attempts = 0 self._cur_delay = self.delay self._cur_stoptime = None
[docs] def copy(self): """Return a clone of this retry manager""" obj = KazooRetry( max_tries=self.max_tries, delay=self.delay, backoff=self.backoff, max_jitter=self.max_jitter, max_delay=self.max_delay, sleep_func=self.sleep_func, deadline=self.deadline, interrupt=self.interrupt, ) obj.retry_exceptions = self.retry_exceptions return obj
[docs] def __call__(self, func, *args, **kwargs): """Call a function with arguments until it completes without throwing a Kazoo exception :param func: Function to call :param args: Positional arguments to call the function with :params kwargs: Keyword arguments to call the function with The function will be called until it doesn't throw one of the retryable exceptions (ConnectionLoss, OperationTimeout, or ForceRetryError), and optionally retrying on session expiration. """ self.reset() while True: try: if self.deadline is not None and self._cur_stoptime is None: self._cur_stoptime = time.monotonic() + self.deadline return func(*args, **kwargs) except ConnectionClosedError: raise except self.retry_exceptions: self._attempts += 1 # Note: max_tries == -1 means infinite tries. if self._attempts == self.max_tries: raise RetryFailedError("Too many retry attempts") jitter = random.uniform( 1.0 - self.max_jitter, 1.0 + self.max_jitter ) sleeptime = self._cur_delay * jitter if ( self._cur_stoptime is not None and time.monotonic() + sleeptime >= self._cur_stoptime ): raise RetryFailedError("Exceeded retry deadline") if self.interrupt: remain_time = sleeptime while remain_time > 0: # Break the time period down and sleep for no # longer than 0.1 before calling the interrupt self.sleep_func(min(0.1, remain_time)) remain_time -= 0.1 if self.interrupt(): raise InterruptedError() else: self.sleep_func(sleeptime) self._cur_delay = min(sleeptime * self.backoff, self.max_delay)