From a8cfd46801c8de7bdfb0b68835148124da0db4f5 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 7 Mar 2024 16:31:35 +0530 Subject: [PATCH] Retry one time on Etcd3 auth error (#3026) But do it only in case if we didn't authenticate right before executing a request. Previously retries only happened when the caller was executed with `Retry.__call__()`, which is not the case for methods like `set_failover_value()` or `set_config_value()`. Also, it seems that existing watchers aren't affected, therefore we will not restart them after reauthentication. In addition to that fix issues with `Retry.ensure_deadline(0)`: 1. the return value was ignored 2. we don't have to set `Retry.deadline` attr, it is not used anywhere Close https://github.com/zalando/patroni/issues/3023 --- patroni/dcs/consul.py | 9 ++++++--- patroni/dcs/etcd3.py | 45 ++++++++++++++++--------------------------- patroni/utils.py | 5 ++--- tests/test_consul.py | 6 ++---- tests/test_etcd3.py | 12 +++++++----- 5 files changed, 34 insertions(+), 43 deletions(-) diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index bbfd2453..c6747d06 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -577,14 +577,17 @@ class Consul(AbstractDCS): try: return retry(self._client.kv.put, self.leader_path, self._name, acquire=self._session) except InvalidSession: - logger.error('Our session disappeared from Consul. Will try to get a new one and retry attempt') self._session = None - retry.ensure_deadline(0) + + if not retry.ensure_deadline(0): + logger.error('Our session disappeared from Consul. Deadline exceeded, giving up') + return False + + logger.error('Our session disappeared from Consul. Will try to get a new one and retry attempt') retry(self._do_refresh_session) retry.ensure_deadline(1, ConsulError('_do_attempt_to_acquire_leader timeout')) - return retry(self._client.kv.put, self.leader_path, self._name, acquire=self._session) @catch_return_false_exception diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index 946f9318..4a840c7f 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -198,12 +198,6 @@ def build_range_request(key: str, range_end: Union[bytes, str, None] = None) -> return fields -class ReAuthenticateMode(IntEnum): - NOT_REQUIRED = 0 - REQUIRED = 1 - WITHOUT_WATCHER_RESTART = 2 - - def _handle_auth_errors(func: Callable[..., Any]) -> Any: def wrapper(self: 'Etcd3Client', *args: Any, **kwargs: Any) -> Any: return self.handle_auth_errors(func, *args, **kwargs) @@ -215,7 +209,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover): ERROR_CLS = Etcd3Error def __init__(self, config: Dict[str, Any], dns_resolver: DnsCachingResolver, cache_ttl: int = 300) -> None: - self._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED + self._reauthenticate = False self._token = None self._cluster_version: Tuple[int, ...] = tuple() super(Etcd3Client, self).__init__({**config, 'version_prefix': '/v3beta'}, dns_resolver, cache_ttl) @@ -294,7 +288,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover): fields['retry'] = retry return self.api_execute(self.version_prefix + method, self._MPOST, fields) - def authenticate(self, *, restart_watcher: bool = True, retry: Optional[Retry] = None) -> bool: + def authenticate(self, *, retry: Optional[Retry] = None) -> bool: if self._use_proxies and not self._cluster_version: kwargs = self._prepare_common_parameters(1) self._ensure_version_prefix(self._base_uri, **kwargs) @@ -316,20 +310,18 @@ class Etcd3Client(AbstractEtcdClientWithFailover): def handle_auth_errors(self: 'Etcd3Client', func: Callable[..., Any], *args: Any, retry: Optional[Retry] = None, **kwargs: Any) -> Any: + reauthenticated = False exc = None while True: - if self._reauthenticate_reason: + if self._reauthenticate: if self.username and self.password: - self.authenticate( - restart_watcher=self._reauthenticate_reason != ReAuthenticateMode.WITHOUT_WATCHER_RESTART, - retry=retry) - self._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED - if retry: - retry.ensure_deadline(0) + self.authenticate(retry=retry) + self._reauthenticate = False else: msg = 'Username or password not set, authentication is not possible' logger.fatal(msg) raise exc or Etcd3Exception(msg) + reauthenticated = True try: return func(self, *args, retry=retry, **kwargs) @@ -347,11 +339,12 @@ class Etcd3Client(AbstractEtcdClientWithFailover): except AuthOldRevision as e: logger.error('Auth token is for old revision of auth store') exc = e - self._reauthenticate_reason = ReAuthenticateMode.WITHOUT_WATCHER_RESTART \ - if isinstance(exc, AuthOldRevision) else ReAuthenticateMode.REQUIRED - if not retry: + self._reauthenticate = True + if retry: + logger.error('retry = %s', retry) + retry.ensure_deadline(0.5, exc) + elif reauthenticated: raise exc - retry.ensure_deadline(0.5, exc) @_handle_auth_errors def range(self, key: str, range_end: Union[bytes, str, None] = None, serializable: bool = True, @@ -603,12 +596,6 @@ class PatroniEtcd3Client(Etcd3Client): super(PatroniEtcd3Client, self).set_base_uri(value) self._restart_watcher() - def authenticate(self, *, restart_watcher: bool = True, retry: Optional[Retry] = None) -> bool: - ret = super(PatroniEtcd3Client, self).authenticate(restart_watcher=restart_watcher, retry=retry) - if ret and restart_watcher: - self._restart_watcher() - return ret - def _wait_cache(self, timeout: float) -> None: stop_time = time.time() + timeout while self._kv_cache and not self._kv_cache.is_ready(): @@ -866,14 +853,16 @@ class Etcd3(AbstractEtcd): try: return _retry(self._client.put, self.leader_path, self._name, self._lease, create_revision='0') except LeaseNotFound: - logger.error('Our lease disappeared from Etcd. Will try to get a new one and retry attempt') self._lease = None - retry.ensure_deadline(0) + if not retry.ensure_deadline(0): + logger.error('Our lease disappeared from Etcd. Deadline exceeded, giving up') + return False + + logger.error('Our lease disappeared from Etcd. Will try to get a new one and retry attempt') _retry(self._do_refresh_lease) retry.ensure_deadline(1, Etcd3Error('_do_attempt_to_acquire_leader timeout')) - return _retry(self._client.put, self.leader_path, self._name, self._lease, create_revision='0') @catch_return_false_exception diff --git a/patroni/utils.py b/patroni/utils.py index dc0a7c95..c40e1248 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -716,7 +716,7 @@ class Retry(object): return self._cur_stoptime or 0 def ensure_deadline(self, timeout: float, raise_ex: Optional[Exception] = None) -> bool: - """Calculates, sets, and checks the remaining deadline time. + """Calculates and checks the remaining deadline time. :param timeout: if the *deadline* is smaller than the provided *timeout* value raise *raise_ex* exception. :param raise_ex: the exception object that will be raised if the *deadline* is smaller than provided *timeout*. @@ -727,8 +727,7 @@ class Retry(object): :raises: :class:`Exception`: *raise_ex* if calculated deadline is smaller than provided *timeout*. """ - self.deadline = self.stoptime - time.time() - if self.deadline < timeout: + if self.stoptime - time.time() < timeout: if raise_ex: raise raise_ex return False diff --git a/tests/test_consul.py b/tests/test_consul.py index 494d1126..87fb4704 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -160,10 +160,8 @@ class TestConsul(unittest.TestCase): self.c.set_ttl(20) self.c._do_refresh_session = Mock() self.assertFalse(self.c.take_leader()) - with patch('time.time', Mock(side_effect=[0, 100])): - self.assertRaises(ConsulError, self.c.take_leader) - with patch('time.time', Mock(side_effect=[0, 0, 0, 0, 0, 0, 100])): - self.assertRaises(ConsulError, self.c.take_leader) + with patch('time.time', Mock(side_effect=[0, 0, 0, 100, 100])): + self.assertFalse(self.c.take_leader()) @patch.object(consul.Consul.KV, 'put', Mock(return_value=True)) def test_set_failover_value(self): diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index fcfd4e4b..3e1bd9e8 100644 --- a/tests/test_etcd3.py +++ b/tests/test_etcd3.py @@ -7,7 +7,7 @@ from mock import Mock, PropertyMock, patch from patroni.dcs import get_dcs from patroni.dcs.etcd import DnsCachingResolver from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Client, \ - Etcd3Error, Etcd3ClientError, ReAuthenticateMode, RetryFailedError, InvalidAuthToken, Unavailable, \ + Etcd3Error, Etcd3ClientError, RetryFailedError, InvalidAuthToken, Unavailable, \ Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, AuthOldRevision, base64_encode from patroni.postgresql.mpp import get_mpp from threading import Thread @@ -166,12 +166,14 @@ class TestPatroniEtcd3Client(BaseTestEtcd3): retry = self.etcd3._retry.copy() with patch('time.time', Mock(side_effect=[0, 10, 20, 30, 40])): self.assertRaises(InvalidAuthToken, retry, self.client.deleteprefix, 'foo', retry=retry) + with patch('time.time', Mock(side_effect=[0, 10])): + self.assertRaises(InvalidAuthToken, self.client.deleteprefix, 'foo') self.client.username = None - self.client._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED + self.client._reauthenticate = False retry = self.etcd3._retry.copy() self.assertRaises(InvalidAuthToken, retry, self.client.deleteprefix, 'foo', retry=retry) mock_urlopen.return_value.content = '{"code":3,"error":"etcdserver: revision of auth store is old"}' - self.client._reauthenticate_reason = ReAuthenticateMode.NOT_REQUIRED + self.client._reauthenticate = False self.assertRaises(AuthOldRevision, retry, self.client.deleteprefix, 'foo', retry=retry) def test__handle_server_response(self): @@ -271,8 +273,8 @@ class TestEtcd3(BaseTestEtcd3): def test_attempt_to_acquire_leader(self): self.assertFalse(self.etcd3.attempt_to_acquire_leader()) - with patch('time.time', Mock(side_effect=[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 100, 200])): - self.assertRaises(Etcd3Error, self.etcd3.attempt_to_acquire_leader) + with patch('time.time', Mock(side_effect=[0, 0, 0, 0, 0, 100, 200])): + self.assertFalse(self.etcd3.attempt_to_acquire_leader()) with patch('time.time', Mock(side_effect=[0, 100, 200, 300, 400])): self.assertRaises(Etcd3Error, self.etcd3.attempt_to_acquire_leader) with patch.object(PatroniEtcd3Client, 'put', Mock(return_value=False)):