From b4afc6830b5c5584c7354d6f966375881e5993c0 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 24 May 2023 10:54:26 +0200 Subject: [PATCH] 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 | 12 ++++++++---- patroni/dcs/kubernetes.py | 6 +++++- tests/test_etcd3.py | 2 +- tests/test_kubernetes.py | 2 +- 4 files changed, 15 insertions(+), 7 deletions(-) diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index 308bb2c3..55923df3 100644 --- a/patroni/dcs/etcd3.py +++ b/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: diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 3e334e9d..3630ae1e 100644 --- a/patroni/dcs/kubernetes.py +++ b/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() diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index 85c2173a..6f56d0e6 100644 --- a/tests/test_etcd3.py +++ b/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): diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index bbe55157..fbefb448 100644 --- a/tests/test_kubernetes.py +++ b/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('{}')