From 58410db9dd804597b6c45cb5ec329f09eb362154 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 3 Sep 2015 11:47:08 +0200 Subject: [PATCH] Implemented Retry class inspired by KazooRetry --- helpers/__init__.py | 16 ++++++++++ helpers/dcs.py | 15 +-------- helpers/utils.py | 77 +++++++++++++++++++++++++++++++++++++++++++++ tests/test_utils.py | 58 +++++++++++++++++++++++++++++++++- 4 files changed, 151 insertions(+), 15 deletions(-) diff --git a/helpers/__init__.py b/helpers/__init__.py index e69de29b..2fe60065 100644 --- a/helpers/__init__.py +++ b/helpers/__init__.py @@ -0,0 +1,16 @@ +class PatroniException(Exception): + pass + + +class DCSError(PatroniException): + """Parent class for all kind of exceptions related to selected distributed configuration store""" + + def __init__(self, value): + self.value = value + + def __str__(self): + """ + >>> str(DCSError('foo')) + "'foo'" + """ + return repr(self.value) diff --git a/helpers/dcs.py b/helpers/dcs.py index f837245c..7ee97ca3 100644 --- a/helpers/dcs.py +++ b/helpers/dcs.py @@ -1,6 +1,7 @@ import abc from collections import namedtuple +from helpers import DCSError from helpers.utils import calculate_ttl, sleep from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl @@ -22,20 +23,6 @@ def parse_connection_string(value): return conn_url, api_url -class DCSError(Exception): - """Parent class for all kind of exceptions related to selected distributed configuration store""" - - def __init__(self, value): - self.value = value - - def __str__(self): - """ - >>> str(DCSError('foo')) - "'foo'" - """ - return repr(self.value) - - class Member(namedtuple('Member', 'index,name,conn_url,api_url,expiration,ttl')): """Immutable object (namedtuple) which represents single member of PostgreSQL cluster. Consists of the following fields: diff --git a/helpers/utils.py b/helpers/utils.py index c725e6d4..f0555f48 100644 --- a/helpers/utils.py +++ b/helpers/utils.py @@ -1,10 +1,13 @@ import datetime import os +import random import re import signal import sys import time +from helpers import DCSError + interrupted_sleep = False reap_children = False @@ -107,3 +110,77 @@ def reap_children(): pass finally: reap_children = False + + +class RetryFailedError(DCSError): + """Raised when retrying an operation ultimately failed, after retrying the maximum number of attempts.""" + + +class Retry: + """Helper for retrying a method in the face of retry-able exceptions""" + + def __init__(self, max_tries=1, delay=0.1, backoff=2, max_jitter=0.8, max_delay=3600, + sleep_func=time.sleep, deadline=None, retry_exceptions=DCSError): + """Create a :class:`Retry` 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: Additional max jitter period to wait between retry attempts to avoid slamming the server. + :param max_delay: Maximum delay in seconds, regardless of other backoff settings. Defaults to one hour. + :param retry_exceptions: single exception or tuple""" + + self.max_tries = max_tries + self.delay = delay + self.backoff = backoff + self.max_jitter = int(max_jitter * 100) + 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 = retry_exceptions + + def reset(self): + """Reset the attempt counter""" + self._attempts = 0 + self._cur_delay = self.delay + self._cur_stoptime = None + + def copy(self): + """Return a clone of this retry manager""" + return Retry(max_tries=self.max_tries, delay=self.delay, backoff=self.backoff, + max_jitter=self.max_jitter / 100.0, max_delay=self.max_delay, sleep_func=self.sleep_func, + deadline=self.deadline, retry_exceptions=self.retry_exceptions) + + def __call__(self, func, *args, **kwargs): + """Call a function with arguments until it completes without throwing a `retry_exceptions` + + :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""" + self.reset() + + while True: + try: + if self.deadline is not None and self._cur_stoptime is None: + self._cur_stoptime = time.time() + self.deadline + return func(*args, **kwargs) + except self.retry_exceptions: + # Note: max_tries == -1 means infinite tries. + if self._attempts == self.max_tries: + raise RetryFailedError("Too many retry attempts") + self._attempts += 1 + sleeptime = self._cur_delay + ( + random.randint(0, self.max_jitter) / 100.0) + + if self._cur_stoptime is not None and \ + time.time() + sleeptime >= self._cur_stoptime: + raise RetryFailedError("Exceeded retry deadline") + else: + self.sleep_func(sleeptime) + self._cur_delay = min(self._cur_delay * self.backoff, + self.max_delay) diff --git a/tests/test_utils.py b/tests/test_utils.py index 312277b6..da43ada3 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -2,7 +2,8 @@ import os import time import unittest -from helpers.utils import reap_children, sigchld_handler, sigterm_handler, sleep +from helpers import DCSError +from helpers.utils import Retry, RetryFailedError, reap_children, sigchld_handler, sigterm_handler, sleep def nop(*args, **kwargs): @@ -43,3 +44,58 @@ class TestUtils(unittest.TestCase): def test_sleep(self): time.sleep = time_sleep sleep(0.01) + + +class TestRetrySleeper(unittest.TestCase): + + def _pass(self): + pass + + def _fail(self, times=1): + scope = dict(times=0) + + def inner(): + if scope['times'] >= times: + pass + else: + scope['times'] += 1 + raise DCSError('Failed!') + return inner + + def _makeOne(self, *args, **kwargs): + return Retry(*args, **kwargs) + + def test_reset(self): + retry = self._makeOne(delay=0, max_tries=2) + retry(self._fail()) + self.assertEquals(retry._attempts, 1) + retry.reset() + self.assertEquals(retry._attempts, 0) + + def test_too_many_tries(self): + retry = self._makeOne(delay=0) + self.assertRaises(RetryFailedError, retry, self._fail(times=999)) + self.assertEquals(retry._attempts, 1) + + def test_maximum_delay(self): + def sleep_func(_time): + pass + + retry = self._makeOne(delay=10, max_tries=100, sleep_func=sleep_func) + retry(self._fail(times=10)) + self.assertTrue(retry._cur_delay < 4000, retry._cur_delay) + # gevent's sleep function is picky about the type + self.assertEquals(type(retry._cur_delay), float) + + def test_deadline(self): + def sleep_func(_time): + pass + + retry = self._makeOne(deadline=0.0001, sleep_func=sleep_func) + self.assertRaises(RetryFailedError, retry, self._fail(times=10)) + + def test_copy(self): + _sleep = lambda t: None + retry = self._makeOne(sleep_func=_sleep) + rcopy = retry.copy() + self.assertTrue(rcopy.sleep_func is _sleep)