gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-26 15:40:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Factor out global configuration into a dedicated class (#2628)

Browse Source 
It will help to avoid code duplications.
This commit is contained in:
Alexander Kukushkin
2023-04-03 08:09:29 +02:00
committed by GitHub
parent c549ea7d5c
commit 6f357a4e17
14 changed files with 329 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/api.py
+51 -24
View File
@@ -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():
patroni/config.py
+132 -32
View File
@@ -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)
patroni/ctl.py
+12 -7
View File
@@ -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):
patroni/dcs/__init__.py
+1 -10
View File
@@ -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 {}
patroni/ha.py
+39 -51
View File
@@ -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())
patroni/postgresql/__init__.py
+20 -6
View File
@@ -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:
patroni/postgresql/config.py
+4 -3
View File
@@ -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)
patroni/utils.py
+10 -17
View File
@@ -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()}
tests/test_api.py
+31 -33
View File
@@ -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: '
tests/test_config.py
+1 -3
View File
@@ -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({
tests/test_ctl.py
+6 -6
View File
@@ -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
tests/test_etcd.py
-1
View File
@@ -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'
tests/test_ha.py
+18 -9
View File
@@ -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):
tests/test_postgresql.py
+4 -2
View File
@@ -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('{'))