From 8f3ed0088643543123fa1373b3748ae2170a1814 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 31 Jul 2023 10:16:19 +0200 Subject: [PATCH] Invalidate cache if txn failed due to revision mismatch (#2783) It was reported in #2779 that the primary was constantly logging messages like `Synchronous replication key updated by someone else`. It happened after Patroni was stuck due to resource starvation. Key updates are performed using create_revision/mod_revision field, which value is taken from the internal cached. Hence, it is a clear symptom of stale cache. Similar issues in K8s implementation were addressed by invalidating the cache and restarting watcher connections every time when update failed due to resource_version mismatch, so we do the same for Etcd3. --- patroni/dcs/etcd3.py | 10 ++++++++++ tests/test_etcd3.py | 8 +++++++- 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index 0bfc4fae..e28da844 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -630,6 +630,16 @@ class PatroniEtcd3Client(Etcd3Client): return ret + def txn(self, compare: Dict[str, Any], success: Dict[str, Any], + failure: Optional[Dict[str, Any]] = None, retry: Optional[Retry] = None) -> Dict[str, Any]: + ret = super(PatroniEtcd3Client, self).txn(compare, success, failure, retry) + # Here we abuse the fact that the `failure` is only set in the call from update_leader(). + # In all other cases the txn() call failure may be an indicator of a stale cache, + # and therefore we want to restart watcher. + if not failure and not ret: + self._restart_watcher() + return ret + class Etcd3(AbstractEtcd): diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index ace59a62..2e3ed59d 100644 --- a/tests/test_etcd3.py +++ b/tests/test_etcd3.py @@ -127,10 +127,16 @@ class TestPatroniEtcd3Client(BaseTestEtcd3): request = {'key': base64_encode('/patroni/test/leader')} mock_urlopen.return_value = MockResponse() mock_urlopen.return_value.content = '{"succeeded":true,"header":{"revision":"1"}}' - self.client.call_rpc('/kv/txn', {'success': [{'request_delete_range': request}]}) self.client.call_rpc('/kv/put', request) self.client.call_rpc('/kv/deleterange', request) + @patch.object(urllib3.PoolManager, 'urlopen') + def test_txn(self, mock_urlopen): + mock_urlopen.return_value = MockResponse() + mock_urlopen.return_value.content = '{"header":{"revision":"1"}}' + self.client.txn({'target': 'MOD', 'mod_revision': '1'}, + {'request_delete_range': {'key': base64_encode('/patroni/test/leader')}}) + @patch('time.time', Mock(side_effect=[1, 10.9, 100])) def test__wait_cache(self): with self.kv_cache.condition: