Merge pull request #39 from zalando/feature/reschedule-next-run

Feature/reschedule next run of ha cycle depending on return value of `watch`

patroni/__init__.py

@@ -89,8 +89,8 @@ class Patroni:
         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
+
+        :param timeout: timeout in seconds
+        :returns: `!True` if you would like to reschedule the 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

@@ -246,3 +246,5 @@ class ZooKeeper(AbstractDCS):
         self.cluster_event.wait(timeout)
         if self.cluster_event.isSet():
             self.fetch_cluster = True
+            return not self.cluster or not self.cluster.leader or 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):
@@ -66,6 +70,10 @@ def get_cluster_initialized_with_leader():
                               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_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)