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/etcd.py b/helpers/etcd.py index 67ac9c92..74ad36f8 100644 --- a/helpers/etcd.py +++ b/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): 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/patroni.py b/patroni.py index cb605275..f0824285 100755 --- a/patroni.py +++ b/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(): diff --git a/tests/test_etcd.py b/tests/test_etcd.py index c42394fa..f5d65e5f 100644 --- a/tests/test_etcd.py +++ b/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())) 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)