From 6f357a4e175d00cc4554495810820aee489a42fe Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Mon, 3 Apr 2023 08:09:29 +0200 Subject: [PATCH] Factor out global configuration into a dedicated class (#2628) It will help to avoid code duplications. --- patroni/api.py | 75 ++++++++++----- patroni/config.py | 164 ++++++++++++++++++++++++++------- patroni/ctl.py | 19 ++-- patroni/dcs/__init__.py | 11 +-- patroni/ha.py | 90 ++++++++---------- patroni/postgresql/__init__.py | 26 ++++-- patroni/postgresql/config.py | 7 +- patroni/utils.py | 27 ++---- tests/test_api.py | 64 +++++++------ tests/test_config.py | 4 +- tests/test_ctl.py | 12 +-- tests/test_etcd.py | 1 - tests/test_ha.py | 27 ++++-- tests/test_postgresql.py | 6 +- 14 files changed, 329 insertions(+), 204 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index 2c84ea8f..c614dcb9 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -15,8 +15,10 @@ from ipaddress import ip_address, ip_network from socketserver import ThreadingMixIn from threading import Thread from urllib.parse import urlparse, parse_qs +from typing import Any, Dict, Optional, Union from . import psycopg +from .dcs import Cluster from .exceptions import PostgresConnectionException, PostgresException from .postgresql.misc import postgres_version_to_int from .utils import deep_compare, enable_keepalive, parse_bool, patch_config, Retry, \ @@ -63,7 +65,8 @@ class RestApiHandler(BaseHTTPRequestHandler): return wrapper - def _write_status_response(self, status_code, response): + def _write_status_response(self, status_code: int, response: Dict[str, Any]) -> None: + """Sends HTTP response with Patroni/Postgres status in JSON format.""" patroni = self.server.patroni tags = patroni.ha.get_effective_tags() if tags: @@ -79,8 +82,6 @@ class RestApiHandler(BaseHTTPRequestHandler): response['scheduled_restart']['schedule'] = (response['scheduled_restart']['schedule']).isoformat() if not patroni.ha.watchdog.is_healthy: response['watchdog_failed'] = True - if patroni.ha.is_paused(): - response['pause'] = True qsize = patroni.logger.queue_size if qsize > patroni.logger.NORMAL_LOG_QUEUE_SIZE: response['logger_queue_size'] = qsize @@ -89,14 +90,21 @@ class RestApiHandler(BaseHTTPRequestHandler): response['logger_records_lost'] = lost self._write_json_response(status_code, response) - def do_GET(self, write_status_code_only=False): - """Default method for processing all GET requests which can not be routed to other methods""" + def do_GET(self, write_status_code_only: Optional[bool] = False) -> None: + """Default method for processing all GET requests which can not be routed to other methods. + + Is used for handling all health-checks requests. E.g. "GET /(primary|replica|sync|async|etc...)" + :param write_status_code_only: indicates that instead of normal HTTP response we should + send only HTTP Status Code and close the connection. + It is useful to when health-checks are executed by HAProxy. + """ path = '/primary' if self.path == '/' else self.path response = self.get_postgresql_status() patroni = self.server.patroni cluster = patroni.dcs.cluster + global_config = patroni.config.get_global_config(cluster) leader_optime = cluster and cluster.last_lsn or 0 replayed_location = response.get('xlog', {}).get('replayed_location', 0) @@ -108,13 +116,13 @@ class RestApiHandler(BaseHTTPRequestHandler): replica_status_code = 200 if not patroni.noloadbalance and not is_lagging and \ response.get('role') == 'replica' and response.get('state') == 'running' else 503 - if not cluster and patroni.ha.is_paused(): + if not cluster and response.get('pause'): leader_status_code = 200 if response.get('role') in ('master', 'primary', 'standby_leader') else 503 primary_status_code = 200 if response.get('role') in ('master', 'primary') else 503 standby_leader_status_code = 200 if response.get('role') == 'standby_leader' else 503 elif patroni.ha.is_leader(): leader_status_code = 200 - if patroni.ha.is_standby_cluster(): + if global_config.is_standby_cluster: primary_status_code = replica_status_code = 503 standby_leader_status_code = 200 if response.get('role') in ('replica', 'standby_leader') else 503 else: @@ -209,9 +217,11 @@ class RestApiHandler(BaseHTTPRequestHandler): response = self.get_postgresql_status(True) self._write_status_response(200, response) - def do_GET_cluster(self): + def do_GET_cluster(self) -> None: + """Sends response with JSON representaion of Cluster topology.""" cluster = self.server.patroni.dcs.get_cluster(True) - self._write_json_response(200, cluster_as_json(cluster)) + global_config = self.server.patroni.config.get_global_config(cluster) + self._write_json_response(200, cluster_as_json(cluster, global_config)) def do_GET_history(self): cluster = self.server.patroni.dcs.cluster or self.server.patroni.dcs.get_cluster() @@ -224,7 +234,8 @@ class RestApiHandler(BaseHTTPRequestHandler): else: self.send_error(502) - def do_GET_metrics(self): + def do_GET_metrics(self) -> None: + """Sends response in Prometheus format.""" postgres = self.get_postgresql_status(True) patroni = self.server.patroni epoch = datetime.datetime(1970, 1, 1, tzinfo=tzutc) @@ -325,8 +336,7 @@ class RestApiHandler(BaseHTTPRequestHandler): metrics.append("# HELP patroni_is_paused Value is 1 if auto failover is disabled, 0 otherwise.") metrics.append("# TYPE patroni_is_paused gauge") - metrics.append("patroni_is_paused{0} {1}" - .format(scope_label, int(patroni.ha.is_paused()))) + metrics.append("patroni_is_paused{0} {1}".format(scope_label, int(postgres.get('pause', 0)))) self._write_response(200, '\n'.join(metrics)+'\n', content_type='text/plain') @@ -423,7 +433,8 @@ class RestApiHandler(BaseHTTPRequestHandler): return (status_code, error, scheduled_at) @check_access - def do_POST_restart(self): + def do_POST_restart(self) -> None: + """Is used to restart postgres, mainly by "patronictl restart".""" status_code = 500 data = 'restart failed' request = self._read_json_content(body_is_optional=True) @@ -434,7 +445,7 @@ class RestApiHandler(BaseHTTPRequestHandler): if request: logger.debug("received restart request: {0}".format(request)) - if cluster.is_paused() and 'schedule' in request: + if self.server.patroni.config.get_global_config(cluster).is_paused and 'schedule' in request: self._write_response(status_code, "Can't schedule restart in the paused state") return @@ -542,16 +553,21 @@ class RestApiHandler(BaseHTTPRequestHandler): logger.debug('Exception occurred during polling %s result: %s', action, e) return 503, action.title() + ' status unknown' - def is_failover_possible(self, cluster, leader, candidate, action): + def is_failover_possible(self, cluster: Cluster, leader: str, candidate: str, action: str) -> Union[str, None]: + """Checks whether there are nodes that could take it over after demoting the primary. + + :returns: a string with the error message or `None` if good nodes are found + """ + is_synchronous_mode = self.server.patroni.config.get_global_config(cluster).is_synchronous_mode if leader and (not cluster.leader or cluster.leader.name != leader): return 'leader name does not match' if candidate: - if action == 'switchover' and cluster.is_synchronous_mode() and not cluster.sync.matches(candidate): + if action == 'switchover' and is_synchronous_mode and not cluster.sync.matches(candidate): return 'candidate name does not match with sync_standby' members = [m for m in cluster.members if m.name == candidate] if not members: return 'candidate does not exists' - elif cluster.is_synchronous_mode(): + elif is_synchronous_mode: members = [m for m in cluster.members if cluster.sync.matches(m.name)] if not members: return action + ' is not possible: can not find sync_standby' @@ -565,7 +581,8 @@ class RestApiHandler(BaseHTTPRequestHandler): return action + ' is not possible: no good candidates have been found' @check_access - def do_POST_failover(self, action='failover'): + def do_POST_failover(self, action: Optional[str] = 'failover') -> None: + """Handles manual failovers/switchovers, mainly from "patronictl".""" request = self._read_json_content() (status_code, data) = (400, '') if not request: @@ -575,6 +592,7 @@ class RestApiHandler(BaseHTTPRequestHandler): candidate = request.get('candidate') or request.get('member') scheduled_at = request.get('scheduled_at') cluster = self.server.patroni.dcs.get_cluster() + global_config = self.server.patroni.config.get_global_config(cluster) logger.info("received %s request with leader=%s candidate=%s scheduled_at=%s", action, leader, candidate, scheduled_at) @@ -587,12 +605,12 @@ class RestApiHandler(BaseHTTPRequestHandler): if not data and scheduled_at: if not leader: data = 'Scheduled {0} is possible only from a specific leader'.format(action) - if not data and cluster.is_paused(): + if not data and global_config.is_paused: data = "Can't schedule {0} in the paused state".format(action) if not data: (status_code, data, scheduled_at) = self.parse_schedule(scheduled_at, action) - if not data and cluster.is_paused() and not candidate: + if not data and global_config.is_paused and not candidate: data = action.title() + ' is possible only to a specific candidate in a paused state' if not data and not scheduled_at: @@ -656,10 +674,17 @@ class RestApiHandler(BaseHTTPRequestHandler): retry = Retry(delay=1, retry_exceptions=PostgresConnectionException) return retry(self.server.query, sql, *params) - def get_postgresql_status(self, retry=False): + def get_postgresql_status(self, retry: Optional[bool] = False) -> Dict[str, Any]: + """Builds an object representing a status of "postgres". + + Some of values are collected by executing a query and other are taken from the state stored in memory. + :param retry: whether the query should be retried if failed or give up immediately + :returns: a dict with the status of Postgres/Patroni + """ postgresql = self.server.patroni.postgresql + cluster = self.server.patroni.dcs.cluster + global_config = self.server.patroni.config.get_global_config(cluster) try: - cluster = self.server.patroni.dcs.cluster if postgresql.state not in ('running', 'restarting', 'starting'): raise RetryFailedError('') @@ -686,10 +711,10 @@ class RestApiHandler(BaseHTTPRequestHandler): }) } - if result['role'] == 'replica' and self.server.patroni.ha.is_standby_cluster(): + if result['role'] == 'replica' and global_config.is_standby_cluster: result['role'] = postgresql.role - if result['role'] == 'replica' and cluster and cluster.is_synchronous_mode()\ + if result['role'] == 'replica' and global_config.is_synchronous_mode\ and cluster.sync.matches(postgresql.name): result['sync_standby'] = True @@ -709,6 +734,8 @@ class RestApiHandler(BaseHTTPRequestHandler): state = 'unknown' result = {'state': state, 'role': postgresql.role} + if global_config.is_paused: + result['pause'] = True if not cluster or cluster.is_unlocked(): result['cluster_unlocked'] = True if self.server.patroni.ha.failsafe_is_active(): diff --git a/patroni/config.py b/patroni/config.py index 02379c20..2ec129ad 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -7,11 +7,13 @@ import yaml from collections import defaultdict from copy import deepcopy -from patroni import PATRONI_ENV_PREFIX -from patroni.exceptions import ConfigParseError -from patroni.dcs import ClusterConfig -from patroni.postgresql.config import CaseInsensitiveDict, ConfigHandler -from patroni.utils import deep_compare, parse_bool, parse_int, patch_config +from typing import Any, Dict, Optional, Union + +from . import PATRONI_ENV_PREFIX +from .exceptions import ConfigParseError +from .dcs import ClusterConfig, Cluster +from .postgresql.config import CaseInsensitiveDict, ConfigHandler +from .utils import deep_compare, parse_bool, parse_int, patch_config logger = logging.getLogger(__name__) @@ -35,6 +37,120 @@ def default_validator(conf): raise ConfigParseError("Config is empty.") +class GlobalConfig(object): + + """A class that wrapps global configuration and provides convinient methods to access/check values. + + It is instantiated by calling :func:`Config.global_config` method which picks either a + configuration from provided :class:`Cluster` object (the most up-to-date) or from the + local cache if :class::`ClusterConfig` is not initialized or doesn't have a valid config. + """ + + def __init__(self, config: Dict[str, Any]) -> None: + """Initialize :class:`GlobalConfig` object. + + :param config: current configuration either from + :class:`ClusterConfig` or from :class:`Config.dynamic_configuration` + """ + self.__config = config + + def get(self, name: str) -> Any: + """Gets global configuration value by name. + + :param name: parameter name + :returns: configuration value or `None` if it is missing + """ + return self.__config.get(name) + + def check_mode(self, mode: str) -> bool: + """Checks whether the certain parameter is enabled. + + :param mode: parameter name could be: synchronous_mode, failsafe_mode, pause, check_timeline, and so on + :returns: `True` if *mode* is enabled in the global configuration. + """ + return bool(parse_bool(self.__config.get(mode))) + + @property + def is_paused(self) -> bool: + """:returns: `True` if cluster is in maintenance mode.""" + return self.check_mode('pause') + + @property + def is_synchronous_mode(self) -> bool: + """:returns: `True` if synchronous replication is requested.""" + return self.check_mode('synchronous_mode') + + @property + def is_synchronous_mode_strict(self) -> bool: + """:returns: `True` if at least one synchronous node is required.""" + return self.check_mode('synchronous_mode_strict') + + def get_standby_cluster_config(self) -> Any: + """:returns: "standby_cluster" configuration.""" + return deepcopy(self.get('standby_cluster')) + + @property + def is_standby_cluster(self) -> bool: + """:returns: `True` if global configuration has a valid "standby_cluster" section.""" + config = self.get_standby_cluster_config() + return isinstance(config, dict) and\ + bool(config.get('host') or config.get('port') or config.get('restore_command')) + + def get_int(self, name: str, default: int = 0) -> int: + """Gets current value from the global configuration and trying to return it as int. + + :param name: name of the parameter + :param default: default value if *name* is not in the configuration or invalid + :returns: currently configured value from the global configuration or *default* if it is not set or invalid. + """ + ret = parse_int(self.get(name)) + return default if ret is None else ret + + @property + def synchronous_node_count(self) -> int: + """:returns: currently configured value from the global configuration or 1 if it is not set or invalid.""" + return self.get_int('synchronous_node_count', 0) + + @property + def maximum_lag_on_failover(self) -> int: + """:returns: currently configured value from the global configuration or 1048576 if it is not set or invalid.""" + return self.get_int('maximum_lag_on_failover', 1048576) + + @property + def maximum_lag_on_syncnode(self) -> int: + """:returns: currently configured value from the global configuration or -1 if it is not set or invalid.""" + return self.get_int('maximum_lag_on_syncnode', -1) + + @property + def primary_start_timeout(self) -> int: + """:returns: currently configured value from the global configuration or 300 if it is not set or invalid.""" + default = 300 + return self.get_int('primary_start_timeout', default)\ + if 'primary_start_timeout' in self.__config else self.get_int('master_start_timeout', default) + + @property + def primary_stop_timeout(self) -> int: + """:returns: currently configured value from the global configuration or 300 if it is not set or invalid.""" + default = 0 + return self.get_int('primary_stop_timeout', default)\ + if 'primary_stop_timeout' in self.__config else self.get_int('master_stop_timeout', default) + + +def get_global_config(cluster: Union[Cluster, None], default: Optional[Dict] = None) -> GlobalConfig: + """Instantiates :class:`GlobalConfig` based on the input. + + :param cluster: the currently known cluster state from DCS + :param default: default configuration, which will be used if there is no valid *cluster.config* + :returns: :class:`GlobalConfig` object + """ + # Try to protect from the case when DCS was wiped out + if cluster and cluster.config and cluster.config.modify_index: + config = cluster.config.data + else: + config = default or {} + return GlobalConfig(deepcopy(config)) + + class Config(object): """ This class is responsible for: @@ -58,21 +174,8 @@ class Config(object): PATRONI_CONFIG_VARIABLE = PATRONI_ENV_PREFIX + 'CONFIGURATION' __CACHE_FILENAME = 'patroni.dynamic.json' - __REMAP_KEYS = { - 'master_start_timeout': 'primary_start_timeout', - 'master_stop_timeout': 'primary_stop_timeout' - } __DEFAULT_CONFIG = { 'ttl': 30, 'loop_wait': 10, 'retry_timeout': 10, - 'maximum_lag_on_failover': 1048576, - 'maximum_lag_on_syncnode': -1, - 'check_timeline': False, - 'primary_start_timeout': 300, - 'primary_stop_timeout': 0, - 'synchronous_mode': False, - 'synchronous_mode_strict': False, - 'synchronous_node_count': 1, - 'failsafe_mode': False, 'standby_cluster': { 'create_replica_methods': '', 'host': '', @@ -125,9 +228,6 @@ class Config(object): def dynamic_configuration(self): return deepcopy(self._dynamic_configuration) - def check_mode(self, mode): - return bool(parse_bool(self._dynamic_configuration.get(mode))) - def _load_config_path(self, path): """ If path is a file, loads the yml file pointed to by path. @@ -228,9 +328,6 @@ class Config(object): config = deepcopy(self.__DEFAULT_CONFIG) for name, value in dynamic_configuration.items(): - # allow copying master_start_timeout->primary_start_timeout when the latter isn't in dynamic_configuration - if name in self.__REMAP_KEYS and self.__REMAP_KEYS[name] not in dynamic_configuration: - name = self.__REMAP_KEYS[name] if name == 'postgresql': for name, value in (value or {}).items(): if name == 'parameters': @@ -243,10 +340,7 @@ class Config(object): if name in self.__DEFAULT_CONFIG['standby_cluster']: config['standby_cluster'][name] = deepcopy(value) elif name in config: # only variables present in __DEFAULT_CONFIG allowed to be overridden from DCS - if name in ('synchronous_mode', 'synchronous_mode_strict', 'failsafe_mode'): - config[name] = value - else: - config[name] = int(value) + config[name] = int(value) return config @staticmethod @@ -458,10 +552,6 @@ class Config(object): 'name', 'scope', 'retry_timeout', - 'synchronous_mode', - 'synchronous_mode_strict', - 'synchronous_node_count', - 'maximum_lag_on_syncnode', 'citus' ) @@ -480,3 +570,13 @@ class Config(object): def copy(self): return deepcopy(self.__effective_configuration) + + def get_global_config(self, cluster: Union[Cluster, None]) -> GlobalConfig: + """Instantiate :class:`GlobalConfig` based on input. + + Use the configuration from provided *cluster* (the most up-to-date) or from the + local cache if *cluster.config* is not initialized or doesn't have a valid config. + :param cluster: the currently known cluster state from DCS + :returns: :class:`GlobalConfig` object + """ + return get_global_config(cluster, self._dynamic_configuration) diff --git a/patroni/ctl.py b/patroni/ctl.py index e64db043..526ac560 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -35,7 +35,7 @@ except ImportError: # pragma: no cover from .dcs import get_dcs as _get_dcs from .exceptions import PatroniException from .postgresql.misc import postgres_version_to_int -from .utils import cluster_as_json, patch_config, polling_loop, is_standby_cluster +from .utils import cluster_as_json, patch_config, polling_loop from .request import PatroniRequest from .version import __version__ @@ -531,7 +531,8 @@ def parse_scheduled(scheduled): @option_force @click.pass_obj def reload(obj, cluster_name, member_names, group, force, role): - cluster = get_dcs(obj, cluster_name, group).get_cluster() + dcs = get_dcs(obj, cluster_name, group) + cluster = dcs.get_cluster() members = get_members(obj, cluster, cluster_name, member_names, role, force, 'reload', group=group) @@ -541,7 +542,7 @@ def reload(obj, cluster_name, member_names, group, force, role): click.echo('No changes to apply on member {0}'.format(member.name)) elif r.status == 202: click.echo('Reload request received for member {0} and will be processed within {1} seconds'.format( - member.name, cluster.config.data.get('loop_wait')) + member.name, cluster.config.data.get('loop_wait', dcs.loop_wait)) ) else: click.echo('Failed: reload for member {0}, status code={1}, ({2})'.format( @@ -597,7 +598,8 @@ def restart(obj, cluster_name, group, member_names, force, role, p_any, schedule content['postgres_version'] = version if scheduled_at: - if cluster.is_paused(): + from patroni.config import get_global_config + if get_global_config(cluster).is_paused: raise PatroniCtlException("Can't schedule restart in the paused state") content['schedule'] = scheduled_at.isoformat() @@ -691,7 +693,8 @@ def _do_failover_or_switchover(obj, action, cluster_name, group, leader, candida if force or action == 'failover': leader = cluster.leader and cluster.leader.name else: - prompt = 'Standby Leader' if is_standby_cluster(cluster.config) else 'Primary' + from patroni.config import get_global_config + prompt = 'Standby Leader' if get_global_config(cluster).is_standby_cluster else 'Primary' leader = click.prompt(prompt, type=str, default=cluster.leader.member.name) if leader is not None and cluster.leader and cluster.leader.member.name != leader: @@ -728,7 +731,8 @@ def _do_failover_or_switchover(obj, action, cluster_name, group, leader, candida scheduled_at = parse_scheduled(scheduled) if scheduled_at: - if cluster.is_paused(): + from patroni.config import get_global_config + if get_global_config(cluster).is_paused: raise PatroniCtlException("Can't schedule switchover in the paused state") scheduled_at_str = scheduled_at.isoformat() @@ -1008,9 +1012,10 @@ def wait_until_pause_is_applied(dcs, paused, old_cluster): def toggle_pause(config, cluster_name, group, paused, wait): + from patroni.config import get_global_config dcs = get_dcs(config, cluster_name, group) cluster = dcs.get_cluster() - if cluster.is_paused() == paused: + if get_global_config(cluster).is_paused == paused: raise PatroniCtlException('Cluster is {0} paused'.format(paused and 'already' or 'not')) for member in get_all_members_leader_first(cluster): diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index b63eb4c9..aa448efd 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -18,7 +18,7 @@ from typing import Any, Dict, List, Optional, Union from urllib.parse import urlparse, urlunparse, parse_qsl from ..exceptions import PatroniFatalException -from ..utils import deep_compare, parse_bool, uri +from ..utils import deep_compare, uri CITUS_COORDINATOR_GROUP_ID = 0 citus_group_re = re.compile('^(0|[1-9][0-9]*)$') @@ -521,15 +521,6 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,' candidates = [m for m in self.members if m.clonefrom and m.is_running and m.name not in exclude] return candidates[randint(0, len(candidates) - 1)] if candidates else self.leader - def check_mode(self, mode): - return bool(self.config and parse_bool(self.config.data.get(mode))) - - def is_paused(self): - return self.check_mode('pause') - - def is_synchronous_mode(self): - return self.check_mode('synchronous_mode') - @property def __permanent_slots(self): return self.config and self.config.permanent_slots or {} diff --git a/patroni/ha.py b/patroni/ha.py index 919e6f6e..402f9ba4 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -17,7 +17,7 @@ from .exceptions import DCSError, PostgresConnectionException, PatroniFatalExcep from .postgresql.callback_executor import CallbackAction from .postgresql.misc import postgres_version_to_int from .postgresql.rewind import Rewind -from .utils import polling_loop, tzutc, is_standby_cluster as _is_standby_cluster, parse_int +from .utils import polling_loop, tzutc from .dcs import Cluster, Leader, Member, RemoteMember logger = logging.getLogger(__name__) @@ -136,6 +136,7 @@ class Ha(object): self._rewind = Rewind(self.state_handler) self.dcs = patroni.dcs self.cluster = None + self.global_config = self.patroni.config.get_global_config(None) self.old_cluster = None self._is_leader = False self._is_leader_lock = RLock() @@ -166,33 +167,22 @@ class Ha(object): # used only in backoff after failing a pre_promote script self._released_leader_key_timestamp = 0 - def check_mode(self, mode): - # Try to protect from the case when DCS was wiped out during pause - if self.cluster and self.cluster.config and self.cluster.config.modify_index: - return self.cluster.check_mode(mode) - else: - return self.patroni.config.check_mode(mode) - - def primary_stop_timeout(self): - """ Primary stop timeout """ - ret = parse_int(self.patroni.config['primary_stop_timeout']) - return ret if ret and ret > 0 and self.is_synchronous_mode() else None + def primary_stop_timeout(self) -> Union[int, None]: + """:returns: "primary_stop_timeout" from the global configuration or `None` when not in synchronous mode.""" + ret = self.global_config.primary_stop_timeout + return ret if ret > 0 and self.is_synchronous_mode() else None def is_paused(self): - return self.check_mode('pause') + """:returns: `True` if in maintenance mode.""" + return self.global_config.is_paused def check_timeline(self): - return self.check_mode('check_timeline') - - def get_standby_cluster_config(self): - if self.cluster and self.cluster.config and self.cluster.config.modify_index: - config = self.cluster.config.data - else: - config = self.patroni.config.dynamic_configuration - return config.get('standby_cluster') + """:returns: `True` if should check whether the timeline is latest during the leader race.""" + return self.global_config.check_mode('check_timeline') def is_standby_cluster(self): - return _is_standby_cluster(self.get_standby_cluster_config()) + """:returns: `True` if global configuration has a valid "standby_cluster" section.""" + return self.global_config.is_standby_cluster def is_leader(self): with self._is_leader_lock: @@ -386,14 +376,15 @@ class Ha(object): else: return 'failed to acquire initialize lock' else: - create_replica_methods = self.get_standby_cluster_config().get('create_replica_methods', []) \ + create_replica_methods = self.global_config.get_standby_cluster_config().get('create_replica_methods', []) \ if self.is_standby_cluster() else None can_bootstrap = self.state_handler.can_create_replica_without_replication_connection(create_replica_methods) concurrent_bootstrap = self.cluster.initialize == "" if can_bootstrap and not concurrent_bootstrap: msg = 'bootstrap (without leader)' return self._async_executor.try_run_async(msg, self.clone) or 'trying to ' + msg - return 'waiting for {0}leader to bootstrap'.format('standby_' if self.is_standby_cluster() else '') + return 'waiting for {0}leader to bootstrap'.format( + 'standby_' if self.is_standby_cluster() else '') def bootstrap_standby_leader(self): """ If we found 'standby' key in the configuration, we need to bootstrap @@ -442,7 +433,7 @@ class Ha(object): self.watchdog.disable() if self.has_lock() and self.update_lock(): - timeout = self.patroni.config['primary_start_timeout'] + timeout = self.global_config.primary_start_timeout if timeout == 0: # We are requested to prefer failing over to restarting primary. But see first if there # is anyone to fail over to. @@ -495,9 +486,7 @@ class Ha(object): def _get_node_to_follow(self, cluster): # determine the node to follow. If replicatefrom tag is set, # try to follow the node mentioned there, otherwise, follow the leader. - standby_config = self.get_standby_cluster_config() - is_standby_cluster = _is_standby_cluster(standby_config) - if is_standby_cluster and (self.cluster.is_unlocked() or self.has_lock(False)): + if self.is_standby_cluster() and (self.cluster.is_unlocked() or self.has_lock(False)): node_to_follow = self.get_remote_member() elif self.patroni.replicatefrom and self.patroni.replicatefrom != self.state_handler.name: node_to_follow = cluster.get_member(self.patroni.replicatefrom) @@ -511,7 +500,8 @@ class Ha(object): params = ('restore_command', 'archive_cleanup_command') for param in params: # It is highly unlikely to happen, but we want to protect from the case node_to_follow.data.pop(param, None) # when above-mentioned params came from outside. - if is_standby_cluster: + if self.is_standby_cluster(): + standby_config = self.global_config.get_standby_cluster_config() node_to_follow.data.update({p: standby_config[p] for p in params if standby_config.get(p)}) return node_to_follow @@ -571,14 +561,13 @@ class Ha(object): return follow_reason - def is_synchronous_mode(self): - return self.check_mode('synchronous_mode') - - def is_synchronous_mode_strict(self): - return self.check_mode('synchronous_mode_strict') + def is_synchronous_mode(self) -> bool: + """:returns: `True` if synchronous replication is requested.""" + return self.global_config.is_synchronous_mode def is_failsafe_mode(self): - return self.check_mode('failsafe_mode') + """:returns: `True` if failsafe_mode is enabled in global configuration.""" + return self.global_config.check_mode('failsafe_mode') def process_sync_replication(self): """Process synchronous standby beahvior. @@ -590,11 +579,11 @@ class Ha(object): promoting standbys that were guaranteed to be replicating synchronously. """ if self.is_synchronous_mode(): - sync_node_count = self.patroni.config['synchronous_node_count'] + sync_node_count = self.global_config.synchronous_node_count + sync_node_maxlag = self.global_config.maximum_lag_on_syncnode current = [] if self.cluster.sync.is_empty else self.cluster.sync.members picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster, sync_node_count, - self.patroni.config[ - 'maximum_lag_on_syncnode']) + sync_node_maxlag) if set(picked) != set(current): # update synchronous standby list in dcs temporarily to point to common nodes in current and picked sync_common = list(set(current).intersection(set(allow_promote))) @@ -607,7 +596,7 @@ class Ha(object): return # Update db param and wait for x secs - if self.is_synchronous_mode_strict() and not picked: + if self.global_config.is_synchronous_mode_strict and not picked: picked = ['*'] logger.warning("No standbys available!") @@ -617,10 +606,8 @@ class Ha(object): if picked and picked[0] != '*' and set(allow_promote) != set(picked) and not allow_promote: # Wait for PostgreSQL to enable synchronous mode and see if we can immediately set sync_standby time.sleep(2) - _, allow_promote = self.state_handler.sync_handler.current_state(self.cluster, - sync_node_count, - self.patroni.config[ - 'maximum_lag_on_syncnode']) + _, allow_promote = self.state_handler.sync_handler.current_state(self.cluster, sync_node_count, + sync_node_maxlag) if allow_promote and set(allow_promote) != set(sync_common): try: cluster = self.dcs.get_cluster() @@ -748,7 +735,7 @@ class Ha(object): # promotion until next cycle. TODO: trigger immediate retry of run_cycle return 'Postponing promotion because synchronous replication state was updated by somebody else' self.state_handler.sync_handler.set_synchronous_standby_names( - ['*'] if self.is_synchronous_mode_strict() else []) + ['*'] if self.global_config.is_synchronous_mode_strict else []) if self.state_handler.role not in ('master', 'promoted', 'primary'): def on_success(): self._rewind.reset_state() @@ -835,7 +822,7 @@ class Ha(object): :returns True when node is lagging """ lag = (self.cluster.last_lsn or 0) - wal_position - return lag > self.patroni.config.get('maximum_lag_on_failover', 0) + return lag > self.global_config.maximum_lag_on_failover def _is_healthiest_node(self, members, check_replication_lag=True): """This method tries to determine whether I am healthy enough to became a new leader candidate or not.""" @@ -1376,7 +1363,7 @@ class Ha(object): # Now that restart is scheduled we can set timeout for startup, it will get reset # once async executor runs and main loop notices PostgreSQL as up. - timeout = restart_data.get('timeout', self.patroni.config['primary_start_timeout']) + timeout = restart_data.get('timeout', self.global_config.primary_start_timeout) self.set_start_timeout(timeout) def before_shutdown(): @@ -1439,7 +1426,7 @@ class Ha(object): """ if self.has_lock() and self.update_lock(): if self._async_executor.scheduled_action == 'doing crash recovery in a single user mode': - time_left = self.patroni.config['primary_start_timeout'] - (time.time() - self._crash_recovery_started) + time_left = self.global_config.primary_start_timeout - (time.time() - self._crash_recovery_started) if time_left <= 0 and self.is_failover_possible(self.cluster.members): logger.info("Demoting self because crash recovery is taking too long") self.state_handler.cancellable.cancel(True) @@ -1545,7 +1532,7 @@ class Ha(object): self.demote('immediate-nolock') return 'stopped PostgreSQL while starting up because leader key was lost' - timeout = self._start_timeout or self.patroni.config['primary_start_timeout'] + timeout = self._start_timeout or self.global_config.primary_start_timeout time_left = timeout - self.state_handler.time_in_state() if time_left <= 0: @@ -1577,9 +1564,10 @@ class Ha(object): try: try: self.load_cluster_from_dcs() - self.state_handler.reset_cluster_info_state(self.cluster, self.patroni.nofailover) + self.global_config = self.patroni.config.get_global_config(self.cluster) + self.state_handler.reset_cluster_info_state(self.cluster, self.patroni.nofailover, self.global_config) except Exception: - self.state_handler.reset_cluster_info_state(None, self.patroni.nofailover) + self.state_handler.reset_cluster_info_state(None) raise if self.is_paused(): @@ -1862,7 +1850,7 @@ class Ha(object): member to stream. Config can be both patroni config or cluster.config.data """ - cluster_params = self.get_standby_cluster_config() + cluster_params = self.global_config.get_standby_cluster_config() if cluster_params: name = member.name if member else 'remote_member:{}'.format(uuid.uuid1()) diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index c58b3290..0b1dba5a 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -12,7 +12,7 @@ from datetime import datetime from dateutil import tz from psutil import TimeoutExpired from threading import current_thread, Lock -from typing import Optional +from typing import Optional, Union, TYPE_CHECKING from .bootstrap import Bootstrap from .callback_executor import CallbackAction, CallbackExecutor @@ -25,10 +25,12 @@ from .postmaster import PostmasterProcess from .slots import SlotsHandler from .sync import SyncHandler from .. import psycopg -from ..dcs import Member +from ..dcs import Cluster, Member from ..exceptions import PostgresConnectionException from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int +if TYPE_CHECKING: # pragma: no cover + from .config import GlobalConfig logger = logging.getLogger(__name__) @@ -65,6 +67,7 @@ class Postgresql(object): self._version_file = os.path.join(self._data_dir, 'PG_VERSION') self._pg_control = os.path.join(self._data_dir, 'global', 'pg_control') self._major_version = self.get_major_version() + self._global_config = None self._state_lock = Lock() self.set_state('stopped') @@ -104,7 +107,6 @@ class Postgresql(object): self._cluster_info_state = {} self._has_permanent_logical_slots = True self._enforce_hot_standby_feedback = False - self._is_synchronous_mode = True self._cached_replica_timeline = None # Last known running process @@ -180,7 +182,8 @@ class Postgresql(object): "FROM pg_catalog.pg_stat_get_wal_senders() w," + " pg_catalog.pg_stat_get_activity(w.pid)" + " WHERE w.state = 'streaming') r)").format(self.wal_name, self.lsn_name) - if self._is_synchronous_mode and self.role in ('master', 'primary') else "'on', '', NULL") + if (not self._global_config or self._global_config.is_synchronous_mode) + and self.role in ('master', 'primary', 'promoted') else "'on', '', NULL") if self._major_version >= 90600: extra = ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, " + @@ -351,7 +354,18 @@ class Postgresql(object): self.config.write_postgresql_conf() self.reload() - def reset_cluster_info_state(self, cluster, nofailover=None): + def reset_cluster_info_state(self, cluster: Union[Cluster, None], nofailover: Optional[bool] = None, + global_config: Optional['GlobalConfig'] = None) -> None: + """Reset monitoring query cache. + + It happens in the beginning of heart-beat loop and on change of `synchronous_standby_names`. + + :param cluster: currently known cluster state from DCS + :param nofailover: whether this node could become a new primary. + Important when there are logical permanent replication slots because "nofailover" + node could do cascading replication and should enable `hot_standby_feedback` + :param global_config: last known :class:`GlobalConfig` object + """ self._cluster_info_state = {} if cluster and cluster.config and cluster.config.modify_index: self._has_permanent_logical_slots =\ @@ -363,7 +377,7 @@ class Postgresql(object): self._has_permanent_logical_slots or cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version)) - self._is_synchronous_mode = cluster.is_synchronous_mode() + self._global_config = global_config def _cluster_info_state_get(self, name): if not self._cluster_info_state: diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 97ee4dd5..87544558 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -10,7 +10,7 @@ from urllib.parse import urlparse, parse_qsl, unquote from .validator import CaseInsensitiveDict, recovery_parameters,\ transform_postgresql_parameter_value, transform_recovery_parameter_value -from ..dcs import slot_name_from_member_name, RemoteMember +from ..dcs import 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 @@ -838,9 +838,10 @@ class ConfigHandler(object): parameters = config['parameters'].copy() listen_addresses, port = split_host_port(config['listen'], 5432) parameters.update(cluster_name=self._postgresql.scope, listen_addresses=listen_addresses, port=str(port)) - if config.get('synchronous_mode', False): + if not self._postgresql._global_config or self._postgresql._global_config.is_synchronous_mode: if self._synchronous_standby_names is None: - if config.get('synchronous_mode_strict', False): + if self._postgresql._global_config and self._postgresql._global_config.is_synchronous_mode_strict\ + and self._postgresql.role in ('master', 'primary', 'promoted'): parameters['synchronous_standby_names'] = '*' else: parameters.pop('synchronous_standby_names', None) diff --git a/patroni/utils.py b/patroni/utils.py index fae246dc..1939185a 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -28,8 +28,9 @@ from urllib3.response import HTTPResponse from .exceptions import PatroniException from .version import __version__ -if TYPE_CHECKING: +if TYPE_CHECKING: # pragma: no cover from .dcs import Cluster + from .config import GlobalConfig tzutc = tz.tzutc() @@ -693,22 +694,12 @@ def iter_response_objects(response: HTTPResponse) -> Iterator[Dict[str, Any]]: prev = chunk[idx:] -def is_standby_cluster(config: Union[Dict[str, Any], None]) -> bool: - """Check provided configuration describes a standby cluster. - - :param config: the configuration to be checked. It is expected to be the :class:`dict` that represents the value of - the ``standby_cluster`` key in the main Patroni configuration. ``None`` can be used if ``standby_cluster`` is - absent in the main Patroni configuration. - - :returns: ``True`` if configuration is a Patroni standby cluster. - """ - return isinstance(config, dict) and (config.get('host') or config.get('port') or config.get('restore_command')) - - -def cluster_as_json(cluster: 'Cluster') -> Dict[str, Any]: +def cluster_as_json(cluster: 'Cluster', global_config: Optional['GlobalConfig'] = None) -> Dict[str, Any]: """Get a JSON representation of *cluster*. :param cluster: the :class:`Cluster` object to be parsed as JSON. + :param global_config: optional :class:`GlobalConfig` object to check the cluster state. + if not provided will be instantiated from the `Cluster.config`. :returns: JSON representation of *cluster*. @@ -734,14 +725,16 @@ def cluster_as_json(cluster: 'Cluster') -> Dict[str, Any]: * ``from``: name of the member to be demoted; * ``to``: name of the member to be promoted. """ + if not global_config: + from patroni.config import get_global_config + global_config = get_global_config(cluster) leader_name = cluster.leader.name if cluster.leader else None cluster_lsn = cluster.last_lsn or 0 ret = {'members': []} for m in cluster.members: if m.name == leader_name: - config = cluster.config.data if cluster.config and cluster.config.modify_index else {} - role = 'standby_leader' if is_standby_cluster(config.get('standby_cluster')) else 'leader' + role = 'standby_leader' if global_config.is_standby_cluster else 'leader' elif cluster.sync.matches(m.name): role = 'sync_standby' else: @@ -769,7 +762,7 @@ def cluster_as_json(cluster: 'Cluster') -> Dict[str, Any]: # sort members by name for consistency ret['members'].sort(key=lambda m: m['name']) - if cluster.is_paused(): + if global_config.is_paused: ret['pause'] = True if cluster.failover and cluster.failover.scheduled_at: ret['scheduled_switchover'] = {'at': cluster.failover.scheduled_at.isoformat()} diff --git a/tests/test_api.py b/tests/test_api.py index 42b41d03..f167735e 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -11,6 +11,7 @@ from mock import Mock, PropertyMock, patch from socketserver import ThreadingMixIn from patroni.api import RestApiHandler, RestApiServer +from patroni.config import GlobalConfig from patroni.dcs import ClusterConfig, Member from patroni.ha import _MemberStatus from patroni.utils import tzutc @@ -116,10 +117,6 @@ class MockHa(object): def is_paused(): return True - @staticmethod - def is_standby_cluster(): - return False - class MockLogger(object): @@ -128,10 +125,16 @@ class MockLogger(object): records_lost = 1 +class MockConfig(object): + + def get_global_config(self, _): + return GlobalConfig({}) + + class MockPatroni(object): ha = MockHa() - config = Mock() + config = MockConfig() postgresql = ha.state_handler dcs = Mock() logger = MockLogger() @@ -185,7 +188,8 @@ class TestRestApiHandler(unittest.TestCase): def test_do_GET(self): MockPatroni.dcs.cluster.last_lsn = 20 MockPatroni.dcs.cluster.sync.members = [MockPostgresql.name] - MockRestApiServer(RestApiHandler, 'GET /replica') + with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=True)): + MockRestApiServer(RestApiHandler, 'GET /replica') MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M') MockRestApiServer(RestApiHandler, 'GET /replica?lag=10MB') MockRestApiServer(RestApiHandler, 'GET /replica?lag=10485760') @@ -207,7 +211,7 @@ class TestRestApiHandler(unittest.TestCase): with patch.object(MockHa, 'is_leader', Mock(return_value=True)): MockRestApiServer(RestApiHandler, 'GET /replica') MockRestApiServer(RestApiHandler, 'GET /read-only-sync') - with patch.object(MockHa, 'is_standby_cluster', Mock(return_value=True)): + with patch.object(GlobalConfig, 'is_standby_cluster', Mock(return_value=True)): MockRestApiServer(RestApiHandler, 'GET /standby_leader') MockPatroni.dcs.cluster = None with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'primary'})): @@ -217,7 +221,8 @@ class TestRestApiHandler(unittest.TestCase): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /primary')) with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, '')])): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni')) - with patch.object(MockHa, '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') # test tags @@ -405,9 +410,7 @@ class TestRestApiHandler(unittest.TestCase): def test_do_POST_sigterm(self): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /sigterm HTTP/1.0' + self._authorization)) - @patch.object(MockPatroni, 'dcs') - def test_do_POST_restart(self, mock_dcs): - mock_dcs.get_cluster.return_value.is_paused.return_value = False + def test_do_POST_restart(self): request = 'POST /restart HTTP/1.0' + self._authorization self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) @@ -449,12 +452,12 @@ class TestRestApiHandler(unittest.TestCase): request = make_request(role='primary', postgres_version='9.5.2') MockRestApiServer(RestApiHandler, request) - mock_dcs.get_cluster.return_value.is_paused.return_value = True - MockRestApiServer(RestApiHandler, make_request(schedule='2016-08-42 12:45TZ+1', role='primary')) - # Valid timeout - MockRestApiServer(RestApiHandler, make_request(timeout='60s')) - # Invalid timeout - MockRestApiServer(RestApiHandler, make_request(timeout='42towels')) + with patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)): + MockRestApiServer(RestApiHandler, make_request(schedule='2016-08-42 12:45TZ+1', role='primary')) + # Valid timeout + MockRestApiServer(RestApiHandler, make_request(timeout='60s')) + # Invalid timeout + MockRestApiServer(RestApiHandler, make_request(timeout='42towels')) def test_do_DELETE_restart(self): for retval in (True, False): @@ -471,10 +474,7 @@ class TestRestApiHandler(unittest.TestCase): mock_dcs.get_cluster.return_value.failover = None self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) - @patch.object(MockPatroni, 'dcs') - def test_do_POST_reinitialize(self, mock_dcs): - cluster = mock_dcs.get_cluster.return_value - cluster.is_paused.return_value = False + def test_do_POST_reinitialize(self): request = 'POST /reinitialize HTTP/1.0' + self._authorization + '\nContent-Length: 15\n\n{"force": true}' MockRestApiServer(RestApiHandler, request) with patch.object(MockHa, 'reinitialize', Mock(return_value=None)): @@ -492,8 +492,6 @@ class TestRestApiHandler(unittest.TestCase): def test_do_POST_switchover(self, dcs): dcs.loop_wait = 10 cluster = dcs.get_cluster.return_value - cluster.is_synchronous_mode.return_value = False - cluster.is_paused.return_value = False post = 'POST /switchover HTTP/1.0' + self._authorization + '\nContent-Length: ' @@ -507,21 +505,22 @@ class TestRestApiHandler(unittest.TestCase): request = post + '25\n\n{"leader": "postgresql1"}' - cluster.is_paused.return_value = True - MockRestApiServer(RestApiHandler, request) - - cluster.is_paused.return_value = False - for cluster.is_synchronous_mode.return_value in (True, False): + with patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)): MockRestApiServer(RestApiHandler, request) + for is_synchronous_mode in (True, False): + with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=is_synchronous_mode)): + MockRestApiServer(RestApiHandler, request) + cluster.leader.name = 'postgresql2' request = post + '53\n\n{"leader": "postgresql1", "candidate": "postgresql2"}' MockRestApiServer(RestApiHandler, request) cluster.leader.name = 'postgresql1' cluster.sync.matches.return_value = False - for cluster.is_synchronous_mode.return_value in (True, False): - MockRestApiServer(RestApiHandler, request) + for is_synchronous_mode in (True, False): + with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=is_synchronous_mode)): + MockRestApiServer(RestApiHandler, request) cluster.members = [Member(0, 'postgresql0', 30, {'api_url': 'http'}), Member(0, 'postgresql2', 30, {'api_url': 'http'})] @@ -555,7 +554,8 @@ class TestRestApiHandler(unittest.TestCase): request = post + '103\n\n{"leader": "postgresql1", "member": "postgresql2",' +\ ' "scheduled_at": "6016-02-15T18:13:30.568224+01:00"}' MockRestApiServer(RestApiHandler, request) - with patch.object(MockPatroni, 'dcs') as d: + with patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)),\ + patch.object(MockPatroni, 'dcs') as d: d.manual_failover.return_value = False MockRestApiServer(RestApiHandler, request) @@ -571,13 +571,11 @@ class TestRestApiHandler(unittest.TestCase): # Invalid date self.assertIsNotNone(MockRestApiServer(RestApiHandler, request + '2010-02-29T18:13:30.568224+01:00"}')) - @patch.object(MockPatroni, 'dcs', Mock()) def test_do_POST_failover(self): post = 'POST /failover HTTP/1.0' + self._authorization + '\nContent-Length: ' MockRestApiServer(RestApiHandler, post + '14\n\n{"leader":"1"}') MockRestApiServer(RestApiHandler, post + '37\n\n{"candidate":"2","scheduled_at": "1"}') - @patch.object(MockPatroni, 'dcs', Mock()) @patch.object(MockHa, 'is_leader', Mock(return_value=True)) def test_do_POST_citus(self): post = 'POST /citus HTTP/1.0' + self._authorization + '\nContent-Length: ' diff --git a/tests/test_config.py b/tests/test_config.py index 4a042945..3cd94a9e 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -20,9 +20,7 @@ class TestConfig(unittest.TestCase): def test_set_dynamic_configuration(self): with patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception)): self.assertIsNone(self.config.set_dynamic_configuration({'foo': 'bar'})) - self.assertTrue(self.config.set_dynamic_configuration({'synchronous_mode': True, - 'standby_cluster': {}, 'master_start_timeout': 1})) - self.assertEqual(self.config.get('primary_start_timeout'), 1) + self.assertTrue(self.config.set_dynamic_configuration({'standby_cluster': {}})) def test_reload_local_configuration(self): os.environ.update({ diff --git a/tests/test_ctl.py b/tests/test_ctl.py index e71ecaa5..7f300deb 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -5,7 +5,7 @@ import unittest from click.testing import CliRunner from datetime import datetime, timedelta -from mock import patch, Mock +from mock import patch, Mock, PropertyMock from patroni.ctl import ctl, load_config, output_members, get_dcs, parse_dcs, \ get_all_members, get_any_member, get_cursor, query_member, PatroniCtlException, apply_config_changes, \ format_config_for_editing, show_diff, invoke_editor, format_pg_version, CONFIG_FILE_PATH, PatronictlPrettyTable @@ -98,7 +98,7 @@ class TestCtl(unittest.TestCase): input='leader\nother\n2300-01-01T12:23:00\ny') assert result.exit_code == 0 - with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): + with patch('patroni.config.GlobalConfig.is_paused', PropertyMock(return_value=True)): result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force', '--scheduled', '2015-01-01T12:00:00']) assert result.exit_code == 1 @@ -309,7 +309,7 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30']) assert 'Failed: flush scheduled restart' in result.output - with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): + with patch('patroni.config.GlobalConfig.is_paused', PropertyMock(return_value=True)): result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30']) assert result.exit_code == 1 @@ -491,7 +491,7 @@ class TestCtl(unittest.TestCase): assert 'Failed' in result.output mock_post.return_value.status = 200 - with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): + with patch('patroni.config.GlobalConfig.is_paused', PropertyMock(return_value=True)): result = self.runner.invoke(ctl, ['pause', 'dummy']) assert 'Cluster is already paused' in result.output @@ -512,11 +512,11 @@ class TestCtl(unittest.TestCase): mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader mock_post.return_value.status = 200 - with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=False)): + with patch('patroni.config.GlobalConfig.is_paused', PropertyMock(return_value=False)): result = self.runner.invoke(ctl, ['resume', 'dummy']) assert 'Cluster is not paused' in result.output - with patch('patroni.dcs.Cluster.is_paused', Mock(return_value=True)): + with patch('patroni.config.GlobalConfig.is_paused', PropertyMock(return_value=True)): result = self.runner.invoke(ctl, ['resume', 'dummy']) assert 'Success' in result.output diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 0bb597f8..817358a2 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -256,7 +256,6 @@ class TestEtcd(unittest.TestCase): def test_get_cluster(self): cluster = self.etcd.get_cluster() self.assertIsInstance(cluster, Cluster) - self.assertFalse(cluster.is_synchronous_mode()) self.etcd._base_path = '/service/legacy' self.assertIsInstance(self.etcd.get_cluster(), Cluster) self.etcd._base_path = '/service/broken' diff --git a/tests/test_ha.py b/tests/test_ha.py index 70e465c6..bd7516ec 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -136,7 +136,6 @@ zookeeper: sys.argv = sys.argv[:1] self.config = Config(None) - self.config.set_dynamic_configuration({'maximum_lag_on_failover': 5}) self.version = '1.5.7' self.postgresql = p self.dcs = d @@ -300,7 +299,8 @@ class TestHa(PostgresInit): self.ha._async_executor.schedule('doing crash recovery in a single user mode') self.ha.state_handler.cancellable._process = Mock() self.ha._crash_recovery_started -= 600 - self.ha.patroni.config.set_dynamic_configuration({'maximum_lag_on_failover': 10}) + self.ha.cluster.config.data.update({'maximum_lag_on_failover': 10}) + self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster) self.assertEqual(self.ha.run_cycle(), 'terminated crash recovery because of startup timeout') @patch.object(Rewind, 'ensure_clean_shutdown', Mock()) @@ -488,6 +488,7 @@ class TestHa(PostgresInit): def test_check_failsafe_topology(self): self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly')) self.ha.cluster = get_cluster_initialized_with_leader_and_failsafe() + self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster) self.ha.dcs._last_failsafe = self.ha.cluster.failsafe self.assertEqual(self.ha.run_cycle(), 'demoting self because DCS is not accessible and I was a leader') self.ha.state_handler.name = self.ha.cluster.leader.name @@ -507,6 +508,7 @@ class TestHa(PostgresInit): def test_no_dcs_connection_primary_failsafe(self): self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly')) self.ha.cluster = get_cluster_initialized_with_leader_and_failsafe() + self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster) self.ha.dcs._last_failsafe = self.ha.cluster.failsafe self.ha.state_handler.name = self.ha.cluster.leader.name self.assertEqual(self.ha.run_cycle(), @@ -523,6 +525,7 @@ class TestHa(PostgresInit): def test_no_dcs_connection_replica_failsafe(self): self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly')) self.ha.cluster = get_cluster_initialized_with_leader_and_failsafe() + self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster) self.ha.update_failsafe({'name': 'leader', 'api_url': 'http://127.0.0.1:8008/patroni', 'conn_url': 'postgres://127.0.0.1:5432/postgres', 'slots': {'foo': 1000}}) self.p.is_leader = false @@ -685,6 +688,8 @@ class TestHa(PostgresInit): self.ha.fetch_node_status = get_node_status(timeline=1) self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock') self.ha.fetch_node_status = get_node_status(wal_position=1) + self.ha.cluster.config.data.update({'maximum_lag_on_failover': 5}) + self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster) self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock') # manual failover from the previous leader to us won't happen if we hold the nofailover flag self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, None)) @@ -841,6 +846,7 @@ class TestHa(PostgresInit): def test__is_healthiest_node(self): self.ha.cluster = get_cluster_initialized_without_leader(sync=('postgresql1', self.p.name)) + self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster) self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members)) self.p.is_leader = false self.ha.fetch_node_status = get_node_status() # accessible, in_recovery @@ -852,6 +858,8 @@ class TestHa(PostgresInit): # in synchronous_mode consider itself healthy if the former leader is accessible in read-only and ahead of us with patch.object(Ha, 'is_synchronous_mode', Mock(return_value=True)): self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members)) + self.ha.cluster.config.data.update({'maximum_lag_on_failover': 5}) + self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster) with patch('patroni.postgresql.Postgresql.last_operation', return_value=1): self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members)) with patch('patroni.postgresql.Postgresql.replica_cached_timeline', return_value=1): @@ -1067,8 +1075,8 @@ class TestHa(PostgresInit): def test_failover_immediately_on_zero_primary_start_timeout(self, demote): self.p.is_running = false self.ha.cluster = get_cluster_initialized_with_leader(sync=(self.p.name, 'other')) - self.ha.cluster.config.data['synchronous_mode'] = True - self.ha.patroni.config.set_dynamic_configuration({'primary_start_timeout': 0}) + self.ha.cluster.config.data.update({'synchronous_mode': True, 'primary_start_timeout': 0}) + self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster) self.ha.has_lock = true self.ha.update_lock = true self.ha.fetch_node_status = get_node_status() # accessible, in_recovery @@ -1077,13 +1085,14 @@ class TestHa(PostgresInit): def test_primary_stop_timeout(self): self.assertEqual(self.ha.primary_stop_timeout(), None) - self.ha.patroni.config.set_dynamic_configuration({'primary_stop_timeout': 30}) + self.ha.cluster.config.data.update({'primary_stop_timeout': 30}) + self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster) with patch.object(Ha, 'is_synchronous_mode', Mock(return_value=True)): self.assertEqual(self.ha.primary_stop_timeout(), 30) - self.ha.patroni.config.set_dynamic_configuration({'primary_stop_timeout': 30}) with patch.object(Ha, 'is_synchronous_mode', Mock(return_value=False)): self.assertEqual(self.ha.primary_stop_timeout(), None) - self.ha.patroni.config.set_dynamic_configuration({'primary_stop_timeout': None}) + self.ha.cluster.config.data['primary_stop_timeout'] = None + self.ha.global_config = self.ha.patroni.config.get_global_config(self.ha.cluster) self.assertEqual(self.ha.primary_stop_timeout(), None) @patch('patroni.postgresql.Postgresql.follow') @@ -1171,9 +1180,9 @@ class TestHa(PostgresInit): # Test sync set to '*' when synchronous_mode_strict is enabled mock_set_sync.reset_mock() - self.ha.is_synchronous_mode_strict = true self.p.sync_handler.current_state = Mock(return_value=([], [])) - self.ha.run_cycle() + with patch('patroni.config.GlobalConfig.is_synchronous_mode_strict', PropertyMock(return_value=True)): + self.ha.run_cycle() mock_set_sync.assert_called_once_with(['*']) def test_sync_replication_become_primary(self): diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 64c534ca..f8fa13c1 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -10,6 +10,7 @@ from mock import Mock, MagicMock, PropertyMock, patch, mock_open import patroni.psycopg as psycopg from patroni.async_executor import CriticalTask +from patroni.config import GlobalConfig from patroni.dcs import RemoteMember from patroni.exceptions import PostgresConnectionException, PatroniException from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE @@ -644,9 +645,10 @@ class TestPostgresql(BaseTestPostgresql): self.assertIsNone(self.p.wait_for_startup()) def test_get_server_parameters(self): - config = {'synchronous_mode': True, 'parameters': {'wal_level': 'hot_standby'}, 'listen': '0'} + config = {'parameters': {'wal_level': 'hot_standby'}, 'listen': '0'} + self.p._global_config = GlobalConfig({'synchronous_mode': True}) self.p.config.get_server_parameters(config) - config['synchronous_mode_strict'] = True + self.p._global_config = GlobalConfig({'synchronous_mode': True, 'synchronous_mode_strict': True}) self.p.config.get_server_parameters(config) self.p.config.set_synchronous_standby_names('foo') self.assertTrue(str(self.p.config.get_server_parameters(config)).startswith('{'))