mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
Refactor manual failover checks
- Implement the dedicated class that represents a manual failover request - Move manual failover/switchover prechecks to the class method and use for both ctl and api - Use a single parse_schedule function in both ctl and api - Implement has_members_eligible_to_promote Ha method - Fix get_members + role='any' exception msg
This commit is contained in:
Polina Bungina
+16
-109
@@ -12,7 +12,6 @@ import json
|
||||
import logging
|
||||
import time
|
||||
import traceback
|
||||
import dateutil.parser
|
||||
import datetime
|
||||
import os
|
||||
import socket
|
||||
@@ -28,11 +27,11 @@ from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, TYPE_CH
|
||||
|
||||
from . import psycopg
|
||||
from .__main__ import Patroni
|
||||
from .dcs import Cluster
|
||||
from .exceptions import PostgresConnectionException, PostgresException
|
||||
from .manual_failover import ManualFailover
|
||||
from .postgresql.misc import postgres_version_to_int
|
||||
from .utils import deep_compare, enable_keepalive, parse_bool, patch_config, Retry, \
|
||||
RetryFailedError, parse_int, split_host_port, tzutc, uri, cluster_as_json
|
||||
RetryFailedError, parse_int, parse_schedule, split_host_port, tzutc, uri, cluster_as_json
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -777,44 +776,6 @@ class RestApiHandler(BaseHTTPRequestHandler):
|
||||
self.server.patroni.api_sigterm()
|
||||
self.write_response(202, 'shutdown scheduled')
|
||||
|
||||
@staticmethod
|
||||
def parse_schedule(schedule: str,
|
||||
action: str) -> Tuple[Union[int, None], Union[str, None], Union[datetime.datetime, None]]:
|
||||
"""Parse the given *schedule* and validate it.
|
||||
|
||||
:param schedule: a string representing a timestamp, e.g. ``2023-04-14T20:27:00+00:00``.
|
||||
:param action: the action to be scheduled (``restart``, ``switchover``, or ``failover``).
|
||||
|
||||
:returns: a tuple composed of 3 items:
|
||||
|
||||
* Suggested HTTP status code for a response:
|
||||
|
||||
* ``None``: if no issue was faced while parsing, leaving it up to the caller to decide the status; or
|
||||
* ``400``: if no timezone information could be found in *schedule*; or
|
||||
* ``422``: if *schedule* is invalid -- in the past or not parsable.
|
||||
|
||||
* An error message, if any error is faced, otherwise ``None``;
|
||||
* Parsed *schedule*, if able to parse, otherwise ``None``.
|
||||
|
||||
"""
|
||||
error = None
|
||||
scheduled_at = None
|
||||
try:
|
||||
scheduled_at = dateutil.parser.parse(schedule)
|
||||
if scheduled_at.tzinfo is None:
|
||||
error = 'Timezone information is mandatory for the scheduled {0}'.format(action)
|
||||
status_code = 400
|
||||
elif scheduled_at < datetime.datetime.now(tzutc):
|
||||
error = 'Cannot schedule {0} in the past'.format(action)
|
||||
status_code = 422
|
||||
else:
|
||||
status_code = None
|
||||
except (ValueError, TypeError):
|
||||
logger.exception('Invalid scheduled %s time: %s', action, schedule)
|
||||
error = 'Unable to parse scheduled timestamp. It should be in an unambiguous format, e.g. ISO 8601'
|
||||
status_code = 422
|
||||
return status_code, error, scheduled_at
|
||||
|
||||
@check_access
|
||||
def do_POST_restart(self) -> None:
|
||||
"""Handle a ``POST`` request to ``/restart`` path.
|
||||
@@ -870,9 +831,9 @@ class RestApiHandler(BaseHTTPRequestHandler):
|
||||
|
||||
for k in request:
|
||||
if k == 'schedule':
|
||||
(_, data, request[k]) = self.parse_schedule(request[k], "restart")
|
||||
if _:
|
||||
status_code = _
|
||||
parse_result, request[k] = parse_schedule(request[k])
|
||||
if parse_result:
|
||||
data, status_code = parse_result.value[0], parse_result.value[1]
|
||||
break
|
||||
elif k == 'role':
|
||||
if request[k] not in ('master', 'primary', 'replica'):
|
||||
@@ -1022,39 +983,6 @@ 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: Cluster, leader: Optional[str], candidate: Optional[str],
|
||||
action: str) -> Optional[str]:
|
||||
"""Checks whether there are nodes that could take over after demoting the primary.
|
||||
|
||||
:param cluster: the Patroni cluster.
|
||||
:param leader: name of the current Patroni leader.
|
||||
:param candidate: name of the Patroni node to be promoted.
|
||||
:param action: the action to be performed (``switchover`` or ``failover``).
|
||||
|
||||
: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 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 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'
|
||||
else:
|
||||
members = [m for m in cluster.members if not cluster.leader or m.name != cluster.leader.name and m.api_url]
|
||||
if not members:
|
||||
return action + ' is not possible: cluster does not have members except leader'
|
||||
for st in self.server.patroni.ha.fetch_nodes_statuses(members):
|
||||
if st.failover_limitation() is None:
|
||||
return None
|
||||
return action + ' is not possible: no good candidates have been found'
|
||||
|
||||
@check_access
|
||||
def do_POST_failover(self, action: str = 'failover') -> None:
|
||||
"""Handle a ``POST`` request to ``/failover`` path.
|
||||
@@ -1083,7 +1011,6 @@ class RestApiHandler(BaseHTTPRequestHandler):
|
||||
:param action: the action to be performed (``switchover`` or ``failover``).
|
||||
"""
|
||||
request = self._read_json_content()
|
||||
(status_code, data) = (400, '')
|
||||
if not request:
|
||||
return
|
||||
|
||||
@@ -1096,33 +1023,15 @@ class RestApiHandler(BaseHTTPRequestHandler):
|
||||
logger.info("received %s request with leader=%s candidate=%s scheduled_at=%s",
|
||||
action, leader, candidate, scheduled_at)
|
||||
|
||||
if action == 'failover' and not candidate:
|
||||
data = 'Failover could be performed only to a specific candidate'
|
||||
elif action == 'switchover' and not leader:
|
||||
data = 'Switchover could be performed only from a specific leader'
|
||||
manual_failover = ManualFailover(action, cluster, leader, candidate, scheduled_at,
|
||||
global_config.is_paused, global_config.is_synchronous_mode,
|
||||
self.server.patroni)
|
||||
data, status_code = manual_failover.run_precheck().value
|
||||
|
||||
if not data and scheduled_at:
|
||||
if action == 'failover':
|
||||
data = "Failover can't be scheduled"
|
||||
elif global_config.is_paused:
|
||||
data = "Can't schedule switchover in the paused state"
|
||||
else:
|
||||
(status_code, data, scheduled_at) = self.parse_schedule(scheduled_at, action)
|
||||
|
||||
if not data and global_config.is_paused and not candidate:
|
||||
data = 'Switchover is possible only to a specific candidate in a paused state'
|
||||
|
||||
if action == 'failover' and leader:
|
||||
logger.warning('received failover request with leader specifed - performing switchover instead')
|
||||
action = 'switchover'
|
||||
|
||||
if not data and leader == candidate:
|
||||
data = 'Switchover target and source are the same'
|
||||
|
||||
if not data and not scheduled_at:
|
||||
data = self.is_failover_possible(cluster, leader, candidate, action)
|
||||
if data:
|
||||
status_code = 412
|
||||
parse_result, scheduled_at = manual_failover.parse_scheduled()
|
||||
if parse_result:
|
||||
data, status_code = parse_result.value[0], parse_result.value[1]
|
||||
|
||||
if not data:
|
||||
if self.server.patroni.dcs.manual_failover(leader, candidate, scheduled_at=scheduled_at):
|
||||
@@ -1136,12 +1045,10 @@ class RestApiHandler(BaseHTTPRequestHandler):
|
||||
else:
|
||||
data = 'failed to write failover key into DCS'
|
||||
status_code = 503
|
||||
# pyright thinks ``status_code`` can be ``None`` because ``parse_schedule`` call may return ``None``. However,
|
||||
# if that's the case, ``status_code`` will be overwritten somewhere between ``parse_schedule`` and
|
||||
# ``write_response`` calls.
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
assert isinstance(status_code, int)
|
||||
self.write_response(status_code, data)
|
||||
|
||||
status_code = status_code or 400
|
||||
self.write_response(status_code, data.format(action=action, leader=leader, candidate=candidate,
|
||||
cluster_name=self.server.patroni.postgresql.scope))
|
||||
|
||||
def do_POST_switchover(self) -> None:
|
||||
"""Handle a ``POST`` request to ``/switchover`` path.
|
||||
|
||||
+57
-93
@@ -16,8 +16,6 @@ import click
|
||||
import codecs
|
||||
import copy
|
||||
import datetime
|
||||
import dateutil.parser
|
||||
import dateutil.tz
|
||||
import difflib
|
||||
import io
|
||||
import json
|
||||
@@ -49,8 +47,9 @@ except ImportError: # pragma: no cover
|
||||
from .config import Config, get_global_config
|
||||
from .dcs import get_dcs as _get_dcs, AbstractDCS, Cluster, Member
|
||||
from .exceptions import PatroniException
|
||||
from .manual_failover import ManualFailover
|
||||
from .postgresql.misc import postgres_version_to_int
|
||||
from .utils import cluster_as_json, patch_config, polling_loop
|
||||
from .utils import cluster_as_json, parse_schedule, patch_config, polling_loop
|
||||
from .request import PatroniRequest
|
||||
from .version import __version__
|
||||
|
||||
@@ -645,7 +644,8 @@ def get_members(obj: Dict[str, Any], cluster: Cluster, cluster_name: str, member
|
||||
if member_names:
|
||||
member_names = list(set(member_names) & candidates)
|
||||
if not member_names:
|
||||
raise PatroniCtlException('No {0} among provided members'.format(role))
|
||||
raise PatroniCtlException(
|
||||
'No{0} among provided members'.format('t a single cluster member' if role == 'any' else ' ' + role))
|
||||
elif action != 'reinitialize':
|
||||
member_names = list(candidates)
|
||||
|
||||
@@ -945,43 +945,6 @@ def check_response(response: urllib3.response.HTTPResponse, member_name: str,
|
||||
return True
|
||||
|
||||
|
||||
def parse_scheduled(scheduled: Optional[str]) -> Optional[datetime.datetime]:
|
||||
"""Parse a string *scheduled* timestamp as a :class:`~datetime.datetime` object.
|
||||
|
||||
:param scheduled: string representation of the timestamp. May also be ``now``.
|
||||
|
||||
:returns: the corresponding :class:`~datetime.datetime` object, if *scheduled* is not ``now``, otherwise ``None``.
|
||||
|
||||
:raises:
|
||||
:class:`PatroniCtlException`: if unable to parse *scheduled* from :class:`str` to :class:`~datetime.datetime`.
|
||||
|
||||
:Example:
|
||||
|
||||
>>> parse_scheduled(None) is None
|
||||
True
|
||||
|
||||
>>> parse_scheduled('now') is None
|
||||
True
|
||||
|
||||
>>> parse_scheduled('2023-05-29T04:32:31')
|
||||
datetime.datetime(2023, 5, 29, 4, 32, 31, tzinfo=tzlocal())
|
||||
|
||||
>>> parse_scheduled('2023-05-29T04:32:31-3')
|
||||
datetime.datetime(2023, 5, 29, 4, 32, 31, tzinfo=tzoffset(None, -10800))
|
||||
"""
|
||||
if scheduled is not None and (scheduled or 'now') != 'now':
|
||||
try:
|
||||
scheduled_at = dateutil.parser.parse(scheduled)
|
||||
if scheduled_at.tzinfo is None:
|
||||
scheduled_at = scheduled_at.replace(tzinfo=dateutil.tz.tzlocal())
|
||||
except (ValueError, TypeError):
|
||||
message = 'Unable to parse scheduled timestamp ({0}). It should be in an unambiguous format (e.g. ISO 8601)'
|
||||
raise PatroniCtlException(message.format(scheduled))
|
||||
return scheduled_at
|
||||
|
||||
return None
|
||||
|
||||
|
||||
@ctl.command('reload', help='Reload cluster member configuration')
|
||||
@click.argument('cluster_name')
|
||||
@click.argument('member_names', nargs=-1)
|
||||
@@ -1061,16 +1024,20 @@ def restart(obj: Dict[str, Any], cluster_name: str, group: Optional[int], member
|
||||
* *version* could not be parsed; or
|
||||
* a restart is attempted against a cluster that is in maintenance mode.
|
||||
"""
|
||||
action = 'restart'
|
||||
cluster = get_dcs(obj, cluster_name, group).get_cluster()
|
||||
|
||||
members = get_members(obj, cluster, cluster_name, member_names, role, force, 'restart', False, group=group)
|
||||
members = get_members(obj, cluster, cluster_name, member_names, role, force, action, False, group=group)
|
||||
if scheduled is None and not force:
|
||||
next_hour = (datetime.datetime.now() + datetime.timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M')
|
||||
next_hour = (datetime.datetime.now() + datetime.timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M+00')
|
||||
scheduled = click.prompt('When should the restart take place (e.g. ' + next_hour + ') ',
|
||||
type=str, default='now')
|
||||
scheduled = scheduled if scheduled != 'now' else None
|
||||
|
||||
scheduled_at = parse_scheduled(scheduled)
|
||||
confirm_members_action(members, force, 'restart', scheduled_at)
|
||||
parse_result, scheduled_at = parse_schedule(scheduled)
|
||||
if parse_result:
|
||||
raise PatroniCtlException(parse_result.value[0].format(action=action))
|
||||
confirm_members_action(members, force, action, scheduled_at)
|
||||
|
||||
if p_any:
|
||||
random.shuffle(members)
|
||||
@@ -1212,6 +1179,9 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
|
||||
click.echo('Current cluster topology')
|
||||
output_members(obj, cluster, cluster_name, group=group)
|
||||
|
||||
# Define everything missing via interactive input or available cluster info (if force mode)
|
||||
|
||||
# Require Citus group
|
||||
if obj.get('citus') and group is None:
|
||||
if force:
|
||||
raise PatroniCtlException('For Citus clusters the --group must me specified')
|
||||
@@ -1222,42 +1192,25 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
|
||||
|
||||
global_config = get_global_config(cluster)
|
||||
|
||||
# leader has to be be defined for switchover only
|
||||
if action == 'switchover':
|
||||
# Leader is required for switchover only
|
||||
if action == 'switchover' and leader is None:
|
||||
if cluster.leader is None or not cluster.leader.name:
|
||||
raise PatroniCtlException('This cluster has no leader')
|
||||
|
||||
if leader is None:
|
||||
if force:
|
||||
leader = cluster.leader.name
|
||||
else:
|
||||
prompt = 'Standby Leader' if global_config.is_standby_cluster else 'Primary'
|
||||
leader = click.prompt(prompt, type=str, default=(cluster.leader and cluster.leader.name))
|
||||
|
||||
if cluster.leader.name != leader:
|
||||
raise PatroniCtlException(f'Member {leader} is not the leader of cluster {cluster_name}')
|
||||
|
||||
# excluding members with nofailover tag
|
||||
candidate_names = [str(m.name) for m in cluster.members if m.name != leader and not m.nofailover]
|
||||
# We sort the names for consistent output to the client
|
||||
candidate_names.sort()
|
||||
|
||||
if not candidate_names:
|
||||
raise PatroniCtlException('No candidates found to {0} to'.format(action))
|
||||
if force:
|
||||
leader = cluster.leader.name
|
||||
else:
|
||||
prompt = 'Standby Leader' if global_config.is_standby_cluster else 'Primary'
|
||||
leader = click.prompt(prompt, type=str, default=(cluster.leader and cluster.leader.name))
|
||||
|
||||
if candidate is None and not force:
|
||||
# Check if there are any candidates available at all
|
||||
candidate_names = [str(m.name) for m in cluster.members if m.name != leader and not m.nofailover]
|
||||
if not candidate_names:
|
||||
raise PatroniCtlException('No candidates found to {0} to'.format(action))
|
||||
candidate_names.sort() # we sort the names for consistent output to the client
|
||||
candidate = click.prompt('Candidate ' + str(candidate_names), type=str, default='')
|
||||
|
||||
if action == 'failover' and not candidate:
|
||||
raise PatroniCtlException('Failover could be performed only to a specific candidate')
|
||||
|
||||
if candidate == leader:
|
||||
raise PatroniCtlException(action.title() + ' target and source are the same.')
|
||||
|
||||
if candidate and candidate not in candidate_names:
|
||||
raise PatroniCtlException(
|
||||
f'Member {candidate} does not exist in cluster {cluster_name} or is tagged as nofailover')
|
||||
|
||||
# We allow manual failover to an aync node in the sync mode, so we better ask for the confirmation
|
||||
if all((not force,
|
||||
action == 'failover',
|
||||
global_config.is_synchronous_mode,
|
||||
@@ -1266,30 +1219,32 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
|
||||
if click.confirm(f'Are you sure you want to failover to the asynchronous node {candidate}'):
|
||||
raise PatroniCtlException('Aborting ' + action)
|
||||
|
||||
if action == 'switchover' and scheduled is None and not force:
|
||||
next_hour = (datetime.datetime.now() + datetime.timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M+00')
|
||||
scheduled = click.prompt('When should the switchover take place (e.g. ' + next_hour + ') ',
|
||||
type=str, default='now')
|
||||
scheduled = scheduled if scheduled != 'now' else None
|
||||
|
||||
# Now, when we collected all the possible info, run checks
|
||||
manual_failover = ManualFailover(action, cluster, leader, candidate, scheduled,
|
||||
global_config.is_paused, global_config.is_synchronous_mode)
|
||||
|
||||
result_text, _ = manual_failover.run_precheck().value
|
||||
if result_text:
|
||||
raise PatroniCtlException(result_text.format(action=action, leader=leader, candidate=candidate,
|
||||
cluster_name=cluster_name))
|
||||
|
||||
scheduled_at_str = None
|
||||
scheduled_at = None
|
||||
|
||||
if action == 'switchover':
|
||||
if scheduled is None and not force:
|
||||
next_hour = (datetime.datetime.now() + datetime.timedelta(hours=1)).strftime('%Y-%m-%dT%H:%M')
|
||||
scheduled = click.prompt('When should the switchover take place (e.g. ' + next_hour + ' ) ',
|
||||
type=str, default='now')
|
||||
|
||||
scheduled_at = parse_scheduled(scheduled)
|
||||
parse_result, scheduled_at = manual_failover.parse_scheduled()
|
||||
if parse_result:
|
||||
raise PatroniCtlException(parse_result.value[0].format(action=action))
|
||||
if scheduled_at:
|
||||
if global_config.is_paused:
|
||||
raise PatroniCtlException("Can't schedule switchover in the paused state")
|
||||
scheduled_at_str = scheduled_at.isoformat()
|
||||
|
||||
failover_value = {'candidate': candidate}
|
||||
if action == 'switchover':
|
||||
failover_value['leader'] = leader
|
||||
if scheduled_at_str:
|
||||
failover_value['scheduled_at'] = scheduled_at_str
|
||||
|
||||
logging.debug(failover_value)
|
||||
|
||||
# By now we have established that the leader exists and the candidate exists
|
||||
# By now we have established that the leader exists and the candidate exists,
|
||||
# so confirm the action that is about to be run
|
||||
if not force:
|
||||
demote_msg = f', demoting current leader {cluster.leader.name}' if cluster.leader else ''
|
||||
if scheduled_at_str:
|
||||
@@ -1302,6 +1257,15 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s
|
||||
if not click.confirm(f'Are you sure you want to {action} cluster {cluster_name}{demote_msg}?'):
|
||||
raise PatroniCtlException('Aborting ' + action)
|
||||
|
||||
# And finally the actual work
|
||||
failover_value = {'candidate': candidate}
|
||||
if action == 'switchover':
|
||||
failover_value['leader'] = leader
|
||||
if scheduled_at_str:
|
||||
failover_value['scheduled_at'] = scheduled_at_str
|
||||
|
||||
logging.debug(failover_value)
|
||||
|
||||
r = None
|
||||
try:
|
||||
member = cluster.leader.member if cluster.leader else candidate and cluster.get_member(candidate, False)
|
||||
|
||||
+22
-15
@@ -922,6 +922,27 @@ class Ha(object):
|
||||
lag = (self.cluster.last_lsn or 0) - wal_position
|
||||
return lag > self.global_config.maximum_lag_on_failover
|
||||
|
||||
def has_members_eligible_to_promote(self, members: List[Member], reference_lsn: int = 0,
|
||||
fast_path: bool = False) -> bool:
|
||||
ret = False
|
||||
cluster_timeline = self.cluster.timeline
|
||||
|
||||
for st in self.fetch_nodes_statuses(members):
|
||||
not_allowed_reason = st.failover_limitation()
|
||||
if not_allowed_reason:
|
||||
logger.info('Member %s is %s', st.member.name, not_allowed_reason)
|
||||
elif fast_path:
|
||||
return True
|
||||
elif reference_lsn and st.wal_position < reference_lsn or \
|
||||
not reference_lsn and self.is_lagging(st.wal_position):
|
||||
logger.info('Member %s exceeds maximum replication lag', st.member.name)
|
||||
elif self.check_timeline() and (not st.timeline or st.timeline < cluster_timeline):
|
||||
logger.info('Timeline %s of member %s is behind the cluster timeline %s',
|
||||
st.timeline, st.member.name, cluster_timeline)
|
||||
else:
|
||||
ret = True
|
||||
return ret
|
||||
|
||||
def _is_healthiest_node(self, members: Collection[Member], check_replication_lag: bool = True) -> bool:
|
||||
"""This method tries to determine whether I am healthy enough to became a new leader candidate or not."""
|
||||
|
||||
@@ -974,21 +995,7 @@ class Ha(object):
|
||||
elif not candidates:
|
||||
logger.warning('%s: candidates list is empty', action)
|
||||
|
||||
ret = False
|
||||
cluster_timeline = self.cluster.timeline
|
||||
for st in self.fetch_nodes_statuses(candidates):
|
||||
not_allowed_reason = st.failover_limitation()
|
||||
if not_allowed_reason:
|
||||
logger.info('Member %s is %s', st.member.name, not_allowed_reason)
|
||||
elif cluster_lsn and st.wal_position < cluster_lsn or \
|
||||
not cluster_lsn and self.is_lagging(st.wal_position):
|
||||
logger.info('Member %s exceeds maximum replication lag', st.member.name)
|
||||
elif self.check_timeline() and (not st.timeline or st.timeline < cluster_timeline):
|
||||
logger.info('Timeline %s of member %s is behind the cluster timeline %s',
|
||||
st.timeline, st.member.name, cluster_timeline)
|
||||
else:
|
||||
ret = True
|
||||
return ret
|
||||
return self.has_members_eligible_to_promote(candidates, cluster_lsn)
|
||||
|
||||
def manual_failover_process_no_leader(self) -> Optional[bool]:
|
||||
"""Handles manual failover/switchover when the old leader already stepped down.
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
from enum import Enum
|
||||
from typing import Optional, Tuple, TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
import datetime
|
||||
|
||||
from .dcs import Cluster
|
||||
from .ha import Patroni
|
||||
from .utils import ParseScheduleErrors
|
||||
|
||||
from .utils import parse_schedule
|
||||
|
||||
|
||||
class ManualFailoverPrecheckStatus(Enum):
|
||||
FAILOVER_NO_CANDIDATE = ('Failover could be performed only to a specific candidate', 400)
|
||||
SWITCHOVER_NO_LEADER = ('Switchover could be performed only from a specific leader', 400)
|
||||
SCHEDULED_FAILOVER = ("Failover can't be scheduled", 400)
|
||||
SCHEDULED_SWITCHOVER_PAUSE = ("Can't schedule switchover in the paused state", 400)
|
||||
SWITCHOVER_PAUSE_NO_CANDIDATE = ('Switchover is possible only to a specific candidate in a paused state', 400)
|
||||
SWITCHOVER_TO_LEADER = ('Switchover target and source are the same', 400)
|
||||
|
||||
CLUSTER_NO_LEADER = ('Cluster {cluster_name} has no leader', 412)
|
||||
LEADER_NOT_MEMBER = ('Member {leader} is not the leader of cluster {cluster_name}', 412)
|
||||
CANDIDATE_NOT_SYNC_STANDBY = ('candidate name does not match with sync_standby', 412)
|
||||
NO_SYNC_CANDIDATE = ('{action} is not possible: can not find sync_standby', 412)
|
||||
ONLY_LEADER = ('{action} is not possible: cluster does not have members except leader', 412)
|
||||
CANDIDATE_NOT_MEMEBER = ('Member {candidate} does not exist in cluster {cluster_name} or is tagged as nofailover',
|
||||
412)
|
||||
NO_GOOD_CANDIDATES = ('{action} is not possible: no good candidates have been found', 412)
|
||||
|
||||
CHECK_PASSED = ('', None)
|
||||
|
||||
|
||||
class ManualFailover(object):
|
||||
|
||||
def __init__(self, action: str, cluster: 'Cluster',
|
||||
leader: Optional[str], candidate: Optional[str], scheduled: Optional[str],
|
||||
paused: bool = False, sync_mode: bool = False, patroni_obj: Optional['Patroni'] = None) -> None:
|
||||
self.action = action
|
||||
self.cluster = cluster
|
||||
self.leader = leader
|
||||
self.candidate = candidate
|
||||
self.scheduled = scheduled
|
||||
self.paused = paused
|
||||
self.sync_mode = sync_mode
|
||||
self.patroni = patroni_obj
|
||||
|
||||
def parse_scheduled(self) -> Tuple[Optional['ParseScheduleErrors'], Optional['datetime.datetime']]:
|
||||
return parse_schedule(self.scheduled)
|
||||
|
||||
def run_precheck(self) -> ManualFailoverPrecheckStatus:
|
||||
if self.action == 'failover' and not self.candidate:
|
||||
return ManualFailoverPrecheckStatus.FAILOVER_NO_CANDIDATE
|
||||
elif self.action == 'switchover' and not self.leader:
|
||||
return ManualFailoverPrecheckStatus.SWITCHOVER_NO_LEADER
|
||||
|
||||
if self.scheduled:
|
||||
if self.action == 'failover':
|
||||
return ManualFailoverPrecheckStatus.SCHEDULED_FAILOVER
|
||||
elif self.paused:
|
||||
return ManualFailoverPrecheckStatus.SCHEDULED_SWITCHOVER_PAUSE
|
||||
|
||||
if self.paused and not self.candidate:
|
||||
return ManualFailoverPrecheckStatus.SWITCHOVER_PAUSE_NO_CANDIDATE
|
||||
|
||||
if self.leader == self.candidate:
|
||||
return ManualFailoverPrecheckStatus.SWITCHOVER_TO_LEADER
|
||||
|
||||
if self.action == 'switchover':
|
||||
if self.cluster.leader is None or not self.cluster.leader.name:
|
||||
return ManualFailoverPrecheckStatus.CLUSTER_NO_LEADER
|
||||
if self.cluster.leader.name != self.leader:
|
||||
return ManualFailoverPrecheckStatus.LEADER_NOT_MEMBER
|
||||
|
||||
if self.candidate:
|
||||
if self.action == 'switchover' and self.sync_mode and not self.cluster.sync.matches(self.candidate):
|
||||
return ManualFailoverPrecheckStatus.CANDIDATE_NOT_SYNC_STANDBY
|
||||
members = [m for m in self.cluster.members if m.name == self.candidate]
|
||||
if not members:
|
||||
return ManualFailoverPrecheckStatus.CANDIDATE_NOT_MEMEBER
|
||||
elif self.sync_mode:
|
||||
members = [m for m in self.cluster.members if self.cluster.sync.matches(m.name)]
|
||||
if not members:
|
||||
return ManualFailoverPrecheckStatus.NO_SYNC_CANDIDATE
|
||||
else:
|
||||
members = [m for m in self.cluster.members if not self.cluster.leader or m.name != self.cluster.leader.name and m.api_url]
|
||||
if not members:
|
||||
return ManualFailoverPrecheckStatus.ONLY_LEADER
|
||||
|
||||
if self.patroni and not self.patroni.ha.has_members_eligible_to_promote(members, fast_path=True):
|
||||
return ManualFailoverPrecheckStatus.NO_GOOD_CANDIDATES
|
||||
|
||||
return ManualFailoverPrecheckStatus.CHECK_PASSED
|
||||
@@ -9,6 +9,8 @@
|
||||
:var DBL_RE: regular expression to match double precision numbers, signed or unsigned. Matches scientific notation too.
|
||||
:var WHITESPACE_RE: regular expression to match whitespace characters
|
||||
"""
|
||||
import datetime
|
||||
import dateutil.parser
|
||||
import errno
|
||||
import logging
|
||||
import os
|
||||
@@ -20,6 +22,7 @@ import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from enum import Enum
|
||||
from shlex import split
|
||||
|
||||
from typing import Any, Callable, Dict, Iterator, List, Optional, Union, Tuple, Type, TYPE_CHECKING
|
||||
@@ -1061,3 +1064,23 @@ def get_major_version(bin_dir: Optional[str] = None, bin_name: str = 'postgres')
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
assert version is not None
|
||||
return '.'.join([version.group(1), version.group(3)]) if int(version.group(1)) < 10 else version.group(1)
|
||||
|
||||
|
||||
class ParseScheduleErrors(Enum):
|
||||
NO_TIMEZONE = ('Timezone information is mandatory for the scheduled {action}', 400)
|
||||
SCHEDULED_IN_PAST = ('Cannot schedule {action} in the past', 422)
|
||||
PARSING_ERROR = ('Unable to parse scheduled timestamp. It should be in an unambiguous format, e.g. ISO 8601', 422)
|
||||
|
||||
|
||||
def parse_schedule(schedule: Optional[str]) -> Tuple[Optional[ParseScheduleErrors], Optional[datetime.datetime]]:
|
||||
scheduled_at = None
|
||||
if schedule is not None:
|
||||
try:
|
||||
scheduled_at = dateutil.parser.parse(schedule)
|
||||
if scheduled_at.tzinfo is None:
|
||||
return ParseScheduleErrors.NO_TIMEZONE, scheduled_at
|
||||
elif scheduled_at < datetime.datetime.now(tzutc):
|
||||
return ParseScheduleErrors.SCHEDULED_IN_PAST, scheduled_at
|
||||
except (ValueError, TypeError):
|
||||
return ParseScheduleErrors.PARSING_ERROR, scheduled_at
|
||||
return None, scheduled_at
|
||||
|
||||
+62
-34
@@ -13,11 +13,12 @@ from patroni.config import GlobalConfig
|
||||
from patroni.dcs import ClusterConfig, Member
|
||||
from patroni.exceptions import PostgresConnectionException
|
||||
from patroni.ha import _MemberStatus
|
||||
from patroni.manual_failover import ManualFailoverPrecheckStatus
|
||||
from patroni.psycopg import OperationalError
|
||||
from patroni.utils import RetryFailedError, tzutc
|
||||
from patroni.utils import ParseScheduleErrors, RetryFailedError, tzutc
|
||||
|
||||
from . import MockConnect, psycopg_connect
|
||||
from .test_ha import get_cluster_initialized_without_leader
|
||||
from .test_ha import get_cluster_initialized_without_leader, get_cluster_initialized_with_leader
|
||||
|
||||
|
||||
future_restart_time = datetime.datetime.now(tzutc) + datetime.timedelta(days=5)
|
||||
@@ -140,6 +141,9 @@ class MockHa(object):
|
||||
def is_paused():
|
||||
return True
|
||||
|
||||
def has_members_eligible_to_promote(*args, **kwargs):
|
||||
return True
|
||||
|
||||
|
||||
class MockLogger(object):
|
||||
|
||||
@@ -519,7 +523,7 @@ class TestRestApiHandler(unittest.TestCase):
|
||||
# Invalid content
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
MockRestApiServer(RestApiHandler, post + '7\n\n{"1":2}')
|
||||
response_mock.assert_called_with(400, 'Switchover could be performed only from a specific leader')
|
||||
response_mock.assert_called_with(*ManualFailoverPrecheckStatus.SWITCHOVER_NO_LEADER.value[::-1])
|
||||
|
||||
# Empty content
|
||||
request = post + '0\n\n'
|
||||
@@ -527,25 +531,19 @@ class TestRestApiHandler(unittest.TestCase):
|
||||
|
||||
# [Switchover without a candidate]
|
||||
|
||||
# Cluster with only a leader
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
cluster.leader.name = 'postgresql1'
|
||||
request = post + '25\n\n{"leader": "postgresql1"}'
|
||||
MockRestApiServer(RestApiHandler, request)
|
||||
response_mock.assert_called_with(
|
||||
412, 'switchover is not possible: cluster does not have members except leader')
|
||||
cluster.leader.name = 'postgresql1'
|
||||
request = post + '25\n\n{"leader": "postgresql1"}'
|
||||
|
||||
# Switchover in pause mode
|
||||
# No candidate in pause mode
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock, \
|
||||
patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)):
|
||||
MockRestApiServer(RestApiHandler, request)
|
||||
response_mock.assert_called_with(
|
||||
400, 'Switchover is possible only to a specific candidate in a paused state')
|
||||
response_mock.assert_called_with(*ManualFailoverPrecheckStatus.SWITCHOVER_PAUSE_NO_CANDIDATE.value[::-1])
|
||||
|
||||
# No healthy nodes to promote in both sync and async mode
|
||||
for is_synchronous_mode, response in (
|
||||
(True, 'switchover is not possible: can not find sync_standby'),
|
||||
(False, 'switchover is not possible: cluster does not have members except leader')):
|
||||
(True, ManualFailoverPrecheckStatus.NO_SYNC_CANDIDATE.value[0].format(action='switchover')),
|
||||
(False, ManualFailoverPrecheckStatus.ONLY_LEADER.value[0].format(action='switchover'))):
|
||||
with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=is_synchronous_mode)), \
|
||||
patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
MockRestApiServer(RestApiHandler, request)
|
||||
@@ -557,20 +555,25 @@ class TestRestApiHandler(unittest.TestCase):
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
request = post + '53\n\n{"leader": "postgresql2", "candidate": "postgresql2"}'
|
||||
MockRestApiServer(RestApiHandler, request)
|
||||
response_mock.assert_called_with(400, 'Switchover target and source are the same')
|
||||
response_mock.assert_called_with(*ManualFailoverPrecheckStatus.SWITCHOVER_TO_LEADER.value[::-1])
|
||||
|
||||
# Current leader is different from the one specified
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
cluster.leader.name = 'postgresql2'
|
||||
request = post + '53\n\n{"leader": "postgresql1", "candidate": "postgresql2"}'
|
||||
MockRestApiServer(RestApiHandler, request)
|
||||
response_mock.assert_called_with(412, 'leader name does not match')
|
||||
response_mock.assert_called_with(
|
||||
ManualFailoverPrecheckStatus.LEADER_NOT_MEMBER.value[1],
|
||||
ManualFailoverPrecheckStatus.LEADER_NOT_MEMBER.value[0].format(leader='postgresql1',
|
||||
cluster_name='dummy'))
|
||||
|
||||
# Candidate to promote is not a member of the cluster
|
||||
# Candidate to promote is not a sync standby/a member of the cluster
|
||||
cluster.leader.name = 'postgresql1'
|
||||
cluster.sync.matches.return_value = False
|
||||
for is_synchronous_mode, response in (
|
||||
(True, 'candidate name does not match with sync_standby'), (False, 'candidate does not exists')):
|
||||
(True, ManualFailoverPrecheckStatus.CANDIDATE_NOT_SYNC_STANDBY.value[0]),
|
||||
(False, ManualFailoverPrecheckStatus.CANDIDATE_NOT_MEMEBER.value[0].format(candidate="postgresql2",
|
||||
cluster_name='dummy'))):
|
||||
with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=is_synchronous_mode)), \
|
||||
patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
MockRestApiServer(RestApiHandler, request)
|
||||
@@ -579,9 +582,21 @@ class TestRestApiHandler(unittest.TestCase):
|
||||
cluster.members = [Member(0, 'postgresql0', 30, {'api_url': 'http'}),
|
||||
Member(0, 'postgresql2', 30, {'api_url': 'http'})]
|
||||
|
||||
# Cluster has no leader
|
||||
cluster.leader.name = None
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
request = post + '53\n\n{"leader": "postgresql1"}'
|
||||
MockRestApiServer(RestApiHandler, request)
|
||||
response_mock.assert_called_with(
|
||||
ManualFailoverPrecheckStatus.CLUSTER_NO_LEADER.value[1],
|
||||
ManualFailoverPrecheckStatus.CLUSTER_NO_LEADER.value[0].format(leader='leader', cluster_name='dummy'))
|
||||
|
||||
cluster.leader.name = 'postgresql1'
|
||||
|
||||
# Failover key is empty in DCS
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
cluster.failover = None
|
||||
request = post + '53\n\n{"leader": "postgresql1", "candidate": "postgresql2"}'
|
||||
MockRestApiServer(RestApiHandler, request)
|
||||
response_mock.assert_called_with(503, 'Switchover failed')
|
||||
|
||||
@@ -616,10 +631,12 @@ class TestRestApiHandler(unittest.TestCase):
|
||||
dcs.manual_failover.return_value = True
|
||||
|
||||
# Candidate is not healthy to be promoted
|
||||
with patch.object(MockHa, 'fetch_nodes_statuses', Mock(return_value=[])), \
|
||||
with patch.object(MockHa, 'has_members_eligible_to_promote', Mock(return_value=False)), \
|
||||
patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
MockRestApiServer(RestApiHandler, request)
|
||||
response_mock.assert_called_with(412, 'switchover is not possible: no good candidates have been found')
|
||||
response_mock.assert_called_with(
|
||||
ManualFailoverPrecheckStatus.NO_GOOD_CANDIDATES.value[1],
|
||||
ManualFailoverPrecheckStatus.NO_GOOD_CANDIDATES.value[0].format(action='switchover'))
|
||||
|
||||
# [Scheduled switchover]
|
||||
|
||||
@@ -630,47 +647,58 @@ class TestRestApiHandler(unittest.TestCase):
|
||||
MockRestApiServer(RestApiHandler, request)
|
||||
response_mock.assert_called_with(202, 'Switchover scheduled')
|
||||
|
||||
# Schedule in paused mode
|
||||
# Scheduled in pause mode
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock, \
|
||||
patch.object(GlobalConfig, 'is_paused', PropertyMock(return_value=True)):
|
||||
dcs.manual_failover.return_value = False
|
||||
MockRestApiServer(RestApiHandler, request)
|
||||
response_mock.assert_called_with(400, "Can't schedule switchover in the paused state")
|
||||
response_mock.assert_called_with(*ManualFailoverPrecheckStatus.SCHEDULED_SWITCHOVER_PAUSE.value[::-1])
|
||||
|
||||
# No timezone specified
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
request = post + '97\n\n{"leader": "postgresql1", "member": "postgresql2",' + \
|
||||
' "scheduled_at": "6016-02-15T18:13:30.568224"}'
|
||||
MockRestApiServer(RestApiHandler, request)
|
||||
response_mock.assert_called_with(400, 'Timezone information is mandatory for the scheduled switchover')
|
||||
response_mock.assert_called_with(
|
||||
ParseScheduleErrors.NO_TIMEZONE.value[1],
|
||||
ParseScheduleErrors.NO_TIMEZONE.value[0].format(action='switchover'))
|
||||
|
||||
request = post + '103\n\n{"leader": "postgresql1", "member": "postgresql2", "scheduled_at": "'
|
||||
|
||||
# Scheduled in the past
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
MockRestApiServer(RestApiHandler, request + '1016-02-15T18:13:30.568224+01:00"}')
|
||||
response_mock.assert_called_with(422, 'Cannot schedule switchover in the past')
|
||||
response_mock.assert_called_with(
|
||||
ParseScheduleErrors.SCHEDULED_IN_PAST.value[1],
|
||||
ParseScheduleErrors.SCHEDULED_IN_PAST.value[0].format(action='switchover'))
|
||||
|
||||
# Invalid date
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
MockRestApiServer(RestApiHandler, request + '2010-02-29T18:13:30.568224+01:00"}')
|
||||
response_mock.assert_called_with(
|
||||
422, 'Unable to parse scheduled timestamp. It should be in an unambiguous format, e.g. ISO 8601')
|
||||
response_mock.assert_called_with(*ParseScheduleErrors.PARSING_ERROR.value[::-1])
|
||||
|
||||
def test_do_POST_failover(self):
|
||||
@patch.object(MockPatroni, 'dcs')
|
||||
def test_do_POST_failover(self, mock_dcs):
|
||||
post = 'POST /failover HTTP/1.0' + self._authorization + '\nContent-Length: '
|
||||
cluster = mock_dcs.get_cluster.return_value
|
||||
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
MockRestApiServer(RestApiHandler, post + '14\n\n{"leader":"1"}')
|
||||
response_mock.assert_called_once_with(400, 'Failover could be performed only to a specific candidate')
|
||||
MockRestApiServer(RestApiHandler, post + '19\n\n{"leader":"leader"}')
|
||||
response_mock.assert_called_once_with(*ManualFailoverPrecheckStatus.FAILOVER_NO_CANDIDATE.value[::-1])
|
||||
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
MockRestApiServer(RestApiHandler, post + '37\n\n{"candidate":"2","scheduled_at": "1"}')
|
||||
response_mock.assert_called_once_with(400, "Failover can't be scheduled")
|
||||
response_mock.assert_called_once_with(*ManualFailoverPrecheckStatus.SCHEDULED_FAILOVER.value[::-1])
|
||||
|
||||
with patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
MockRestApiServer(RestApiHandler, post + '30\n\n{"leader":"1","candidate":"2"}')
|
||||
response_mock.assert_called_once_with(412, 'leader name does not match')
|
||||
# Candidate is not healthy to be promoted
|
||||
cluster.members = [Member(0, 'postgresql0', 30, {'api_url': 'http'}),
|
||||
Member(0, 'postgresql2', 30, {'api_url': 'http'})]
|
||||
with patch.object(MockHa, 'has_members_eligible_to_promote', Mock(return_value=False)), \
|
||||
patch.object(RestApiHandler, 'write_response') as response_mock:
|
||||
MockRestApiServer(RestApiHandler, post + '27\n\n{"candidate":"postgresql2"}')
|
||||
response_mock.assert_called_with(
|
||||
ManualFailoverPrecheckStatus.NO_GOOD_CANDIDATES.value[1],
|
||||
ManualFailoverPrecheckStatus.NO_GOOD_CANDIDATES.value[0].format(action='failover'))
|
||||
|
||||
@patch.object(MockHa, 'is_leader', Mock(return_value=True))
|
||||
def test_do_POST_citus(self):
|
||||
|
||||
+200
-113
@@ -6,12 +6,14 @@ import unittest
|
||||
from click.testing import CliRunner
|
||||
from datetime import datetime, timedelta
|
||||
from mock import patch, Mock, PropertyMock
|
||||
from patroni.config import GlobalConfig
|
||||
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
|
||||
from patroni.dcs.etcd import AbstractEtcdClientWithFailover, Cluster, Failover
|
||||
from patroni.manual_failover import ManualFailoverPrecheckStatus
|
||||
from patroni.psycopg import OperationalError
|
||||
from patroni.utils import tzutc
|
||||
from patroni.utils import ParseScheduleErrors, tzutc
|
||||
from prettytable import PrettyTable, ALL
|
||||
from urllib3 import PoolManager
|
||||
|
||||
@@ -35,6 +37,10 @@ DEFAULT_CONFIG = {
|
||||
class TestCtl(unittest.TestCase):
|
||||
TEST_ROLES = ('master', 'primary', 'leader')
|
||||
|
||||
SCHEDULED_TS = '2055-01-01T12:00:00+01:00'
|
||||
SCHEDULED_TS_NO_TZ = '2055-01-01T12:00:00'
|
||||
SCHEDULED_TS_INVALID = '2055-02-30T12:00:00'
|
||||
|
||||
@patch('socket.getaddrinfo', socket_getaddrinfo)
|
||||
@patch.object(AbstractEtcdClientWithFailover, '_get_machines_list', Mock(return_value=['http://remotehost:2379']))
|
||||
def setUp(self):
|
||||
@@ -116,70 +122,6 @@ class TestCtl(unittest.TestCase):
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force'])
|
||||
self.assertEqual(result.exit_code, 0)
|
||||
|
||||
# Scheduled (confirm)
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'],
|
||||
input='leader\nother\n2300-01-01T12:23:00\ny')
|
||||
self.assertEqual(result.exit_code, 0)
|
||||
|
||||
# Scheduled (abort)
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
|
||||
'--scheduled', '2015-01-01T12:00:00+01:00'], input='leader\nother\n\nN')
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
|
||||
# Scheduled with --force option
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
|
||||
'--force', '--scheduled', '2015-01-01T12:00:00+01:00'])
|
||||
self.assertEqual(result.exit_code, 0)
|
||||
|
||||
# Scheduled in pause mode
|
||||
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'])
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn("Can't schedule switchover in the paused state", result.output)
|
||||
|
||||
# Target and source are equal
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nleader\n\ny')
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn('Switchover target and source are the same', result.output)
|
||||
|
||||
# Candidate is not a member of the cluster
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nReality\n\ny')
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn('Member Reality does not exist in cluster dummy or is tagged as nofailover', result.output)
|
||||
|
||||
# Invalid timestamp
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force', '--scheduled', 'invalid'])
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn('Unable to parse scheduled timestamp', result.output)
|
||||
|
||||
# Invalid timestamp
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
|
||||
'--force', '--scheduled', '2115-02-30T12:00:00+01:00'])
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn('Unable to parse scheduled timestamp', result.output)
|
||||
|
||||
# Specifying wrong leader
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='dummy')
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn('Member dummy is not the leader of cluster dummy', result.output)
|
||||
|
||||
# Errors while sending Patroni REST API request
|
||||
with patch.object(PoolManager, 'request', Mock(side_effect=Exception)):
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'],
|
||||
input='leader\nother\n2300-01-01T12:23:00\ny')
|
||||
self.assertIn('falling back to DCS', result.output)
|
||||
|
||||
with patch.object(PoolManager, 'request') as mock_api_request:
|
||||
mock_api_request.return_value.status = 500
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
|
||||
self.assertIn('Switchover failed', result.output)
|
||||
|
||||
mock_api_request.return_value.status = 501
|
||||
mock_api_request.return_value.data = b'Server does not support this operation'
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
|
||||
self.assertIn('Switchover failed', result.output)
|
||||
|
||||
# No members available
|
||||
mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_only_leader
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
|
||||
@@ -197,6 +139,132 @@ class TestCtl(unittest.TestCase):
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn('For Citus clusters the --group must me specified', result.output)
|
||||
|
||||
# [Scheduled]
|
||||
|
||||
mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
|
||||
|
||||
# Scheduled (confirm)
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'],
|
||||
input=f'leader\nother\n{self.SCHEDULED_TS}\ny')
|
||||
self.assertEqual(result.exit_code, 0)
|
||||
self.assertIn(f'Are you sure you want to schedule switchover of cluster dummy '
|
||||
f'at {self.SCHEDULED_TS}, demoting current leader', result.output)
|
||||
|
||||
# Scheduled (abort)
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
|
||||
'--scheduled', self.SCHEDULED_TS], input='leader\nother\n\nN')
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
|
||||
# Scheduled with --force option
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
|
||||
'--force', '--scheduled', self.SCHEDULED_TS])
|
||||
self.assertEqual(result.exit_code, 0)
|
||||
|
||||
# Scheduled in pause mode
|
||||
with patch('patroni.config.GlobalConfig.is_paused', PropertyMock(return_value=True)):
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
|
||||
'--force', '--scheduled', self.SCHEDULED_TS])
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn(ManualFailoverPrecheckStatus.SCHEDULED_SWITCHOVER_PAUSE.value[0], result.output)
|
||||
|
||||
# Invalid timestamp with force
|
||||
result = self.runner.invoke(ctl,['switchover', 'dummy', '--group', '0', '--force', '--scheduled',
|
||||
self.SCHEDULED_TS_INVALID])
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn('Unable to parse scheduled timestamp', result.output)
|
||||
|
||||
# Invalid timestamp
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
|
||||
'--force', '--scheduled', self.SCHEDULED_TS_INVALID])
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn('Unable to parse scheduled timestamp', result.output)
|
||||
|
||||
# Invalid timestamp - no timezone
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0',
|
||||
'--force', '--scheduled', self.SCHEDULED_TS_NO_TZ])
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn(ParseScheduleErrors.NO_TIMEZONE.value[0].format(action='switchover'), result.output)
|
||||
|
||||
# [Other erroneous combinations]
|
||||
|
||||
# No candidate in pause mode
|
||||
with patch('patroni.config.GlobalConfig.is_paused', PropertyMock(return_value=True)):
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\n\n\ny')
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn(ManualFailoverPrecheckStatus.SWITCHOVER_PAUSE_NO_CANDIDATE.value[0], result.output)
|
||||
|
||||
# Target and source are equal
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nleader\n\ny')
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn(ManualFailoverPrecheckStatus.SWITCHOVER_TO_LEADER.value[0], result.output)
|
||||
|
||||
# Candidate is not a member of the cluster
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nReality\n\ny')
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn(ManualFailoverPrecheckStatus.CANDIDATE_NOT_MEMEBER.value[0].format(candidate='Reality',
|
||||
cluster_name='dummy'),
|
||||
result.output)
|
||||
|
||||
# Specifying wrong leader
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='dummy')
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn(
|
||||
ManualFailoverPrecheckStatus.LEADER_NOT_MEMBER.value[0].format(leader='dummy',
|
||||
cluster_name='dummy'),
|
||||
result.output)
|
||||
|
||||
mock_get_dcs.return_value.get_cluster = Mock(
|
||||
return_value=get_cluster_initialized_with_leader(sync=('leader', 'other')))
|
||||
|
||||
# Candidate is not a sync standby
|
||||
with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=True)):
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\notherMember\n\ny')
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn(ManualFailoverPrecheckStatus.CANDIDATE_NOT_SYNC_STANDBY.value[0], result.output)
|
||||
|
||||
# No healthy nodes to promote in sync mode
|
||||
mock_get_dcs.return_value.get_cluster = Mock(return_value=get_cluster_initialized_with_leader(sync=('leader')))
|
||||
with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=True)):
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force'])
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn(ManualFailoverPrecheckStatus.NO_SYNC_CANDIDATE.value[0].format(action='switchover'),
|
||||
result.output)
|
||||
|
||||
# No healthy nodes to promote in async mode
|
||||
mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_only_leader
|
||||
with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=False)):
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--force'])
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn(ManualFailoverPrecheckStatus.ONLY_LEADER.value[0].format(action='switchover'),
|
||||
result.output)
|
||||
|
||||
# Cluster has no leader
|
||||
mock_get_dcs.return_value.get_cluster = get_cluster_initialized_without_leader
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0', '--leader', 'leader', '--force'])
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn(
|
||||
ManualFailoverPrecheckStatus.CLUSTER_NO_LEADER.value[0].format(leader='leader', cluster_name='dummy'),
|
||||
result.output)
|
||||
|
||||
# [Errors while sending Patroni REST API request]
|
||||
|
||||
mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
|
||||
|
||||
with patch.object(PoolManager, 'request', Mock(side_effect=Exception)):
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'],
|
||||
input=f'leader\nother\n{self.SCHEDULED_TS}\ny')
|
||||
self.assertIn('falling back to DCS', result.output)
|
||||
|
||||
with patch.object(PoolManager, 'request') as mock_api_request:
|
||||
mock_api_request.return_value.status = 500
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
|
||||
self.assertIn('Switchover failed', result.output)
|
||||
|
||||
mock_api_request.return_value.status = 501
|
||||
mock_api_request.return_value.data = b'Server does not support this operation'
|
||||
result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny')
|
||||
self.assertIn('Switchover failed', result.output)
|
||||
|
||||
@patch('patroni.ctl.get_dcs')
|
||||
@patch.object(PoolManager, 'request', Mock(return_value=MockResponse()))
|
||||
@patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse()))
|
||||
@@ -206,8 +274,9 @@ class TestCtl(unittest.TestCase):
|
||||
# No candidate specified
|
||||
mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
|
||||
result = self.runner.invoke(ctl, ['failover', 'dummy'], input='0\n')
|
||||
self.assertIn('Failover could be performed only to a specific candidate', result.output)
|
||||
self.assertIn(ManualFailoverPrecheckStatus.FAILOVER_NO_CANDIDATE.value[0], result.output)
|
||||
|
||||
# Failover to an async member in sync mode (confirm)
|
||||
cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
|
||||
|
||||
# Temp test to check a fallback to switchover if leader is specified
|
||||
@@ -320,12 +389,9 @@ class TestCtl(unittest.TestCase):
|
||||
|
||||
@patch.object(PoolManager, 'request')
|
||||
@patch('patroni.ctl.get_dcs')
|
||||
def test_restart_reinit(self, mock_get_dcs, mock_post):
|
||||
def test_reinit(self, mock_get_dcs, mock_post):
|
||||
mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
|
||||
mock_post.return_value.status = 503
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha'], input='now\ny\n')
|
||||
assert 'Failed: restart for' in result.output
|
||||
assert result.exit_code == 0
|
||||
|
||||
result = self.runner.invoke(ctl, ['reinit', 'alpha'], input='y')
|
||||
assert result.exit_code == 1
|
||||
@@ -334,67 +400,88 @@ class TestCtl(unittest.TestCase):
|
||||
result = self.runner.invoke(ctl, ['reinit', 'alpha', 'other'], input='y\ny')
|
||||
assert result.exit_code == 0
|
||||
|
||||
# Aborted restart
|
||||
@patch.object(PoolManager, 'request')
|
||||
@patch('patroni.ctl.get_dcs')
|
||||
def test_restart(self, mock_get_dcs, mock_post):
|
||||
mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader
|
||||
mock_post.return_value.status = 200
|
||||
|
||||
# Successful restart
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha'], input='now\ny\n')
|
||||
self.assertEqual(result.exit_code, 0)
|
||||
|
||||
# Aborted
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha'], input='now\nN')
|
||||
assert result.exit_code == 1
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
|
||||
# With pending the flag
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha', '--pending', '--force'])
|
||||
assert result.exit_code == 0
|
||||
|
||||
# Aborted scheduled restart
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha', '--scheduled', '2019-10-01T14:30'], input='N')
|
||||
assert result.exit_code == 1
|
||||
self.assertEqual(result.exit_code, 0)
|
||||
|
||||
# Not a member
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha', 'dummy', '--any'], input='now\ny')
|
||||
assert result.exit_code == 1
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn('Not a single cluster member among provided members', result.output)
|
||||
|
||||
# Not a member with the specified role
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--role', 'primary'], input='now\ny')
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn('No primary among provided members', result.output)
|
||||
|
||||
# Wrong pg version
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha', '--any', '--pg-version', '9.1'], input='now\ny')
|
||||
assert 'Error: Invalid PostgreSQL version format' in result.output
|
||||
assert result.exit_code == 1
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn('Error: Invalid PostgreSQL version format', result.output)
|
||||
|
||||
# Restart with timeout
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha', '--pending', '--force', '--timeout', '10min'])
|
||||
assert result.exit_code == 0
|
||||
self.assertEqual(result.exit_code, 0)
|
||||
|
||||
# normal restart, the schedule is actually parsed, but not validated in patronictl
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30'])
|
||||
assert 'Failed: flush scheduled restart' in result.output
|
||||
# Scheduled restart
|
||||
|
||||
# Aborted scheduled restart
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha', '--scheduled', self.SCHEDULED_TS], input='N')
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
|
||||
# Error parsing scheduled flag value (no tz)
|
||||
result = self.runner.invoke(ctl,
|
||||
['restart', 'alpha', 'other', '--force', '--scheduled', self.SCHEDULED_TS_NO_TZ])
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn(ParseScheduleErrors.NO_TIMEZONE.value[0].format(action='restart'), result.output)
|
||||
|
||||
# Error parsing scheduled flag value (invalid date)
|
||||
result = self.runner.invoke(ctl,
|
||||
['restart', 'alpha', 'other', '--force', '--scheduled', self.SCHEDULED_TS_INVALID])
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn('Unable to parse scheduled timestamp', result.output)
|
||||
|
||||
# Successfully scheduled restart
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha', '--scheduled', self.SCHEDULED_TS], input='Y')
|
||||
self.assertEqual(result.exit_code, 0)
|
||||
self.assertIn('Success: restart on member other', result.output)
|
||||
|
||||
# Not possible to schedule in pause mode
|
||||
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
|
||||
['restart', 'alpha', 'other', '--force', '--scheduled', self.SCHEDULED_TS])
|
||||
self.assertEqual(result.exit_code, 1)
|
||||
self.assertIn("Can't schedule restart in the paused state", result.output)
|
||||
|
||||
# force restart with restart already present
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', '2300-10-01T14:30'])
|
||||
assert result.exit_code == 0
|
||||
|
||||
ctl_args = ['restart', 'alpha', '--pg-version', '99.0', '--scheduled', '2300-10-01T14:30']
|
||||
# normal restart, the schedule is actually parsed, but not validated in patronictl
|
||||
mock_post.return_value.status = 200
|
||||
result = self.runner.invoke(ctl, ctl_args, input='y')
|
||||
assert result.exit_code == 0
|
||||
# Force restart with restart already scheduled
|
||||
result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force', '--scheduled', self.SCHEDULED_TS])
|
||||
self.assertEqual(result.exit_code, 0)
|
||||
|
||||
# get restart with the non-200 return code
|
||||
# normal restart, the schedule is actually parsed, but not validated in patronictl
|
||||
mock_post.return_value.status = 204
|
||||
result = self.runner.invoke(ctl, ctl_args, input='y')
|
||||
assert result.exit_code == 0
|
||||
|
||||
# get restart with the non-200 return code
|
||||
# normal restart, the schedule is actually parsed, but not validated in patronictl
|
||||
mock_post.return_value.status = 202
|
||||
result = self.runner.invoke(ctl, ctl_args, input='y')
|
||||
assert 'Success: restart scheduled' in result.output
|
||||
assert result.exit_code == 0
|
||||
|
||||
# get restart with the non-200 return code
|
||||
# normal restart, the schedule is actually parsed, but not validated in patronictl
|
||||
mock_post.return_value.status = 409
|
||||
result = self.runner.invoke(ctl, ctl_args, input='y')
|
||||
assert 'Failed: another restart is already' in result.output
|
||||
assert result.exit_code == 0
|
||||
ctl_args = ['restart', 'alpha', '--pg-version', '99.0', '--scheduled', self.SCHEDULED_TS]
|
||||
for code, output in [
|
||||
(204, 'Failed: restart for member other, status code=204'),
|
||||
(202, 'Success: restart scheduled'),
|
||||
(409, 'Failed: another restart is already')
|
||||
]:
|
||||
mock_post.return_value.status = code
|
||||
result = self.runner.invoke(ctl, ctl_args, input='y')
|
||||
self.assertEqual(result.exit_code, 0)
|
||||
self.assertIn(output, result.output)
|
||||
|
||||
@patch('patroni.ctl.get_dcs')
|
||||
def test_remove(self, mock_get_dcs):
|
||||
|
||||
@@ -1630,3 +1630,13 @@ class TestHa(PostgresInit):
|
||||
self.assertEqual(self.ha.patroni.request.call_args[1]['timeout'], 2)
|
||||
mock_logger.assert_called()
|
||||
self.assertTrue(mock_logger.call_args[0][0].startswith('Request to Citus coordinator'))
|
||||
|
||||
def test_has_members_eligible_to_promote(self):
|
||||
self.ha.fetch_node_status = get_node_status()
|
||||
members = [
|
||||
Member(0, 'test', 1, {'api_url': 'http://127.0.0.1:8011/patroni', 'conn_url': 'postgres://127.0.0.1:5432/postgres'}),
|
||||
Member(0, 'test2', 1, {'api_url': 'http://127.0.0.1:8011/patroni', 'conn_url': 'postgres://127.0.0.1:5432/postgres'}),
|
||||
]
|
||||
with patch('patroni.ha.logger.info') as mock_logger:
|
||||
self.assertTrue(self.ha.has_members_eligible_to_promote(members, fast_path=True))
|
||||
mock_logger.assert_not_called()
|
||||
|
||||
Reference in New Issue
Block a user