diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index cb13f4e7..51110a90 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -658,8 +658,10 @@ 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) + result = self._client.watch(self.leader_path, index=leader_index, timeout=timeout + 0.5) self._has_failed = False + if result.action == 'compareAndSwap': + time.sleep(0.01) # Synchronous work of all cluster members with etcd is less expensive # than reestablishing http connection every time from every replica. return True diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index 5a852547..af247669 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -70,6 +70,7 @@ class ZooKeeper(AbstractDCS): self._client.add_listener(self.session_listener) self._fetch_cluster = True + self._fetch_optime = True self._orig_kazoo_connect = self._client._connection._connect self._client._connection._connect = self._kazoo_connect @@ -95,9 +96,13 @@ class ZooKeeper(AbstractDCS): if state in [KazooState.SUSPENDED, KazooState.LOST]: self.cluster_watcher(None) + def optime_watcher(self, event): + self._fetch_optime = True + self.event.set() + def cluster_watcher(self, event): self._fetch_cluster = True - self.event.set() + self.optime_watcher(event) def reload_config(self, config): self.set_retry_timeout(config['retry_timeout']) @@ -141,6 +146,12 @@ class ZooKeeper(AbstractDCS): except NoNodeError: return None + def get_leader_optime(self, leader): + watch = self.optime_watcher if not leader or leader.name != self._name else None + optime = self.get_node(self.leader_optime_path, watch) + self._fetch_optime = False + return optime and int(optime[0]) or 0 + @staticmethod def member(name, value, znode): return Member.from_node(znode.version, name, znode.ephemeralOwner, value) @@ -178,10 +189,6 @@ class ZooKeeper(AbstractDCS): history = self.get_node(self.history_path, watch=self.cluster_watcher) if self._HISTORY in nodes else None history = history and TimelineHistory.from_node(history[1].mzxid, history[0]) - # get last leader operation - last_leader_operation = self._OPTIME in nodes and self._fetch_cluster and self.get_node(self.leader_optime_path) - last_leader_operation = last_leader_operation and int(last_leader_operation[0]) or 0 - # get synchronization state sync = self.get_node(self.sync_path, watch=self.cluster_watcher) if self._SYNC in nodes else None sync = SyncState.from_node(sync and sync[1].version, sync and sync[0]) @@ -206,6 +213,9 @@ class ZooKeeper(AbstractDCS): leader = Leader(leader[1].version, leader[1].ephemeralOwner, member) self._fetch_cluster = member.index == -1 + # get last leader operation + last_leader_operation = self._OPTIME in nodes and self.get_leader_optime(leader) + # failover key failover = self.get_node(self.failover_path, watch=self.cluster_watcher) if self._FAILOVER in nodes else None failover = failover and Failover.from_node(failover[1].version, failover[0]) @@ -221,6 +231,15 @@ class ZooKeeper(AbstractDCS): logger.exception('get_cluster') self.cluster_watcher(None) raise ZooKeeperError('ZooKeeper in not responding properly') + # Optime ZNode was updated or doesn't exist and we are not leader + elif (self._fetch_optime and not self._fetch_cluster or not cluster.last_leader_operation) and\ + not (cluster.leader and cluster.leader.name == self._name): + try: + optime = self.get_leader_optime(cluster.leader) + cluster = Cluster(cluster.initialize, cluster.config, cluster.leader, optime, + cluster.members, cluster.failover, cluster.sync, cluster.history) + except Exception: + pass return cluster def _bypass_caches(self): @@ -349,6 +368,6 @@ class ZooKeeper(AbstractDCS): return self.set_sync_state_value("{}", index) def watch(self, leader_index, timeout): - if super(ZooKeeper, self).watch(leader_index, timeout): + if super(ZooKeeper, self).watch(leader_index, timeout) and not self._fetch_optime: self._fetch_cluster = True return self._fetch_cluster diff --git a/patroni/ha.py b/patroni/ha.py index d144dd02..72bd72b1 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -1411,7 +1411,7 @@ class Ha(object): def watch(self, timeout): # watch on leader key changes if the postgres is running and leader is known and current node is not lock owner - if self._async_executor.busy or self.cluster.is_unlocked() or self.has_lock(False): + if self._async_executor.busy or not self.cluster or self.cluster.is_unlocked() or self.has_lock(False): leader_index = None else: leader_index = self.cluster.leader.index diff --git a/tests/test_etcd.py b/tests/test_etcd.py index ceef0d2d..258ab9a7 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -17,7 +17,7 @@ def etcd_watch(self, key, index=None, timeout=None, recursive=None): if timeout == 2.0: raise etcd.EtcdWatchTimedOut elif timeout == 5.0: - return etcd.EtcdResult('delete', {}) + return etcd.EtcdResult('compareAndSwap', {}) elif 5 < timeout <= 10.0: raise etcd.EtcdException elif timeout == 20.0: @@ -285,7 +285,8 @@ class TestEtcd(unittest.TestCase): self.etcd.watch(None, 0) self.etcd.get_cluster() self.etcd.watch(20729, 1.5) - self.etcd.watch(20729, 4.5) + with patch('time.sleep', Mock()): + self.etcd.watch(20729, 4.5) with patch.object(AbstractDCS, 'watch', Mock()): self.assertTrue(self.etcd.watch(20729, 19.5)) self.assertRaises(SleepException, self.etcd.watch, 20729, 9.5) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index e121de93..3df1ad3e 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -37,6 +37,8 @@ class MockKazooClient(Mock): b'postgres://repuser:rep-pass@localhost:5434/postgres?application_name=http://127.0.0.1:8009/patroni', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0 if self.exists else -1, 0, 0, 0) ) + elif path.endswith('/optime/leader'): + return (b'500', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0)) elif path.endswith('/leader'): if self.leader: return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0)) @@ -153,6 +155,12 @@ class TestZooKeeper(unittest.TestCase): cluster = self.zk.get_cluster(True) self.assertIsInstance(cluster.leader, Leader) self.zk.touch_member({'foo': 'foo'}) + self.zk._name = 'bar' + self.zk.optime_watcher(None) + with patch.object(ZooKeeper, 'get_node', Mock(side_effect=Exception)): + self.zk.get_cluster() + cluster = self.zk.get_cluster() + self.assertEqual(cluster.last_leader_operation, 500) def test_delete_leader(self): self.assertTrue(self.zk.delete_leader()) @@ -213,6 +221,7 @@ class TestZooKeeper(unittest.TestCase): def test_watch(self): self.zk.watch(None, 0) self.zk.event.isSet = Mock(return_value=True) + self.zk._fetch_optime = False self.zk.watch(None, 0) def test__kazoo_connect(self):