gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Retry one time on Etcd3 auth error (#3026)

Browse Source 
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
This commit is contained in:
Alexander Kukushkin
2024-03-07 12:01:35 +01:00
committed by GitHub
parent fd3e3ca472
commit a8cfd46801
5 changed files with 34 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/dcs/consul.py
+6 -3
View File
@@ -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
patroni/dcs/etcd3.py
+17 -28
View File
@@ -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
patroni/utils.py
+2 -3
View File
@@ -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
tests/test_consul.py
+2 -4
View File
@@ -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):
tests/test_etcd3.py
+7 -5
View File
@@ -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)):