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

Merge remote-tracking branch 'origin/master' into feature/failover-switchover-definition

Browse Source
This commit is contained in:
Polina Bungina
2023-07-31 15:23:47 +02:00
parent 3684a16b41 94bfea1a81
commit 3e96e89e8b
31 changed files with 145 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/environment.py
+1 -1
View File
@@ -59,7 +59,7 @@ class AbstractController(abc.ABC):
break
time.sleep(1)
else:
assert False,\
assert False, \
"{0} instance is not available for queries after {1} seconds".format(self._name, max_wait_limit)
def stop(self, kill=False, timeout=15, _=False):
features/steps/basic_replication.py
+4 -4
View File
@@ -21,7 +21,7 @@ def start_duplicate_patroni(context, name, port):
context.pctl.start('dup-' + name, custom_config=config)
assert False, "Process was expected to fail"
except AssertionError as e:
assert 'is not running after being started' in str(e),\
assert 'is not running after being started' in str(e), \
"No error was raised by duplicate start of {0} ".format(name)
@@ -88,14 +88,14 @@ def table_is_present_on(context, table_name, pg_name, max_replication_delay):
break
sleep(1)
else:
assert False,\
assert False, \
"Table {0} is not present on {1} after {2} seconds".format(table_name, pg_name, max_replication_delay)
@then('{pg_name:w} role is the {pg_role:w} after {max_promotion_timeout:d} seconds')
def check_role(context, pg_name, pg_role, max_promotion_timeout):
max_promotion_timeout *= context.timeout_multiplier
assert context.pctl.check_role_has_changed_to(pg_name, pg_role, timeout=int(max_promotion_timeout)),\
assert context.pctl.check_role_has_changed_to(pg_name, pg_role, timeout=int(max_promotion_timeout)), \
"{0} role didn't change to {1} after {2} seconds".format(pg_name, pg_role, max_promotion_timeout)
@@ -111,5 +111,5 @@ def replication_works(context, primary, replica, time_limit):
@then('there is a "{message}" {level:w} in the {node} patroni log')
def check_patroni_log(context, message, level, node):
messsages_of_level = context.pctl.read_patroni_log(node, level)
assert any(message in line for line in messsages_of_level),\
assert any(message in line for line in messsages_of_level), \
"There was no {0} {1} in the {2} patroni log".format(message, level, node)
features/steps/citus.py
+1 -1
View File
@@ -125,5 +125,5 @@ def check_transaction(context, name):
@step("a transaction finishes in {timeout:d} seconds")
def check_transaction_timeout(context, timeout):
assert (datetime.now(tzutc) - context.xact_start).seconds > timeout,\
assert (datetime.now(tzutc) - context.xact_start).seconds > timeout, \
"a transaction finished earlier than in {0} seconds".format(timeout)
features/steps/patroni_api.py
+2 -2
View File
@@ -98,7 +98,7 @@ def do_run(context, cmd):
@then('I receive a response {component:w} {data}')
def check_response(context, component, data):
if component == 'code':
assert context.status_code == int(data),\
assert context.status_code == int(data), \
"status code {0} != {1}, response: {2}".format(context.status_code, data, context.response)
elif component == 'returncode':
assert context.status_code == int(data), "return code {0} != {1}, {2}".format(context.status_code,
@@ -158,7 +158,7 @@ def check_http_response(context, url, value, timeout, negate=False):
break
time.sleep(1)
else:
assert False,\
assert False, \
"Value {0} is {1} present in response after {2} seconds".format(value, "not" if not negate else "", timeout)
patroni/config.py
+13 -3
View File
@@ -329,9 +329,19 @@ class Config(object):
@staticmethod
def _process_postgresql_parameters(parameters: Dict[str, Any], is_local: bool = False) -> Dict[str, Any]:
return {name: value for name, value in (parameters or {}).items()
if name not in ConfigHandler.CMDLINE_OPTIONS
or not is_local and ConfigHandler.CMDLINE_OPTIONS[name][1](value)}
pg_params: Dict[str, Any] = {}
for name, value in (parameters or {}).items():
if name not in ConfigHandler.CMDLINE_OPTIONS:
pg_params[name] = value
elif not is_local:
if ConfigHandler.CMDLINE_OPTIONS[name][1](value):
pg_params[name] = value
else:
logging.warning("postgresql parameter %s=%s failed validation, defaulting to %s",
name, value, ConfigHandler.CMDLINE_OPTIONS[name][0])
return pg_params
def _safe_copy_dynamic_configuration(self, dynamic_configuration: Dict[str, Any]) -> Dict[str, Any]:
config = deepcopy(self.__DEFAULT_CONFIG)
patroni/ctl.py
+1 -1
View File
@@ -1552,7 +1552,7 @@ def output_members(obj: Dict[str, Any], cluster: Cluster, name: str,
logging.debug(member)
lag = member.get('lag', '')
member.update(c=name, member=member['name'], group=g,
member.update(cluster=name, member=member['name'], group=g,
host=member.get('host', ''), tl=member.get('timeline', ''),
role=member['role'].replace('_', ' ').title(),
lag_in_mb=round(lag / 1024 / 1024) if isinstance(lag, int) else lag,
patroni/dcs/__init__.py
+3 -1
View File
@@ -1033,7 +1033,9 @@ class AbstractDCS(abc.ABC):
raise
self._last_seen = int(time.time())
self._last_status = {self._OPTIME: cluster.last_lsn, 'slots': cluster.slots}
self._last_status = {self._OPTIME: cluster.last_lsn}
if cluster.slots:
self._last_status['slots'] = cluster.slots
self._last_failsafe = cluster.failsafe
with self._cluster_thread_lock:
patroni/dcs/consul.py
+1 -1
View File
@@ -15,7 +15,7 @@ from urllib3.exceptions import HTTPError
from urllib.parse import urlencode, urlparse, quote
from typing import Any, Callable, Dict, List, Mapping, NamedTuple, Optional, Union, Tuple, TYPE_CHECKING
from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, \
TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
from ..exceptions import DCSError
from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT
patroni/dcs/etcd.py
+1 -1
View File
@@ -21,7 +21,7 @@ from urllib.parse import urlparse
from urllib3 import Timeout
from urllib3.exceptions import HTTPError, ReadTimeoutError, ProtocolError
from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, \
TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
from ..exceptions import DCSError
from ..request import get as requests_get
patroni/dcs/etcd3.py
+11 -1
View File
@@ -15,7 +15,7 @@ from urllib3.exceptions import ReadTimeoutError, ProtocolError
from threading import Condition, Lock, Thread
from typing import Any, Callable, Collection, Dict, Iterator, List, Optional, Tuple, Type, TYPE_CHECKING, Union
from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState,\
from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState, \
TimelineHistory, catch_return_false_exception, citus_group_re
from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors, DnsCachingResolver, Retry
from ..exceptions import DCSError, PatroniException
@@ -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):
patroni/dcs/kubernetes.py
+2 -2
View File
@@ -19,10 +19,10 @@ from urllib3.exceptions import HTTPError
from threading import Condition, Lock, Thread
from typing import Any, Callable, Collection, Dict, List, Optional, Tuple, Type, Union, TYPE_CHECKING
from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, \
TimelineHistory, CITUS_COORDINATOR_GROUP_ID, citus_group_re
from ..exceptions import DCSError
from ..utils import deep_compare, iter_response_objects, keepalive_socket_options,\
from ..utils import deep_compare, iter_response_objects, keepalive_socket_options, \
Retry, RetryFailedError, tzutc, uri, USER_AGENT
if TYPE_CHECKING: # pragma: no cover
from ..config import Config
patroni/ha.py
+22 -9
View File
@@ -497,6 +497,7 @@ class Ha(object):
role = 'replica'
if self.has_lock() and not self.is_standby_cluster():
self._rewind.reset_state() # we want to later trigger CHECKPOINT after promote
msg = "starting as readonly because i had the session lock"
node_to_follow = None
else:
@@ -769,18 +770,14 @@ class Ha(object):
self.state_handler.sync_handler.set_synchronous_standby_names(
CaseInsensitiveSet('*') if self.global_config.is_synchronous_mode_strict else CaseInsensitiveSet())
if self.state_handler.role not in ('master', 'promoted', 'primary'):
def on_success():
self._rewind.reset_state()
logger.info("cleared rewind state after becoming the leader")
def before_promote():
self.notify_citus_coordinator('before_promote')
with self._async_response:
self._async_response.reset()
self._async_executor.try_run_async('promote', self.state_handler.promote,
args=(self.dcs.loop_wait, self._async_response,
before_promote, on_success))
args=(self.dcs.loop_wait, self._async_response, before_promote))
return promote_message
def fetch_node_status(self, member: Member) -> _MemberStatus:
@@ -999,9 +996,22 @@ class Ha(object):
return ret
if self.state_handler.is_leader():
# in pause leader is the healthiest only when no initialize or sysid matches with initialize!
return not self.is_paused() or not self.cluster.initialize\
or self.state_handler.sysid == self.cluster.initialize
if self.is_paused():
# in pause leader is the healthiest only when no initialize or sysid matches with initialize!
return not self.cluster.initialize or self.state_handler.sysid == self.cluster.initialize
# We want to protect from the following scenario:
# 1. node1 is stressed so much that heart-beat isn't running regularly and the leader lock expires.
# 2. node2 promotes, gets heavy load and the situation described in 1 repeats.
# 3. Patroni on node1 comes back, notices that Postgres is running as primary but there is
# no leader key and "happily" acquires the leader lock.
# That is, node1 discarded promotion of node2. To avoid it we want to detect timeline change.
my_timeline = self.state_handler.get_primary_timeline()
if my_timeline < self.cluster.timeline:
logger.warning('My timeline %s is behind last known cluster timeline %s',
my_timeline, self.cluster.timeline)
return False
return True
if self.is_paused():
return False
@@ -1620,6 +1630,9 @@ class Ha(object):
else:
if self._was_paused:
self.state_handler.schedule_sanity_checks_after_pause()
# during pause people could manually do something with Postgres, therefore we want
# to double check rewind conditions on replicas and maybe run CHECKPOINT on the primary
self._rewind.reset_state()
self._was_paused = False
if not self.cluster.has_member(self.state_handler.name):
patroni/postgresql/__init__.py
+2 -4
View File
@@ -1125,8 +1125,8 @@ class Postgresql(object):
except Exception as e:
logger.error('Exception when calling `%s`: %r', cmd, e)
def promote(self, wait_seconds: int, task: CriticalTask, before_promote: Optional[Callable[..., Any]] = None,
on_success: Optional[Callable[..., Any]] = None) -> Optional[bool]:
def promote(self, wait_seconds: int, task: CriticalTask,
before_promote: Optional[Callable[..., Any]] = None) -> Optional[bool]:
if self.role in ('promoted', 'master', 'primary'):
return True
@@ -1152,8 +1152,6 @@ class Postgresql(object):
ret = self.pg_ctl('promote', '-W')
if ret:
self.set_role('promoted')
if on_success is not None:
on_success()
self.call_nowait(CallbackAction.ON_ROLE_CHANGE)
ret = self._wait_promote(wait_seconds)
return ret
patroni/postgresql/config.py
+8 -8
View File
@@ -15,7 +15,7 @@ from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name
from ..exceptions import PatroniFatalException
from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, validate_directory, is_subpath
from ..validator import IntValidator
from ..validator import IntValidator, EnumValidator
if TYPE_CHECKING: # pragma: no cover
from . import Postgresql
@@ -258,14 +258,14 @@ def _false_validator(value: Any) -> bool:
return False
def _wal_level_validator(value: Any) -> bool:
return str(value).lower() in ('hot_standby', 'replica', 'logical')
def _bool_validator(value: Any) -> bool:
return parse_bool(value) is not None
def _bool_is_true_validator(value: Any) -> bool:
return parse_bool(value) is True
class ConfigHandler(object):
# List of parameters which must be always passed to postmaster as command line options
@@ -286,8 +286,8 @@ class ConfigHandler(object):
'listen_addresses': (None, _false_validator, 90100),
'port': (None, _false_validator, 90100),
'cluster_name': (None, _false_validator, 90500),
'wal_level': ('hot_standby', _wal_level_validator, 90100),
'hot_standby': ('on', _false_validator, 90100),
'wal_level': ('hot_standby', EnumValidator(('hot_standby', 'replica', 'logical')), 90100),
'hot_standby': ('on', _bool_is_true_validator, 90100),
'max_connections': (100, IntValidator(min=25), 90100),
'max_wal_senders': (10, IntValidator(min=3), 90100),
'wal_keep_segments': (8, IntValidator(min=1), 90100),
@@ -297,7 +297,7 @@ class ConfigHandler(object):
'track_commit_timestamp': ('off', _bool_validator, 90500),
'max_replication_slots': (10, IntValidator(min=4), 90400),
'max_worker_processes': (8, IntValidator(min=2), 90400),
'wal_log_hints': ('on', _false_validator, 90400)
'wal_log_hints': ('on', _bool_is_true_validator, 90400)
})
_RECOVERY_PARAMETERS = CaseInsensitiveSet(recovery_parameters.keys())
patroni/postgresql/rewind.py
+1 -1
View File
@@ -280,7 +280,7 @@ class Rewind(object):
"""After promote issue a CHECKPOINT from a new thread and asynchronously check the result.
In case if CHECKPOINT failed, just check that timeline in pg_control was updated."""
if self._state == REWIND_STATUS.INITIAL and self._postgresql.is_leader():
if self._state != REWIND_STATUS.CHECKPOINT and self._postgresql.is_leader():
with self._checkpoint_task_lock:
if self._checkpoint_task:
with self._checkpoint_task:
patroni/validator.py
+3 -3
View File
@@ -785,7 +785,7 @@ class IntValidator(object):
self.base_unit = base_unit
self.raise_assert = raise_assert
def __call__(self, value: Union[int, str]) -> bool:
def __call__(self, value: Any) -> bool:
"""Check if *value* is a valid integer and within the expected range.
.. note::
@@ -821,7 +821,7 @@ class EnumValidator(object):
self.allowed_values = set(allowed_values) if case_sensitive else CaseInsensitiveSet(allowed_values)
self.raise_assert = raise_assert
def __call__(self, value: str) -> bool:
def __call__(self, value: Any) -> bool:
"""Check if provided *value* could be found within *allowed_values*.
.. note::
@@ -829,7 +829,7 @@ class EnumValidator(object):
:param value: value to be checked.
:returns: ``True`` if *value* could be found within *allowed_values*.
"""
ret = value in self.allowed_values
ret = isinstance(value, str) and value in self.allowed_values
if self.raise_assert:
assert_(ret)
tests/test_api.py
+1 -1
View File
@@ -226,7 +226,7 @@ class TestRestApiHandler(unittest.TestCase):
self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /primary'))
with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, None, None, '')])):
self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni'))
with patch.object(GlobalConfig, 'is_standby_cluster', Mock(return_value=True)),\
with patch.object(GlobalConfig, 'is_standby_cluster', Mock(return_value=True)), \
patch.object(GlobalConfig, 'is_paused', Mock(return_value=True)):
MockRestApiServer(RestApiHandler, 'GET /standby_leader')
tests/test_bootstrap.py
+9 -9
View File
@@ -155,9 +155,9 @@ class TestBootstrap(BaseTestPostgresql):
config = {'users': {'replicator': {'password': 'rep-pass', 'options': ['replication']}}}
with patch.object(Postgresql, 'is_running', Mock(return_value=False)),\
patch.object(Postgresql, 'get_major_version', Mock(return_value=140000)),\
patch('multiprocessing.Process', Mock(side_effect=Exception)),\
with patch.object(Postgresql, 'is_running', Mock(return_value=False)), \
patch.object(Postgresql, 'get_major_version', Mock(return_value=140000)), \
patch('multiprocessing.Process', Mock(side_effect=Exception)), \
patch('multiprocessing.get_context', Mock(side_effect=Exception), create=True):
self.assertRaises(Exception, self.b.bootstrap, config)
with open(os.path.join(self.p.data_dir, 'pg_hba.conf')) as f:
@@ -185,12 +185,12 @@ class TestBootstrap(BaseTestPostgresql):
self.assertFalse(self.b.bootstrap(config))
mock_cancellable_subprocess_call.return_value = 0
with patch('multiprocessing.Process', Mock(side_effect=Exception("42"))),\
patch('multiprocessing.get_context', Mock(side_effect=Exception("42")), create=True),\
patch('os.path.isfile', Mock(return_value=True)),\
patch('os.unlink', Mock()),\
patch.object(ConfigHandler, 'save_configuration_files', Mock()),\
patch.object(ConfigHandler, 'restore_configuration_files', Mock()),\
with patch('multiprocessing.Process', Mock(side_effect=Exception("42"))), \
patch('multiprocessing.get_context', Mock(side_effect=Exception("42")), create=True), \
patch('os.path.isfile', Mock(return_value=True)), \
patch('os.unlink', Mock()), \
patch.object(ConfigHandler, 'save_configuration_files', Mock()), \
patch.object(ConfigHandler, 'restore_configuration_files', Mock()), \
patch.object(ConfigHandler, 'write_recovery_conf', Mock()):
with self.assertRaises(Exception) as e:
self.b.bootstrap(config)
tests/test_citus.py
+2 -2
View File
@@ -52,7 +52,7 @@ class TestCitus(BaseTestPostgresql):
'leader': 'leader', 'timeout': 30, 'cooldown': 10})
def test_add_task(self):
with patch('patroni.postgresql.citus.logger.error') as mock_logger,\
with patch('patroni.postgresql.citus.logger.error') as mock_logger, \
patch('patroni.postgresql.citus.urlparse', Mock(side_effect=Exception)):
self.c.add_task('', 1, None)
mock_logger.assert_called_once()
@@ -107,7 +107,7 @@ class TestCitus(BaseTestPostgresql):
self.c.process_tasks()
self.c.add_task('after_promote', 0, 'postgres://host3:5432/postgres')
with patch('patroni.postgresql.citus.logger.error') as mock_logger,\
with patch('patroni.postgresql.citus.logger.error') as mock_logger, \
patch.object(CitusHandler, 'query', Mock(side_effect=Exception)):
self.c.process_tasks()
mock_logger.assert_called_once()
tests/test_config.py
+2 -1
View File
@@ -21,7 +21,8 @@ class TestConfig(unittest.TestCase):
with patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception)):
self.assertFalse(self.config.set_dynamic_configuration({'foo': 'bar'}))
self.assertTrue(self.config.set_dynamic_configuration({'standby_cluster': {}, 'postgresql': {
'parameters': {'cluster_name': 1, 'wal_keep_size': 1, 'track_commit_timestamp': 1, 'wal_level': 1}}}))
'parameters': {'cluster_name': 1, 'hot_standby': 1, 'wal_keep_size': 1,
'track_commit_timestamp': 1, 'wal_level': 1}}}))
def test_reload_local_configuration(self):
os.environ.update({
tests/test_ctl.py
+5 -1
View File
@@ -83,9 +83,13 @@ class TestCtl(unittest.TestCase):
scheduled_at = datetime.now(tzutc) + timedelta(seconds=600)
cluster = get_cluster_initialized_with_leader(Failover(1, 'foo', 'bar', scheduled_at))
del cluster.members[1].data['conn_url']
for fmt in ('pretty', 'json', 'yaml', 'tsv', 'topology'):
for fmt in ('pretty', 'json', 'yaml', 'topology'):
self.assertIsNone(output_members({}, cluster, name='abc', fmt=fmt))
with patch('click.echo') as mock_echo:
self.assertIsNone(output_members({}, cluster, name='abc', fmt='tsv'))
self.assertEqual(mock_echo.call_args[0][0], 'abc\tother\t\tReplica\trunning\t\tunknown')
@patch('patroni.ctl.get_dcs')
@patch.object(PoolManager, 'request', Mock(return_value=MockResponse()))
def test_switchover(self, mock_get_dcs):
tests/test_etcd.py
+2 -2
View File
@@ -172,12 +172,12 @@ class TestClient(unittest.TestCase):
self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'})
self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'})
with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\
with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])), \
patch.object(EtcdClient, '_load_machines_cache', Mock(side_effect=Exception)):
self.client.http.request = Mock(side_effect=socket.error)
self.assertRaises(etcd.EtcdException, rtry, self.client.api_execute, '/', 'GET', params={'retry': rtry})
with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])),\
with patch.object(EtcdClient, '_calculate_timeouts', Mock(side_effect=[(1, 1, 0), (1, 1, 0), (0, 1, 0)])), \
patch.object(EtcdClient, '_load_machines_cache', Mock(return_value=True)):
self.assertRaises(etcd.EtcdException, rtry, self.client.api_execute, '/', 'GET', params={'retry': rtry})
tests/test_etcd3.py
+11 -4
View File
@@ -5,8 +5,9 @@ import urllib3
from mock import Mock, PropertyMock, patch
from patroni.dcs.etcd import DnsCachingResolver
from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3Client, Etcd3Error, Etcd3ClientError, RetryFailedError,\
InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode, Etcd3
from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Client, \
Etcd3Error, Etcd3ClientError, RetryFailedError, InvalidAuthToken, Unavailable, \
Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode
from threading import Thread
from . import SleepException, MockResponse
@@ -126,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:
@@ -241,7 +248,7 @@ class TestEtcd3(BaseTestEtcd3):
self.etcd3.update_leader(leader, '123', failsafe={'foo': 'bar'})
self.etcd3._last_lease_refresh = 0
self.etcd3.update_leader(leader, '124')
with patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=True)),\
with patch.object(PatroniEtcd3Client, 'lease_keepalive', Mock(return_value=True)), \
patch('time.time', Mock(side_effect=[0, 100, 200, 300])):
self.assertRaises(Etcd3Error, self.etcd3.update_leader, leader, '126')
self.etcd3._lease = leader.session
tests/test_ha.py
+11 -5
View File
@@ -307,7 +307,7 @@ class TestHa(PostgresInit):
self.p.is_running = false
self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID}
self.assertEqual(self.ha.run_cycle(), 'doing crash recovery in a single user mode')
with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)),\
with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)), \
patch.object(Ha, 'check_timeline', Mock(return_value=False)):
self.ha._async_executor.schedule('doing crash recovery in a single user mode')
self.ha.state_handler.cancellable._process = Mock()
@@ -340,7 +340,7 @@ class TestHa(PostgresInit):
self.ha._rewind.check_leader_is_not_in_recovery = true
with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True)):
self.assertEqual(self.ha.run_cycle(), 'running pg_rewind from leader')
with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=False)),\
with patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=False)), \
patch.object(Ha, 'is_synchronous_mode', Mock(return_value=True)):
self.p.follow = true
self.assertEqual(self.ha.run_cycle(), 'starting as a secondary')
@@ -375,6 +375,12 @@ class TestHa(PostgresInit):
def test_acquire_lock_as_primary(self):
self.assertEqual(self.ha.run_cycle(), 'acquired session lock as a leader')
def test_leader_race_stale_primary(self):
with patch.object(Postgresql, 'get_primary_timeline', Mock(return_value=1)), \
patch('patroni.ha.logger.warning') as mock_logger:
self.assertEqual(self.ha.run_cycle(), 'demoting self because i am not the healthiest node')
self.assertEqual(mock_logger.call_args[0][0], 'My timeline %s is behind last known cluster timeline %s')
def test_promoted_by_acquiring_lock(self):
self.ha.is_healthiest_node = true
self.p.is_leader = false
@@ -608,7 +614,7 @@ class TestHa(PostgresInit):
self.e.initialize = true
self.ha.bootstrap()
self.p.is_leader = true
with patch.object(Watchdog, 'activate', Mock(return_value=False)),\
with patch.object(Watchdog, 'activate', Mock(return_value=False)), \
patch('patroni.ha.logger.error') as mock_logger:
self.assertEqual(self.ha.post_bootstrap(), 'running post_bootstrap')
self.assertRaises(PatroniFatalException, self.ha.post_bootstrap)
@@ -669,9 +675,9 @@ class TestHa(PostgresInit):
self.ha.update_lock = false
self.p.set_role('primary')
with patch('patroni.async_executor.CriticalTask.cancel', Mock(return_value=False)),\
with patch('patroni.async_executor.CriticalTask.cancel', Mock(return_value=False)), \
patch('patroni.async_executor.CriticalTask.result',
PropertyMock(return_value=PostmasterProcess(os.getpid())), create=True),\
PropertyMock(return_value=PostmasterProcess(os.getpid())), create=True), \
patch('patroni.postgresql.Postgresql.terminate_starting_postmaster') as mock_terminate:
self.assertEqual(self.ha.run_cycle(), 'lost leader lock during restart')
mock_terminate.assert_called()
tests/test_kubernetes.py
+10 -10
View File
@@ -8,8 +8,8 @@ import unittest
import urllib3
from mock import Mock, PropertyMock, mock_open, patch
from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed,\
K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException,\
from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed, \
K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException, \
Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME
from threading import Thread
from . import MockResponse, SleepException
@@ -86,8 +86,8 @@ class TestK8sConfig(unittest.TestCase):
with patch('os.environ', env):
self.assertRaises(k8s_config.ConfigException, k8s_config.load_incluster_config)
with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}),\
patch('os.path.isfile', Mock(side_effect=[False, True, True, False, True, True, True, True])),\
with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}), \
patch('os.path.isfile', Mock(side_effect=[False, True, True, False, True, True, True, True])), \
patch('builtins.open', Mock(side_effect=[
mock_open()(), mock_open(read_data='a')(), mock_open(read_data='a')(),
mock_open()(), mock_open(read_data='a')(), mock_open(read_data='a')()])):
@@ -98,8 +98,8 @@ class TestK8sConfig(unittest.TestCase):
self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer a')
def test_refresh_token(self):
with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}),\
patch('os.path.isfile', Mock(side_effect=[True, True, False, True, True, True])),\
with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}), \
patch('os.path.isfile', Mock(side_effect=[True, True, False, True, True, True])), \
patch('builtins.open', Mock(side_effect=[
mock_open(read_data='cert')(), mock_open(read_data='a')(),
mock_open()(), mock_open(read_data='b')(), mock_open(read_data='c')()])):
@@ -138,10 +138,10 @@ class TestK8sConfig(unittest.TestCase):
config["users"][0]["user"]["client-key-data"] = base64.b64encode(b'foobar').decode('utf-8')
config["clusters"][0]["cluster"]["certificate-authority-data"] = base64.b64encode(b'foobar').decode('utf-8')
with patch('builtins.open', mock_open(read_data=json.dumps(config))),\
patch('os.write', Mock()), patch('os.close', Mock()),\
patch('os.remove') as mock_remove,\
patch('atexit.register') as mock_atexit,\
with patch('builtins.open', mock_open(read_data=json.dumps(config))), \
patch('os.write', Mock()), patch('os.close', Mock()), \
patch('os.remove') as mock_remove, \
patch('atexit.register') as mock_atexit, \
patch('tempfile.mkstemp') as mock_mkstemp:
mock_mkstemp.side_effect = [(3, '1.tmp'), (4, '2.tmp')]
k8s_config.load_kube_config()
tests/test_log.py
+1 -1
View File
@@ -43,7 +43,7 @@ class TestPatroniLogger(unittest.TestCase):
_LOG.exception('test')
logger.start()
with patch.object(logging.Handler, 'format', Mock(side_effect=Exception)),\
with patch.object(logging.Handler, 'format', Mock(side_effect=Exception)), \
patch('_pytest.logging.LogCaptureHandler.emit', Mock()):
logging.error('test')
tests/test_postgresql.py
+3 -3
View File
@@ -333,7 +333,7 @@ class TestPostgresql(BaseTestPostgresql):
mock_read_auto = mock_open(read_data=read_data)
mock_read_auto.return_value.__iter__ = lambda o: iter(o.readline, '')
with patch('builtins.open', Mock(side_effect=[mock_open()(), mock_read_auto(), IOError])),\
with patch('builtins.open', Mock(side_effect=[mock_open()(), mock_read_auto(), IOError])), \
patch('os.chmod', Mock()):
self.p.config.write_postgresql_conf()
@@ -496,8 +496,8 @@ class TestPostgresql(BaseTestPostgresql):
self.p.remove_data_directory()
with patch('os.path.isfile', Mock(return_value=True)):
self.p.remove_data_directory()
with patch('os.path.islink', Mock(side_effect=[False, False, True, True])),\
patch('os.listdir', Mock(return_value=['12345'])),\
with patch('os.path.islink', Mock(side_effect=[False, False, True, True])), \
patch('os.listdir', Mock(return_value=['12345'])), \
patch('os.path.realpath', Mock(side_effect=['../foo', '../foo_tsp'])):
self.p.remove_data_directory()
tests/test_raft.py
+1 -1
View File
@@ -4,7 +4,7 @@ import tempfile
import time
from mock import Mock, PropertyMock, patch
from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL,\
from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL, \
Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport
from pysyncobj import SyncObjConf, FAIL_REASON
tests/test_rewind.py
+3 -3
View File
@@ -65,14 +65,14 @@ class TestRewind(BaseTestPostgresql):
def test_pg_rewind(self):
r = {'user': '', 'host': '', 'port': '', 'database': '', 'password': ''}
with patch.object(Postgresql, 'major_version', PropertyMock(return_value=150000)),\
with patch.object(Postgresql, 'major_version', PropertyMock(return_value=150000)), \
patch.object(CancellableSubprocess, 'call', Mock(return_value=None)):
with patch('subprocess.check_output', Mock(return_value=b'boo')):
self.assertFalse(self.r.pg_rewind(r))
with patch('subprocess.check_output', Mock(side_effect=Exception)):
self.assertFalse(self.r.pg_rewind(r))
with patch.object(Postgresql, 'major_version', PropertyMock(return_value=120000)),\
with patch.object(Postgresql, 'major_version', PropertyMock(return_value=120000)), \
patch('subprocess.check_output', Mock(return_value=b'foo %f %p %r %% % %')):
with patch.object(CancellableSubprocess, 'call', mock_cancellable_call):
self.assertFalse(self.r.pg_rewind(r))
@@ -91,7 +91,7 @@ class TestRewind(BaseTestPostgresql):
'Latest checkpoint location': '0/'})):
self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
with patch.object(Postgresql, 'is_running', Mock(return_value=True)),\
with patch.object(Postgresql, 'is_running', Mock(return_value=True)), \
patch.object(MockCursor, 'fetchone',
Mock(side_effect=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None, None, None), Exception])):
self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
tests/test_slots.py
+7 -7
View File
@@ -43,12 +43,12 @@ class TestSlotsHandler(BaseTestPostgresql):
with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg.OperationalError)):
self.s.sync_replication_slots(cluster, False)
self.p.set_role('standby_leader')
with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))),\
with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))), \
patch('patroni.postgresql.slots.logger.debug') as mock_debug:
self.s.sync_replication_slots(cluster, False)
mock_debug.assert_called_once()
self.p.set_role('replica')
with patch.object(Postgresql, 'is_leader', Mock(return_value=False)),\
with patch.object(Postgresql, 'is_leader', Mock(return_value=False)), \
patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop:
self.s.sync_replication_slots(cluster, False, paused=True)
mock_drop.assert_not_called()
@@ -96,8 +96,8 @@ class TestSlotsHandler(BaseTestPostgresql):
with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)):
self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
self.s._schedule_load_slots = False
with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)),\
patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))),\
with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \
patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))), \
patch.object(psycopg.OperationalError, 'diag') as mock_diag:
type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
self.assertEqual(self.s.sync_replication_slots(self.cluster, False), ['ls'])
@@ -119,10 +119,10 @@ class TestSlotsHandler(BaseTestPostgresql):
@patch.object(Postgresql, 'is_leader', Mock(return_value=False))
def test_check_logical_slots_readiness(self):
self.s.copy_logical_slots(self.cluster, ['ls'])
with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \
patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)):
self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None))
with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))), \
patch.object(MockCursor, 'fetchone', Mock(return_value=(False,))):
self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None))
with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('ls', 100)]))):
@@ -144,7 +144,7 @@ class TestSlotsHandler(BaseTestPostgresql):
self.assertRaises(OSError, fsync_dir, 'foo')
def test_slots_advance_thread(self):
with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)),\
with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \
patch.object(psycopg.OperationalError, 'diag') as mock_diag:
type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
self.s.schedule_advance_slots({'foo': {'bar': 100}})
tests/test_zookeeper.py
+1 -1
View File
@@ -7,7 +7,7 @@ from kazoo.handlers.threading import SequentialThreadingHandler
from kazoo.protocol.states import KeeperState, ZnodeStat
from kazoo.retry import RetryFailedError
from mock import Mock, PropertyMock, patch
from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient,\
from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient, \
PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError