Merge pull request #20 from zalando/feature/retry

Retry requests to configuration store

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)

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:

helpers/etcd.py

@@ -11,7 +11,7 @@ import urllib3
 from dns.exception import DNSException
 from dns import resolver
 from helpers.dcs import AbstractDCS, Cluster, DCSError, Leader, Member, parse_connection_string
-from helpers.utils import sleep
+from helpers.utils import Retry, RetryFailedError, sleep
 from requests.exceptions import RequestException
 
 logger = logging.getLogger(__name__)
@@ -63,7 +63,12 @@ class Client(etcd.Client):
 
     # try to workarond bug in python-etcd: https://github.com/jplana/python-etcd/issues/81
     def _result_from_response(self, response):
-        response.data.decode('utf-8')
+        try:
+            response.data.decode('utf-8')
+        except urllib3.exceptions.TimeoutError:
+            raise
+        except Exception as e:
+            raise etcd.EtcdException('Unable to decode server response: %s' % e)
         return super(Client, self)._result_from_response(response)
 
     def _get_machines_cache_from_srv(self, discovery_srv):
@@ -131,7 +136,7 @@ def catch_etcd_errors(func):
     def wrapper(*args, **kwargs):
         try:
             return not func(*args, **kwargs) is None
-        except etcd.EtcdException:
+        except (RetryFailedError, etcd.EtcdException):
             return False
     return wrapper
 
@@ -142,9 +147,17 @@ class Etcd(AbstractDCS):
         super(Etcd, self).__init__(name, config)
         self.ttl = config['ttl']
         self.member_ttl = config.get('member_ttl', 3600)
+        self._retry = Retry(deadline=10, max_delay=1, max_tries=-1,
+                            retry_exceptions=(etcd.EtcdConnectionFailed,
+                                              etcd.EtcdLeaderElectionInProgress,
+                                              etcd.EtcdWatcherCleared,
+                                              etcd.EtcdEventIndexCleared))
         self.client = self.get_etcd_client(config)
         self.cluster = None
 
+    def retry(self, *args, **kwargs):
+        return self._retry.copy()(*args, **kwargs)
+
     def get_etcd_client(self, config):
         client = None
         while not client:
@@ -162,7 +175,7 @@ class Etcd(AbstractDCS):
 
     def get_cluster(self):
         try:
-            result = self.client.read(self.client_path(''), recursive=True)
+            result = self.retry(self.client.read, self.client_path(''), recursive=True)
             nodes = {os.path.relpath(node.key, result.key): node for node in result.leaves}
 
             # get initialize flag
@@ -193,17 +206,22 @@ class Etcd(AbstractDCS):
 
     @catch_etcd_errors
     def touch_member(self, connection_string, ttl=None):
-        return self.client.set(self.client_path('/members/' + self._name), connection_string, ttl or self.member_ttl)
+        return self.retry(self.client.set, self.client_path('/members/' + self._name),
+                          connection_string, ttl or self.member_ttl)
 
     @catch_etcd_errors
     def take_leader(self):
-        return self.client.set(self.client_path('/leader'), self._name, self.ttl)
+        return self.retry(self.client.set, self.client_path('/leader'), self._name, self.ttl)
 
-    @catch_etcd_errors
     def attempt_to_acquire_leader(self):
-        ret = self.client.write(self.client_path('/leader'), self._name, ttl=self.ttl, prevExist=False)
-        ret or logger.info('Could not take out TTL lock')
-        return ret
+        try:
+            return not self.retry(self.client.write, self.client_path('/leader'),
+                                  self._name, ttl=self.ttl, prevExist=False) is None
+        except etcd.EtcdAlreadyExist:
+            logger.info('Could not take out TTL lock')
+        except (RetryFailedError, etcd.EtcdException):
+            pass
+        return False
 
     @catch_etcd_errors
     def write_leader_optime(self, state_handler):
@@ -211,13 +229,13 @@ class Etcd(AbstractDCS):
 
     @catch_etcd_errors
     def update_leader(self, state_handler):
-        ret = self.client.test_and_set(self.client_path('/leader'), self._name, self._name, self.ttl)
+        ret = self.retry(self.client.test_and_set, self.client_path('/leader'), self._name, self._name, self.ttl)
         ret and self.write_leader_optime(state_handler)
         return ret
 
     @catch_etcd_errors
     def race(self, path):
-        return self.client.write(self.client_path(path), self._name, prevExist=False)
+        return self.retry(self.client.write, self.client_path(path), self._name, prevExist=False)
 
     @catch_etcd_errors
     def delete_leader(self):

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)

patroni.py

@@ -94,8 +94,8 @@ class Patroni:
                     self.ha.state_handler.drop_replication_slots()
             except:
                 logger.exception('Exception when changing replication slots')
-            self.schedule_next_run()
             reap_children()
+            self.schedule_next_run()
 
 
 def main():

tests/test_etcd.py

@@ -26,6 +26,10 @@ class MockResponse:
 
     @property
     def data(self):
+        if self.content == 'TimeoutError':
+            raise urllib3.exceptions.TimeoutError
+        if self.content == 'Exception':
+            raise Exception
         return self.content
 
     @property
@@ -76,6 +80,8 @@ def etcd_watch(key, index=None, timeout=None, recursive=None):
 
 
 def etcd_write(key, value, **kwargs):
+    if key == '/service/exists/leader':
+        raise etcd.EtcdAlreadyExist
     if key == '/service/test/leader':
         if kwargs.get('prevValue', None) == 'foo' or not kwargs.get('prevExist', True):
             return True
@@ -190,6 +196,15 @@ class TestClient(unittest.TestCase):
         self.assertEquals(self.client.get_srv_record('blabla'), [])
         self.assertEquals(self.client.get_srv_record('exception'), [])
 
+    def test__result_from_response(self):
+        response = MockResponse()
+        response.content = 'TimeoutError'
+        self.assertRaises(urllib3.exceptions.TimeoutError, self.client._result_from_response, response)
+        response.content = 'Exception'
+        self.assertRaises(etcd.EtcdException, self.client._result_from_response, response)
+        response.content = b'{}'
+        self.assertRaises(etcd.EtcdException, self.client._result_from_response, response)
+
     def test__get_machines_cache_from_srv(self):
         self.client.get_srv_record = lambda e: [('localhost', 2380)]
         self.client._get_machines_cache_from_srv('blabla')
@@ -242,6 +257,12 @@ class TestEtcd(unittest.TestCase):
     def test_take_leader(self):
         self.assertFalse(self.etcd.take_leader())
 
+    def testattempt_to_acquire_leader(self):
+        self.etcd._base_path = '/service/exists'
+        self.assertFalse(self.etcd.attempt_to_acquire_leader())
+        self.etcd._base_path = '/service/failed'
+        self.assertFalse(self.etcd.attempt_to_acquire_leader())
+
     def test_update_leader(self):
         self.assertTrue(self.etcd.update_leader(MockPostgresql()))
 

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)