Little fixes in etcd3 and kubernetes (#2689)

- Always pass etcd3 key revision as a string
- Make sure the leader key isn't unconditionally overwritten. It may happen that the leader heart-beat loop didn't run properly and the session has expired. In this case the leader may create a new session and a new leader key. But, there are chances that the other node already created a leader key and we don't want to overwrite it.
- Try to sync HA loops between nodes by adding 0.5 seconds to timeout on non-leader nodes

patroni/dcs/etcd3.py

@@ -813,7 +813,7 @@ class Etcd3(AbstractEtcd):
             return retry(*args, **kwargs)
 
         try:
-            return _retry(self._client.put, self.leader_path, self._name, self._lease, 0)
+            return _retry(self._client.put, self.leader_path, self._name, self._lease, '0')
         except LeaseNotFound:
             logger.error('Our lease disappeared from Etcd. Will try to get a new one and retry attempt')
             self._lease = None
@@ -825,7 +825,7 @@ class Etcd3(AbstractEtcd):
             if retry.deadline < 1:
                 raise Etcd3Error('_do_attempt_to_acquire_leader timeout')
 
-            return _retry(self._client.put, self.leader_path, self._name, self._lease, 0)
+            return _retry(self._client.put, self.leader_path, self._name, self._lease, '0')
 
     @catch_return_false_exception
     def attempt_to_acquire_leader(self) -> bool:
@@ -886,14 +886,14 @@ class Etcd3(AbstractEtcd):
 
                 try:
                     self._run_and_handle_exceptions(self._client.put, self.leader_path,
-                                                    self._name, self._lease, retry=_retry)
+                                                    self._name, self._lease, '0', retry=_retry)
                 except ReturnFalseException:
                     pass
         return bool(self._lease)
 
     @catch_etcd_errors
     def initialize(self, create_new: bool = True, sysid: str = ""):
-        return self.retry(self._client.put, self.initialize_path, sysid, None, 0 if create_new else None)
+        return self.retry(self._client.put, self.initialize_path, sysid, None, '0' if create_new else None)
 
     @catch_etcd_errors
     def _delete_leader(self) -> bool:
@@ -928,6 +928,10 @@ class Etcd3(AbstractEtcd):
             self.__do_not_watch = False
             return True
 
+        # We want to give a bit more time to non-leader nodes to synchronize HA loops
+        if leader_version:
+            timeout += 0.5
+
         try:
             return super(Etcd3, self).watch(None, timeout)
         finally:

patroni/dcs/kubernetes.py

@@ -1350,7 +1350,11 @@ class Kubernetes(AbstractDCS):
             self.__do_not_watch = False
             return True
 
+        # We want to give a bit more time to non-leader nodes to synchronize HA loops
+        if leader_version:
+            timeout += 0.5
+
         try:
-            return super(Kubernetes, self).watch(None, timeout + 0.5)
+            return super(Kubernetes, self).watch(None, timeout)
         finally:
             self.event.clear()

tests/test_etcd3.py

@@ -306,7 +306,7 @@ class TestEtcd3(BaseTestEtcd3):
     def test_watch(self):
         self.etcd3.set_ttl(10)
         self.etcd3.watch(None, 0)
-        self.etcd3.watch(None, 0)
+        self.etcd3.watch('5', 0)
 
     def test_set_socket_options(self):
         with patch('socket.SIO_KEEPALIVE_VALS', 1, create=True):

tests/test_kubernetes.py

@@ -315,7 +315,7 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
     def test_watch(self):
         self.k.set_ttl(10)
         self.k.watch(None, 0)
-        self.k.watch(None, 0)
+        self.k.watch('5', 0)
 
     def test_set_history_value(self):
         self.k.set_history_value('{}')