diff --git a/features/environment.py b/features/environment.py index ec503bb5..e3c21252 100644 --- a/features/environment.py +++ b/features/environment.py @@ -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): diff --git a/features/steps/basic_replication.py b/features/steps/basic_replication.py index 6499a03f..6718cb01 100644 --- a/features/steps/basic_replication.py +++ b/features/steps/basic_replication.py @@ -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) diff --git a/features/steps/citus.py b/features/steps/citus.py index 1af70a30..644219c7 100644 --- a/features/steps/citus.py +++ b/features/steps/citus.py @@ -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) diff --git a/features/steps/patroni_api.py b/features/steps/patroni_api.py index a745b192..2c76d32d 100644 --- a/features/steps/patroni_api.py +++ b/features/steps/patroni_api.py @@ -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) diff --git a/patroni/config.py b/patroni/config.py index 66ffd891..facf16fd 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -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) diff --git a/patroni/ctl.py b/patroni/ctl.py index 5d978a87..265e12e1 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -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, diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index f07ce6cd..29a4d766 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -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: diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 73755a41..7b310d15 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -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 diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 94e546ac..447c7dab 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -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 diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index 32d00359..e28da844 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -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): diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index bda4bdc8..96097338 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -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 diff --git a/patroni/ha.py b/patroni/ha.py index 6ed4c168..4d4de6be 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -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): diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 632cbee7..04fab0a2 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -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 diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 9461714f..3a61a31d 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -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()) diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index ff6fc751..270d629c 100644 --- a/patroni/postgresql/rewind.py +++ b/patroni/postgresql/rewind.py @@ -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: diff --git a/patroni/validator.py b/patroni/validator.py index b68f9654..a8ada1f5 100644 --- a/patroni/validator.py +++ b/patroni/validator.py @@ -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) diff --git a/tests/test_api.py b/tests/test_api.py index 67db9ba7..bfad88df 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -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') diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py index 4c7c0fcc..9f98fecb 100644 --- a/tests/test_bootstrap.py +++ b/tests/test_bootstrap.py @@ -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) diff --git a/tests/test_citus.py b/tests/test_citus.py index 40d4df8a..7c2d63bb 100644 --- a/tests/test_citus.py +++ b/tests/test_citus.py @@ -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() diff --git a/tests/test_config.py b/tests/test_config.py index 57a717f5..3f7e4049 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -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({ diff --git a/tests/test_ctl.py b/tests/test_ctl.py index e846f4bf..99555403 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -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): diff --git a/tests/test_etcd.py b/tests/test_etcd.py index b5add201..f9e07144 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -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}) diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index a737f199..2e3ed59d 100644 --- a/tests/test_etcd3.py +++ b/tests/test_etcd3.py @@ -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 diff --git a/tests/test_ha.py b/tests/test_ha.py index 771edf15..06b72fba 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -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() diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index f79539a7..662e22a8 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -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() diff --git a/tests/test_log.py b/tests/test_log.py index 1a383908..ebdb4945 100644 --- a/tests/test_log.py +++ b/tests/test_log.py @@ -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') diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index c759cde0..036e225b 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -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() diff --git a/tests/test_raft.py b/tests/test_raft.py index ab68c2e6..1fe8733c 100644 --- a/tests/test_raft.py +++ b/tests/test_raft.py @@ -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 diff --git a/tests/test_rewind.py b/tests/test_rewind.py index a4fafcff..22181266 100644 --- a/tests/test_rewind.py +++ b/tests/test_rewind.py @@ -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) diff --git a/tests/test_slots.py b/tests/test_slots.py index 3f21998f..6d3f17d3 100644 --- a/tests/test_slots.py +++ b/tests/test_slots.py @@ -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}}) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index baf9ad10..c72fefe9 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -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