From 856a13e24c9e81fa23857e69a0434b34b771bb2f Mon Sep 17 00:00:00 2001 From: Ants Aasma Date: Thu, 20 Apr 2017 13:40:15 +0300 Subject: [PATCH] Remove error spinning on etcd failure and reduce log spam (#429) When all etcd servers refuse connections during watch the call will fail with an exception and will be immediately retried. This creates a huge amount of log spam potentially creating additional issues on top of losing the DCS. This patch takes note if etcd failures are repeating and starting from the second failure will sleep for a second before retrying. It additionally omits the stack trace after the first failure in a streak of failures. --- patroni/dcs/etcd.py | 52 +++++++++++++++++++++++++++++---------------- tests/test_etcd.py | 5 +++-- 2 files changed, 37 insertions(+), 20 deletions(-) diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 3aececdd..8a6ec569 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -296,19 +296,6 @@ class Client(etcd.Client): self._machines_cache_updated = time.time() -def catch_etcd_errors(func): - def wrapper(*args, **kwargs): - try: - return func(*args, **kwargs) is not None - except (RetryFailedError, etcd.EtcdException): - return False - except: - logger.exception("") - raise EtcdError("unexpected error") - - return wrapper - - class Etcd(AbstractDCS): def __init__(self, config): @@ -320,10 +307,36 @@ class Etcd(AbstractDCS): etcd.EtcdEventIndexCleared)) self._client = self.get_etcd_client(config) self.__do_not_watch = False + self._has_failed = False def retry(self, *args, **kwargs): return self._retry.copy()(*args, **kwargs) + def _handle_exception(self, e, name='', do_sleep=False, raise_ex=None): + if not self._has_failed: + logger.exception(name) + else: + logger.error(e) + if do_sleep: + time.sleep(1) + self._has_failed = True + if isinstance(raise_ex, Exception): + raise raise_ex + + def catch_etcd_errors(func): + def wrapper(self, *args, **kwargs): + try: + retval = func(self, *args, **kwargs) is not None + self._has_failed = False + return retval + except (RetryFailedError, etcd.EtcdException) as e: + self._handle_exception(e) + return False + except Exception as e: + self._handle_exception(e, raise_ex=EtcdError('unexpected error')) + + return wrapper + @staticmethod def get_etcd_client(config): if 'proxy' in config: @@ -448,9 +461,9 @@ class Etcd(AbstractDCS): self._cluster = Cluster(initialize, config, leader, last_leader_operation, members, failover, sync) except etcd.EtcdKeyNotFound: self._cluster = Cluster(None, None, None, None, [], None, None) - except: - logger.exception('get_cluster') - raise EtcdError('Etcd is not responding properly') + except Exception as e: + self._handle_exception(e, 'get_cluster', raise_ex=EtcdError('Etcd is not responding properly')) + self._has_failed = False @catch_etcd_errors def touch_member(self, data, ttl=None, permanent=False): @@ -524,16 +537,19 @@ class Etcd(AbstractDCS): while timeout >= 1: # when timeout is too small urllib3 doesn't have enough time to connect try: self._client.watch(self.leader_path, index=leader_index, timeout=timeout + 0.5) + self._has_failed = False # Synchronous work of all cluster members with etcd is less expensive # than reestablishing http connection every time from every replica. return True except etcd.EtcdWatchTimedOut: self._client.http.clear() + self._has_failed = False return False except (etcd.EtcdEventIndexCleared, etcd.EtcdWatcherCleared): # Watch failed + self._has_failed = False return True # leave the loop, because watch with the same parameters will fail anyway - except etcd.EtcdException: - logger.exception('watch') + except etcd.EtcdException as e: + self._handle_exception(e, 'watch', True) timeout = end_time - time.time() diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 38513982..b9ba6475 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -59,7 +59,7 @@ def etcd_watch(self, key, index=None, timeout=None, recursive=None): raise etcd.EtcdWatchTimedOut elif timeout == 5.0: return etcd.EtcdResult('delete', {}) - elif timeout == 10.0: + elif 5 < timeout <= 10.0: raise etcd.EtcdException elif timeout == 20.0: raise etcd.EtcdEventIndexCleared @@ -302,6 +302,7 @@ class TestEtcd(unittest.TestCase): def test_delete_cluster(self): self.assertFalse(self.etcd.delete_cluster()) + @patch('time.sleep', Mock(side_effect=SleepException)) @patch.object(etcd.Client, 'watch', etcd_watch) def test_watch(self): self.etcd.watch(None, 0) @@ -310,7 +311,7 @@ class TestEtcd(unittest.TestCase): self.etcd.watch(20729, 4.5) with patch.object(AbstractDCS, 'watch', Mock()): self.assertTrue(self.etcd.watch(20729, 19.5)) - self.etcd.watch(20729, 9.5) + self.assertRaises(SleepException, self.etcd.watch, 20729, 9.5) def test_other_exceptions(self): self.etcd.retry = Mock(side_effect=AttributeError('foo'))