Next run of ha cycle is rescheduled depending on return value of watch

Current etcd implementation does not yet support timeout option when
`wait=true`: https://github.com/coreos/etcd/issues/2468

Originaly I've implemented `watch` method for `Etcd` class in a
following manner: if the leader key was updated just because master
needs to update ttl and watch timeout is not yet expired, I was
recalculating timeout and starting `watch` call once again.
Usually after "restart" we were getting urllib3.exceptions.TimeoutError.
The only possible way to recover after such exception - close socket and
establish a new connection. With pure http it's relatively cheap, but
with https and some kind of authorization on etcd side it would became
rather expensive and should be avoided.

patroni/__init__.py

@@ -84,15 +84,13 @@ class Patroni:
             self.postgresql.load_replication_slots()
 
     def schedule_next_run(self):
-        if self.postgresql.is_promoted:
-            self.next_run = time.time()
         self.next_run += self.nap_time
         current_time = time.time()
         nap_time = self.next_run - current_time
         if nap_time <= 0:
             self.next_run = current_time
-        else:
-            self.ha.dcs.watch(nap_time)
+        elif self.ha.dcs.watch(nap_time):
+            self.next_run = time.time()
 
     def run(self):
         self.api.start()

patroni/dcs.py

@@ -182,4 +182,11 @@ class AbstractDCS:
         """ Removes the initialize key for a cluster """
 
     def watch(self, timeout):
+        """If the current node is a master it should just sleep.
+        Any other node should watch for changes of leader key with a given timeout
+
+        :returns: `!True` if you would like reschedule next run of ha cycle
+        """
+
         sleep(timeout)
+        return False

patroni/etcd.py

@@ -248,18 +248,19 @@ class Etcd(AbstractDCS):
         if self.cluster and self.cluster.leader and self.cluster.leader.name != self._name:
             end_time = time.time() + timeout
             index = self.cluster.leader.index
+
             while index and timeout >= 1:  # when timeout is too small urllib3 doesn't have enough time to connect
                 try:
-                    res = self.client.watch(self.leader_path, index=index + 1, timeout=timeout)
-                    if res.action not in ['set', 'compareAndSwap'] or res.value != self.cluster.leader.name:
-                        return
-                    index = res.modifiedIndex
+                    self.client.watch(self.leader_path, index=index + 1, timeout=timeout)
+                    # Synchronous work of all cluster members with etcd is less expensive
+                    # than reestablishing http connection every time from every replica.
+                    return True
                 except urllib3.exceptions.TimeoutError:
                     self.client.http.clear()
-                    return
+                    return False
                 except etcd.EtcdException:
-                    index = None
+                    logging.exception('watch')
 
                 timeout = end_time - time.time()
 
-        timeout > 0 and super(Etcd, self).watch(timeout)
+        return timeout > 0 and super(Etcd, self).watch(timeout)

patroni/zookeeper.py

@@ -245,3 +245,5 @@ class ZooKeeper(AbstractDCS):
         self.cluster_event.wait(timeout)
         if self.cluster_event.isSet():
             self.fetch_cluster = True
+            return self.cluster and self.cluster.leader and self.cluster.leader.name != self._name
+        return False

tests/test_patroni.py

@@ -11,7 +11,7 @@ from mock import Mock, patch
 from patroni.api import RestApiServer
 from patroni.dcs import Cluster, Member, Leader
 from patroni.etcd import Etcd
-from patroni.exceptions import PostgresException
+from patroni.exceptions import DCSError, PostgresException
 from patroni import Patroni, main
 from patroni.zookeeper import ZooKeeper
 from six.moves import BaseHTTPServer
@@ -33,6 +33,10 @@ def time_sleep(*args):
     raise SleepException()
 
 
+def keyboard_interrupt(*args):
+    raise KeyboardInterrupt
+
+
 class Mock_BaseServer__is_shut_down:
 
     def set(self):
@@ -61,11 +65,15 @@ def get_cluster_not_initialized_with_leader():
 
 
 def get_cluster_initialized_with_leader():
-    return get_cluster(True, Leader(0, 0, 0, 
+    return get_cluster(True, Leader(0, 0, 0,
                        Member(0, 'leader', 'postgres://replicator:[email protected]:5435/postgres',
                               None, None, 28)))
 
 
+def get_cluster_dcs_error():
+    raise DCSError('')
+
+
 class TestPatroni(unittest.TestCase):
 
     def __init__(self, method_name='runTest'):
@@ -122,6 +130,9 @@ class TestPatroni(unittest.TestCase):
 
             self.assertRaises(SleepException, main)
 
+            Patroni.run = keyboard_interrupt
+            main()
+
             Patroni.run = run
             Patroni.touch_member = touch_member
 
@@ -178,7 +189,12 @@ class TestPatroni(unittest.TestCase):
         self.p.postgresql.data_directory_empty = true
         self.p.initialize()
 
+        self.p.ha.dcs.get_cluster = get_cluster_dcs_error
+        self.assertRaises(SleepException, self.p.initialize)
+
     def test_schedule_next_run(self):
+        self.p.ha.dcs.watch = lambda e: True
+        self.p.schedule_next_run()
         self.p.next_run = time.time() - self.p.nap_time - 1
         self.p.schedule_next_run()
 

tests/test_postgresql.py

@@ -17,10 +17,6 @@ def subprocess_call(cmd, shell=False, env=None):
     return 0
 
 
-def false(*args, **kwargs):
-    return False
-
-
 class MockCursor:
 
     def __init__(self):

tests/test_zookeeper.py

@@ -192,3 +192,5 @@ class TestZooKeeper(unittest.TestCase):
 
     def test_watch(self):
         self.zk.watch(0)
+        self.zk.cluster_event.isSet = lambda: False
+        self.zk.watch(0)