From 04b9fb9dd4c082a7a009e50e67641191e0ca9a5d Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 15 Jul 2020 10:31:32 +0200 Subject: [PATCH] Make sure cached last_leader_operation is up-to-date on replicas (#1600) Patroni is caching the cluster view in the DCS object because not all operations require the most up-to-date values. The cached version is valid for TTL seconds. So far it worked quite well, the only known problem was that the `last_leader_operation` for some DCS implementations was not very up-to-date: * Etcd: since the `/optime/leader` key is updated right after the `/leader` key, usually all replicas get the value from the previous HA loop. Therefore the value is somewhere between `loop_wait` and `loop_wait*2` old. We improve it by using the 10ms artificial sleep after receiving watch notification from `compareAndSwap` operation on the leader key. It usually gives enough time for the primary to update the `/optime/leader`. On average that makes the cached version `loop_wait/2` old. * ZooKeeper: Patroni itself is not so much interested in most up-to-date values of member and leader/optime ZNodes. In case of the leader race it just reads everything from ZooKeeper, but during normal operation it is relying on cache. In order to see the recent value on replicas they are doing watch on the `leader/optime` Znode and will re-read it after it was updated by the primary. On average that makes the cached version `loop_wait/2` old. * Kubernetes: last_leader_operation is stored in the same object as the leader key itself and therefore update is atomic and we always see the latest version. That makes the cached version `loop_wait/2` old on avg. * Consul: HA loops on the primary and replicas are not synchronized, therefore at the moment when we read the cluster state from the Consul KV we see the last_leader_operation value that is between 0 and loop_wait old. On average that makes the cached version `loop_wait` old. Unfortunately we can't make it much better without performing periodic updates from Consul, which might have negative side effects. Since the `optime/leader` is only updated at most once per HA loop cycle, the value stored in the DCS is usually `loop_wait/2` old on avg. For majority of DCS implementations we could promise that the cached version in Patroni will match the value in DCS most of the time, therefore there is no need to make additional requests. The only exception is Consul, but probably we could just document it, so when someone relying on last_leader_operation value to check the replication lag can correspondingly adjust thresholds. Will help to implement #1599 --- patroni/dcs/etcd.py | 4 +++- patroni/dcs/zookeeper.py | 31 +++++++++++++++++++++++++------ patroni/ha.py | 2 +- tests/test_etcd.py | 5 +++-- tests/test_zookeeper.py | 9 +++++++++ 5 files changed, 41 insertions(+), 10 deletions(-) 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):