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

@@ -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()
 

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'))