diff --git a/patroni/__init__.py b/patroni/__init__.py index 491edfee..c190e2de 100644 --- a/patroni/__init__.py +++ b/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() diff --git a/patroni/dcs.py b/patroni/dcs.py index 9017da7f..908386b5 100644 --- a/patroni/dcs.py +++ b/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 diff --git a/patroni/etcd.py b/patroni/etcd.py index e24fa66b..c0950c85 100644 --- a/patroni/etcd.py +++ b/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) diff --git a/patroni/zookeeper.py b/patroni/zookeeper.py index bc16a8ea..9cd15fbc 100644 --- a/patroni/zookeeper.py +++ b/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 diff --git a/tests/test_patroni.py b/tests/test_patroni.py index bdba1fa0..1c83eeff 100644 --- a/tests/test_patroni.py +++ b/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() diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index b9defcb3..f2141707 100644 --- a/tests/test_zookeeper.py +++ b/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)