From 7f8e95b334a4b07db3dee04d92dcdb353be76234 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 16 Sep 2015 10:38:34 +0200 Subject: [PATCH 1/3] 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 | 6 ++---- patroni/dcs.py | 7 +++++++ patroni/etcd.py | 15 ++++++++------- patroni/zookeeper.py | 2 ++ tests/test_patroni.py | 20 ++++++++++++++++++-- tests/test_postgresql.py | 4 ---- tests/test_zookeeper.py | 2 ++ 7 files changed, 39 insertions(+), 17 deletions(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index cc008fda..b9ec980e 100644 --- a/patroni/__init__.py +++ b/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() diff --git a/patroni/dcs.py b/patroni/dcs.py index 9017da7f..ed876de0 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 + + :returns: `!True` if you would like reschedule 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 eae810ed..9dfe47c4 100644 --- a/patroni/zookeeper.py +++ b/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 diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 317e1e5c..7257cc64 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): @@ -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:rep-pass@127.0.0.1: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() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 56dbc557..37e10de0 100644 --- a/tests/test_postgresql.py +++ b/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): 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) From c240fd1ee87af1a5662feee308615130dac5ef2c Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 16 Sep 2015 11:09:53 +0200 Subject: [PATCH 2/3] More documentation for watch method --- patroni/dcs.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/patroni/dcs.py b/patroni/dcs.py index ed876de0..908386b5 100644 --- a/patroni/dcs.py +++ b/patroni/dcs.py @@ -185,8 +185,8 @@ class AbstractDCS: """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 - """ + :param timeout: timeout in seconds + :returns: `!True` if you would like to reschedule the next run of ha cycle""" sleep(timeout) return False From 246f0dbbafb173e2cb0a1e56984fdb03d7bc321b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 16 Sep 2015 14:23:47 +0200 Subject: [PATCH 3/3] watch should return true if you are not leader --- patroni/zookeeper.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/patroni/zookeeper.py b/patroni/zookeeper.py index 9dfe47c4..f8d0958e 100644 --- a/patroni/zookeeper.py +++ b/patroni/zookeeper.py @@ -245,5 +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 not self.cluster or not self.cluster.leader or self.cluster.leader.name != self._name return False