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

@@ -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

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

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

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)

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):