diff --git a/patroni/ctl.py b/patroni/ctl.py index b4c57873..2d92457a 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -255,15 +255,23 @@ def load_config(path: str, dcs_url: Optional[str]) -> Dict[str, Any]: return config +def _get_configuration() -> Dict[str, Any]: + """Get configuration object. + + :returns: configuration object from the current context. + """ + return click.get_current_context().obj['__config'] + + option_format = click.option('--format', '-f', 'fmt', help='Output format', default='pretty', type=click.Choice(['pretty', 'tsv', 'json', 'yaml', 'yml'])) option_watchrefresh = click.option('-w', '--watch', type=float, help='Auto update the screen every X seconds') option_watch = click.option('-W', is_flag=True, help='Auto update the screen every 2 seconds') option_force = click.option('--force', is_flag=True, help='Do not ask for confirmation at any point') arg_cluster_name = click.argument('cluster_name', required=False, - default=lambda: click.get_current_context().obj.get('scope')) + default=lambda: _get_configuration().get('scope')) option_default_citus_group = click.option('--group', required=False, type=int, help='Citus group', - default=lambda: click.get_current_context().obj.get('citus', {}).get('group')) + default=lambda: _get_configuration().get('citus', {}).get('group')) option_citus_group = click.option('--group', required=False, type=int, help='Citus group') role_choice = click.Choice(['leader', 'primary', 'standby-leader', 'replica', 'standby', 'any', 'master']) @@ -301,15 +309,23 @@ def ctl(ctx: click.Context, config_file: str, dcs_url: Optional[str], insecure: level = os.environ.get(name, level) logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=level) logging.captureWarnings(True) # Capture eventual SSL warning - ctx.obj = load_config(config_file, dcs_url) + config = load_config(config_file, dcs_url) # backward compatibility for configuration file where ctl section is not defined - ctx.obj.setdefault('ctl', {})['insecure'] = ctx.obj.get('ctl', {}).get('insecure') or insecure + config.setdefault('ctl', {})['insecure'] = config.get('ctl', {}).get('insecure') or insecure + ctx.obj = {'__config': config} -def get_dcs(config: Dict[str, Any], scope: str, group: Optional[int]) -> AbstractDCS: +def is_citus_cluster() -> bool: + """Check if we are working with Citus cluster. + + :returns: ``True`` if configuration has ``citus`` section, otherwise ``False``. + """ + return bool(_get_configuration().get('citus')) + + +def get_dcs(scope: str, group: Optional[int]) -> AbstractDCS: """Get the DCS object. - :param config: Patroni configuration. :param scope: cluster name. :param group: if *group* is defined, use it to select which alternative Citus group this DCS refers to. If *group* is ``None`` and a Citus configuration exists, assume this is the coordinator. Coordinator has the group ``0``. @@ -320,13 +336,14 @@ def get_dcs(config: Dict[str, Any], scope: str, group: Optional[int]) -> Abstrac :raises: :class:`PatroniCtlException`: if not suitable DCS configuration could be found. """ + config = _get_configuration() config.update({'scope': scope, 'patronictl': True}) if group is not None: config['citus'] = {'group': group} config.setdefault('name', scope) try: dcs = _get_dcs(config) - if config.get('citus') and group is None: + if is_citus_cluster() and group is None: dcs.is_citus_coordinator = lambda: True return dcs except PatroniException as e: @@ -347,7 +364,7 @@ def request_patroni(member: Member, method: str = 'GET', ctx = click.get_current_context() # the current click context request_executor = ctx.obj.get('__request_patroni') if not request_executor: - request_executor = ctx.obj['__request_patroni'] = PatroniRequest(ctx.obj) + request_executor = ctx.obj['__request_patroni'] = PatroniRequest(_get_configuration()) return request_executor(member, method, endpoint, data) @@ -452,11 +469,9 @@ def watching(w: bool, watch: Optional[int], max_count: Optional[int] = None, cle yield 0 -def get_all_members(obj: Dict[str, Any], cluster: Cluster, - group: Optional[int], role: str = 'leader') -> Iterator[Member]: +def get_all_members(cluster: Cluster, group: Optional[int], role: str = 'leader') -> Iterator[Member]: """Get all cluster members that have the given *role*. - :param obj: the Patroni configuration. :param cluster: the Patroni cluster. :param group: filter which Citus group we should get members from. If ``None`` get from all groups. :param role: role to filter members. Can be one among: @@ -470,7 +485,7 @@ def get_all_members(obj: Dict[str, Any], cluster: Cluster, :yields: members that have the given *role*. """ clusters = {0: cluster} - if obj.get('citus') and group is None: + if is_citus_cluster() and group is None: clusters.update(cluster.workers) if role in ('leader', 'master', 'primary', 'standby-leader'): # In the DCS the members' role can be one among: ``primary``, ``master``, ``replica`` or ``standby_leader``. @@ -492,11 +507,10 @@ def get_all_members(obj: Dict[str, Any], cluster: Cluster, yield m -def get_any_member(obj: Dict[str, Any], cluster: Cluster, group: Optional[int], +def get_any_member(cluster: Cluster, group: Optional[int], role: Optional[str] = None, member: Optional[str] = None) -> Optional[Member]: """Get the first found cluster member that has the given *role*. - :param obj: the Patroni configuration. :param cluster: the Patroni cluster. :param group: filter which Citus group we should get members from. If ``None`` get from all groups. :param role: role to filter members. See :func:`get_all_members` for available options. @@ -514,7 +528,7 @@ def get_any_member(obj: Dict[str, Any], cluster: Cluster, group: Optional[int], elif role is None: role = 'leader' - for m in get_all_members(obj, cluster, group, role): + for m in get_all_members(cluster, group, role): if member is None or m.name == member: return m @@ -535,7 +549,7 @@ def get_all_members_leader_first(cluster: Cluster) -> Iterator[Member]: yield member -def get_cursor(obj: Dict[str, Any], cluster: Cluster, group: Optional[int], connect_parameters: Dict[str, Any], +def get_cursor(cluster: Cluster, group: Optional[int], connect_parameters: Dict[str, Any], role: Optional[str] = None, member_name: Optional[str] = None) -> Union['cursor', 'Cursor[Any]', None]: """Get a cursor object to execute queries against a member that has the given *role* or *member_name*. @@ -544,7 +558,6 @@ def get_cursor(obj: Dict[str, Any], cluster: Cluster, group: Optional[int], conn * ``fallback_application_name``: as ``Patroni ctl``; * ``connect_timeout``: as ``5``. - :param obj: the Patroni configuration. :param cluster: the Patroni cluster. :param group: filter which Citus group we should get members to create a cursor against. If ``None`` consider members from all groups. @@ -559,7 +572,7 @@ def get_cursor(obj: Dict[str, Any], cluster: Cluster, group: Optional[int], conn * A :class:`psycopg2.extensions.cursor` if using :mod:`psycopg2`; * ``None`` if not able to get a cursor that attendees *role* and *member_name*. """ - member = get_any_member(obj, cluster, group, role=role, member=member_name) + member = get_any_member(cluster, group, role=role, member=member_name) if member is None: return None @@ -594,7 +607,7 @@ def get_cursor(obj: Dict[str, Any], cluster: Cluster, group: Optional[int], conn return None -def get_members(obj: Dict[str, Any], cluster: Cluster, cluster_name: str, member_names: List[str], role: str, +def get_members(cluster: Cluster, cluster_name: str, member_names: List[str], role: str, force: bool, action: str, ask_confirmation: bool = True, group: Optional[int] = None) -> List[Member]: """Get the list of members based on the given filters. @@ -618,7 +631,6 @@ def get_members(obj: Dict[str, Any], cluster: Cluster, cluster_name: str, member ``ask_confirmation=False``, and later call :func:`confirm_members_action` manually in the caller method. That way the workflow won't look broken to the user that is interacting with ``patronictl``. - :param obj: Patroni configuration. :param cluster: Patroni cluster. :param cluster_name: name of the Patroni cluster. :param member_names: used to filter which members should take the *action* based on their names. Each item is the @@ -647,13 +659,13 @@ def get_members(obj: Dict[str, Any], cluster: Cluster, cluster_name: str, member * Cluster does not have members that match the given *member_names*; or * No member with given *role* is found among the specified *member_names*. """ - members = list(get_all_members(obj, cluster, group, role)) + members = list(get_all_members(cluster, group, role)) candidates = {m.name for m in members} if not force or role: if not member_names and not candidates: raise PatroniCtlException('{0} cluster doesn\'t have any members'.format(cluster_name)) - output_members(obj, cluster, cluster_name, group=group) + output_members(cluster, cluster_name, group=group) if member_names: member_names = list(set(member_names) & candidates) @@ -713,9 +725,7 @@ def confirm_members_action(members: List[Member], force: bool, action: str, @click.option('--member', '-m', help='Generate a dsn for this member', type=str) @arg_cluster_name @option_citus_group -@click.pass_obj -def dsn(obj: Dict[str, Any], cluster_name: str, group: Optional[int], - role: Optional[str], member: Optional[str]) -> None: +def dsn(cluster_name: str, group: Optional[int], role: Optional[str], member: Optional[str]) -> None: """Process ``dsn`` command of ``patronictl`` utility. Get DSN to connect to *member*. @@ -723,7 +733,6 @@ def dsn(obj: Dict[str, Any], cluster_name: str, group: Optional[int], .. note:: If no *role* nor *member* is given assume *role* as ``leader``. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter which Citus group we should get members to get DSN from. Refer to the module note for more details. @@ -736,8 +745,8 @@ def dsn(obj: Dict[str, Any], cluster_name: str, group: Optional[int], * both *role* and *member* are provided; or * No member matches requested *member* or *role*. """ - cluster = get_dcs(obj, cluster_name, group).get_cluster() - m = get_any_member(obj, cluster, group, role=role, member=member) + cluster = get_dcs(cluster_name, group).get_cluster() + m = get_any_member(cluster, group, role=role, member=member) if m is None: raise PatroniCtlException('Can not find a suitable member') @@ -759,9 +768,7 @@ def dsn(obj: Dict[str, Any], cluster_name: str, group: Optional[int], @click.option('--delimiter', help='The column delimiter', default='\t') @click.option('--command', '-c', help='The SQL commands to execute') @click.option('-d', '--dbname', help='database name to connect to', type=str) -@click.pass_obj def query( - obj: Dict[str, Any], cluster_name: str, group: Optional[int], role: Optional[str], @@ -780,7 +787,6 @@ def query( Perform a Postgres query in a Patroni node. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter which Citus group we should get members from to perform the query. Refer to the module note for more details. @@ -820,24 +826,22 @@ def query( if dbname: connect_parameters['dbname'] = dbname - dcs = get_dcs(obj, cluster_name, group) + dcs = get_dcs(cluster_name, group) cluster = cursor = None for _ in watching(w, watch, clear=False): if cluster is None: cluster = dcs.get_cluster() -# cursor = get_cursor(obj, cluster, group, connect_parameters, role=role, member=member) - output, header = query_member(obj, cluster, group, cursor, member, role, sql, connect_parameters) + output, header = query_member(cluster, group, cursor, member, role, sql, connect_parameters) print_output(header, output, fmt=fmt, delimiter=delimiter) -def query_member(obj: Dict[str, Any], cluster: Cluster, group: Optional[int], - cursor: Union['cursor', 'Cursor[Any]', None], member: Optional[str], role: Optional[str], - command: str, connect_parameters: Dict[str, Any]) -> Tuple[List[List[Any]], Optional[List[Any]]]: +def query_member(cluster: Cluster, group: Optional[int], cursor: Union['cursor', 'Cursor[Any]', None], + member: Optional[str], role: Optional[str], command: str, + connect_parameters: Dict[str, Any]) -> Tuple[List[List[Any]], Optional[List[Any]]]: """Execute SQL *command* against a member. - :param obj: Patroni configuration. :param cluster: the Patroni cluster. :param group: filter which Citus group we should get members from to perform the query. Refer to the module note for more details. @@ -866,7 +870,7 @@ def query_member(obj: Dict[str, Any], cluster: Cluster, group: Optional[int], from . import psycopg try: if cursor is None: - cursor = get_cursor(obj, cluster, group, connect_parameters, role=role, member_name=member) + cursor = get_cursor(cluster, group, connect_parameters, role=role, member_name=member) if cursor is None: if member is not None: @@ -893,13 +897,11 @@ def query_member(obj: Dict[str, Any], cluster: Cluster, group: Optional[int], @click.argument('cluster_name') @option_citus_group @option_format -@click.pass_obj -def remove(obj: Dict[str, Any], cluster_name: str, group: Optional[int], fmt: str) -> None: +def remove(cluster_name: str, group: Optional[int], fmt: str) -> None: """Process ``remove`` command of ``patronictl`` utility. Remove cluster *cluster_name* from the DCS. - :param obj: Patroni configuration. :param cluster_name: name of the cluster which information will be wiped out of the DCS. :param group: which Citus group should have its information wiped out of the DCS. Refer to the module note for more details. @@ -913,12 +915,12 @@ def remove(obj: Dict[str, Any], cluster_name: str, group: Optional[int], fmt: st * use did not type the correct leader name when requesting removal of a healthy cluster. """ - dcs = get_dcs(obj, cluster_name, group) + dcs = get_dcs(cluster_name, group) cluster = dcs.get_cluster() - if obj.get('citus') and group is None: + if is_citus_cluster() and group is None: raise PatroniCtlException('For Citus clusters the --group must me specified') - output_members(obj, cluster, cluster_name, fmt=fmt) + output_members(cluster, cluster_name, fmt=fmt) confirm = click.prompt('Please confirm the cluster name to remove', type=str) if confirm != cluster_name: @@ -1003,24 +1005,21 @@ def parse_scheduled(scheduled: Optional[str]) -> Optional[datetime.datetime]: @option_citus_group @click.option('--role', '-r', help='Reload only members with this role', type=role_choice, default='any') @option_force -@click.pass_obj -def reload(obj: Dict[str, Any], cluster_name: str, member_names: List[str], - group: Optional[int], force: bool, role: str) -> None: +def reload(cluster_name: str, member_names: List[str], group: Optional[int], force: bool, role: str) -> None: """Process ``reload`` command of ``patronictl`` utility. Reload configuration of cluster members based on given filters. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param member_names: name of the members which configuration should be reloaded. :param group: filter which Citus group we should reload members. Refer to the module note for more details. :param force: perform the reload without asking for confirmations. :param role: role to filter members. See :func:`get_all_members` for available options. """ - dcs = get_dcs(obj, cluster_name, group) + dcs = get_dcs(cluster_name, group) cluster = dcs.get_cluster() - members = get_members(obj, cluster, cluster_name, member_names, role, force, 'reload', group=group) + members = get_members(cluster, cluster_name, member_names, role, force, 'reload', group=group) for member in members: r = request_patroni(member, 'post', 'reload') @@ -1050,15 +1049,13 @@ def reload(obj: Dict[str, Any], cluster_name: str, member_names: List[str], @click.option('--pending', help='Restart if pending', is_flag=True) @click.option('--timeout', help='Return error and fail over if necessary when restarting takes longer than this.') @option_force -@click.pass_obj -def restart(obj: Dict[str, Any], cluster_name: str, group: Optional[int], member_names: List[str], +def restart(cluster_name: str, group: Optional[int], member_names: List[str], force: bool, role: str, p_any: bool, scheduled: Optional[str], version: Optional[str], pending: bool, timeout: Optional[str]) -> None: """Process ``restart`` command of ``patronictl`` utility. Restart Postgres on cluster members based on given filters. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter which Citus group we should restart members. Refer to the module note for more details. :param member_names: name of the members that should be restarted. @@ -1076,9 +1073,9 @@ 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. """ - cluster = get_dcs(obj, cluster_name, group).get_cluster() + cluster = get_dcs(cluster_name, group).get_cluster() - members = get_members(obj, cluster, cluster_name, member_names, role, force, 'restart', False, group=group) + members = get_members(cluster, cluster_name, member_names, role, force, 'restart', 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') scheduled = click.prompt('When should the restart take place (e.g. ' + next_hour + ') ', @@ -1140,9 +1137,7 @@ def restart(obj: Dict[str, Any], cluster_name: str, group: Optional[int], member @click.argument('member_names', nargs=-1) @option_force @click.option('--wait', help='Wait until reinitialization completes', is_flag=True) -@click.pass_obj -def reinit(obj: Dict[str, Any], cluster_name: str, group: Optional[int], - member_names: List[str], force: bool, wait: bool) -> None: +def reinit(cluster_name: str, group: Optional[int], member_names: List[str], force: bool, wait: bool) -> None: """Process ``reinit`` command of ``patronictl`` utility. Reinitialize cluster members based on given filters. @@ -1150,15 +1145,14 @@ def reinit(obj: Dict[str, Any], cluster_name: str, group: Optional[int], .. note:: Only reinitialize replica members, not a leader. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter which Citus group we should reinit members. Refer to the module note for more details. :param member_names: name of the members that should be reinitialized. :param force: perform the restart without asking for confirmations. :param wait: wait for the operation to complete. """ - cluster = get_dcs(obj, cluster_name, group).get_cluster() - members = get_members(obj, cluster, cluster_name, member_names, 'replica', force, 'reinitialize', group=group) + cluster = get_dcs(cluster_name, group).get_cluster() + members = get_members(cluster, cluster_name, member_names, 'replica', force, 'reinitialize', group=group) wait_on_members: List[Member] = [] for member in members: @@ -1189,8 +1183,8 @@ def reinit(obj: Dict[str, Any], cluster_name: str, group: Optional[int], wait_on_members.remove(member) -def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: str, - group: Optional[int], leader: Optional[str], candidate: Optional[str], +def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[int], + leader: Optional[str], candidate: Optional[str], force: bool, scheduled: Optional[str] = None) -> None: """Perform a failover or a switchover operation in the cluster. @@ -1200,7 +1194,6 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s .. note:: If not able to perform the operation through the REST API, write directly to the DCS as a fall back. - :param obj: Patroni configuration. :param action: action to be taken -- ``failover`` or ``switchover``. :param cluster_name: name of the Patroni cluster. :param group: filter Citus group within we should perform a failover or switchover. If ``None``, user will be @@ -1222,17 +1215,17 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s * trying to schedule a switchover in a cluster that is in maintenance mode; or * user aborts the operation. """ - dcs = get_dcs(obj, cluster_name, group) + dcs = get_dcs(cluster_name, group) cluster = dcs.get_cluster() click.echo('Current cluster topology') - output_members(obj, cluster, cluster_name, group=group) + output_members(cluster, cluster_name, group=group) - if obj.get('citus') and group is None: + if is_citus_cluster() and group is None: if force: raise PatroniCtlException('For Citus clusters the --group must me specified') else: group = click.prompt('Citus group', type=int) - dcs = get_dcs(obj, cluster_name, group) + dcs = get_dcs(cluster_name, group) cluster = dcs.get_cluster() global_config = get_global_config(cluster) @@ -1342,7 +1335,7 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s click.echo('{0} Could not {1} using Patroni api, falling back to DCS'.format(timestamp(), action)) dcs.manual_failover(leader, candidate, scheduled_at=scheduled_at) - output_members(obj, cluster, cluster_name, group=group) + output_members(cluster, cluster_name, group=group) @ctl.command('failover', help='Failover to a replica') @@ -1351,8 +1344,7 @@ def _do_failover_or_switchover(obj: Dict[str, Any], action: str, cluster_name: s @click.option('--leader', '--primary', '--master', 'leader', help='The name of the current leader', default=None) @click.option('--candidate', help='The name of the candidate', default=None) @option_force -@click.pass_obj -def failover(obj: Dict[str, Any], cluster_name: str, group: Optional[int], +def failover(cluster_name: str, group: Optional[int], leader: Optional[str], candidate: Optional[str], force: bool) -> None: """Process ``failover`` command of ``patronictl`` utility. @@ -1366,7 +1358,6 @@ def failover(obj: Dict[str, Any], cluster_name: str, group: Optional[int], .. seealso:: Refer to :func:`_do_failover_or_switchover` for details. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter Citus group within we should perform a failover or switchover. If ``None``, user will be prompted for filling it -- unless *force* is ``True``, in which case an exception is raised by @@ -1381,7 +1372,7 @@ def failover(obj: Dict[str, Any], cluster_name: str, group: Optional[int], click.echo(click.style( 'Supplying a leader name using this command is deprecated and will be removed in a future version of' ' Patroni, change your scripts to use `switchover` instead.\nExecuting switchover!', fg='red')) - _do_failover_or_switchover(obj, action, cluster_name, group, leader, candidate, force) + _do_failover_or_switchover(action, cluster_name, group, leader, candidate, force) @ctl.command('switchover', help='Switchover to a replica') @@ -1392,9 +1383,8 @@ def failover(obj: Dict[str, Any], cluster_name: str, group: Optional[int], @click.option('--scheduled', help='Timestamp of a scheduled switchover in unambiguous format (e.g. ISO 8601)', default=None) @option_force -@click.pass_obj -def switchover(obj: Dict[str, Any], cluster_name: str, group: Optional[int], - leader: Optional[str], candidate: Optional[str], force: bool, scheduled: Optional[str]) -> None: +def switchover(cluster_name: str, group: Optional[int], leader: Optional[str], + candidate: Optional[str], force: bool, scheduled: Optional[str]) -> None: """Process ``switchover`` command of ``patronictl`` utility. Perform a switchover operation in the cluster. @@ -1402,7 +1392,6 @@ def switchover(obj: Dict[str, Any], cluster_name: str, group: Optional[int], .. seealso:: Refer to :func:`_do_failover_or_switchover` for details. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter Citus group within we should perform a switchover. If ``None``, user will be prompted for filling it -- unless *force* is ``True``, in which case an exception is raised by @@ -1412,7 +1401,7 @@ def switchover(obj: Dict[str, Any], cluster_name: str, group: Optional[int], :param force: perform the switchover without asking for confirmations. :param scheduled: timestamp when the switchover should be scheduled to occur. If ``now`` perform immediately. """ - _do_failover_or_switchover(obj, 'switchover', cluster_name, group, leader, candidate, force, scheduled) + _do_failover_or_switchover('switchover', cluster_name, group, leader, candidate, force, scheduled) def generate_topology(level: int, member: Dict[str, Any], @@ -1514,8 +1503,8 @@ def get_cluster_service_info(cluster: Dict[str, Any]) -> List[str]: return service_info -def output_members(obj: Dict[str, Any], cluster: Cluster, name: str, - extended: bool = False, fmt: str = 'pretty', group: Optional[int] = None) -> None: +def output_members(cluster: Cluster, name: str, extended: bool = False, + fmt: str = 'pretty', group: Optional[int] = None) -> None: """Print information about the Patroni cluster and its members. Information is printed to console through :func:`print_output`, and contains: @@ -1540,7 +1529,6 @@ def output_members(obj: Dict[str, Any], cluster: Cluster, name: str, The 3 extended columns are always included if *extended*, even if the member has no value for a given column. If not *extended*, these columns may still be shown if any of the members has any information for them. - :param obj: Patroni configuration. :param cluster: Patroni cluster. :param name: name of the Patroni cluster. :param extended: if extended information (pending restarts, scheduled restarts, node tags) should be printed, if @@ -1558,8 +1546,7 @@ def output_members(obj: Dict[str, Any], cluster: Cluster, name: str, clusters = {group or 0: cluster_as_json(cluster)} - is_citus_cluster = obj.get('citus') - if is_citus_cluster: + if is_citus_cluster(): columns.insert(1, 'Group') if group is None: clusters.update({g: cluster_as_json(c) for g, c in cluster.workers.items()}) @@ -1597,10 +1584,12 @@ def output_members(obj: Dict[str, Any], cluster: Cluster, name: str, rows.append([member.get(n.lower().replace(' ', '_'), '') for n in columns]) - title = 'Citus cluster' if is_citus_cluster else 'Cluster' - title_details = f' ({initialize})' - if is_citus_cluster: + if is_citus_cluster(): + title = 'Citus cluster' title_details = '' if group is None else f' (group: {group}, {initialize})' + else: + title = 'Cluster' + title_details = f' ({initialize})' title = f' {title}: {name}{title_details} ' print_output(columns, rows, {'Group': 'r', 'Lag in MB': 'r', 'TL': 'r'}, fmt, title) @@ -1611,7 +1600,7 @@ def output_members(obj: Dict[str, Any], cluster: Cluster, name: str, for g, c in sorted(clusters.items()): service_info = get_cluster_service_info(c) if service_info: - if is_citus_cluster and group is None: + if is_citus_cluster() and group is None: click.echo('Citus group: {0}'.format(g)) click.echo(' ' + '\n '.join(service_info)) @@ -1624,16 +1613,14 @@ def output_members(obj: Dict[str, Any], cluster: Cluster, name: str, @option_format @option_watch @option_watchrefresh -@click.pass_obj -def members(obj: Dict[str, Any], cluster_names: List[str], group: Optional[int], - fmt: str, watch: Optional[int], w: bool, extended: bool, ts: bool) -> None: +def members(cluster_names: List[str], group: Optional[int], fmt: str, + watch: Optional[int], w: bool, extended: bool, ts: bool) -> None: """Process ``list`` command of ``patronictl`` utility. Print information about the Patroni cluster through :func:`output_members`. - :param obj: Patroni configuration. :param cluster_names: name of clusters that should be printed. If ``None`` consider only the cluster present in - ``scope`` key of *obj*. + ``scope`` key of the configuration. :param group: filter which Citus group we should get members from. Refer to the module note for more details. :param fmt: the output table printing format. See :func:`print_output` for available options. :param watch: if given print output every *watch* seconds. @@ -1642,9 +1629,10 @@ def members(obj: Dict[str, Any], cluster_names: List[str], group: Optional[int], more details. :param ts: if timestamp should be included in the output. """ + config = _get_configuration() if not cluster_names: - if 'scope' in obj: - cluster_names = [obj['scope']] + if 'scope' in config: + cluster_names = [config['scope']] if not cluster_names: return logging.warning('Listing members: No cluster names were provided') @@ -1653,10 +1641,10 @@ def members(obj: Dict[str, Any], cluster_names: List[str], group: Optional[int], click.echo(timestamp(0)) for cluster_name in cluster_names: - dcs = get_dcs(obj, cluster_name, group) + dcs = get_dcs(cluster_name, group) cluster = dcs.get_cluster() - output_members(obj, cluster, cluster_name, extended, fmt, group) + output_members(cluster, cluster_name, extended, fmt, group) @ctl.command('topology', help='Prints ASCII topology for given cluster') @@ -1698,14 +1686,12 @@ def timestamp(precision: int = 6) -> str: @click.argument('target', type=click.Choice(['restart', 'switchover'])) @click.option('--role', '-r', help='Flush only members with this role', type=role_choice, default='any') @option_force -@click.pass_obj -def flush(obj: Dict[str, Any], cluster_name: str, group: Optional[int], +def flush(cluster_name: str, group: Optional[int], member_names: List[str], force: bool, role: str, target: str) -> None: """Process ``flush`` command of ``patronictl`` utility. Discard scheduled restart or switchover events. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter which Citus group we should flush an event. Refer to the module note for more details. :param member_names: name of the members which events should be flushed. @@ -1713,11 +1699,11 @@ def flush(obj: Dict[str, Any], cluster_name: str, group: Optional[int], :param role: role to filter members. See :func:`get_all_members` for available options. :param target: the event that should be flushed -- ``restart`` or ``switchover``. """ - dcs = get_dcs(obj, cluster_name, group) + dcs = get_dcs(cluster_name, group) cluster = dcs.get_cluster() if target == 'restart': - for member in get_members(obj, cluster, cluster_name, member_names, role, force, 'flush', group=group): + for member in get_members(cluster, cluster_name, member_names, role, force, 'flush', group=group): if member.data.get('scheduled_restart'): r = request_patroni(member, 'delete', 'restart') check_response(r, member.name, 'flush scheduled restart') @@ -1775,10 +1761,9 @@ def wait_until_pause_is_applied(dcs: AbstractDCS, paused: bool, old_cluster: Clu return click.echo('Success: cluster management is {0}'.format(paused and 'paused' or 'resumed')) -def toggle_pause(config: Dict[str, Any], cluster_name: str, group: Optional[int], paused: bool, wait: bool) -> None: +def toggle_pause(cluster_name: str, group: Optional[int], paused: bool, wait: bool) -> None: """Toggle the ``pause`` state in the cluster members. - :param config: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter which Citus group we should toggle the pause state of. Refer to the module note for more details. @@ -1790,7 +1775,7 @@ def toggle_pause(config: Dict[str, Any], cluster_name: str, group: Optional[int] * ``pause`` state is already *paused*; or * cluster contains no accessible members. """ - dcs = get_dcs(config, cluster_name, group) + dcs = get_dcs(cluster_name, group) cluster = dcs.get_cluster() if get_global_config(cluster).is_paused == paused: raise PatroniCtlException('Cluster is {0} paused'.format(paused and 'already' or 'not')) @@ -1819,37 +1804,33 @@ def toggle_pause(config: Dict[str, Any], cluster_name: str, group: Optional[int] @ctl.command('pause', help='Disable auto failover') @arg_cluster_name @option_default_citus_group -@click.pass_obj @click.option('--wait', help='Wait until pause is applied on all nodes', is_flag=True) -def pause(obj: Dict[str, Any], cluster_name: str, group: Optional[int], wait: bool) -> None: +def pause(cluster_name: str, group: Optional[int], wait: bool) -> None: """Process ``pause`` command of ``patronictl`` utility. Put the cluster in maintenance mode. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter which Citus group we should pause. Refer to the module note for more details. :param wait: ``True`` if it should block until the operation is finished or ``false`` for returning immediately. """ - return toggle_pause(obj, cluster_name, group, True, wait) + return toggle_pause(cluster_name, group, True, wait) @ctl.command('resume', help='Resume auto failover') @arg_cluster_name @option_default_citus_group @click.option('--wait', help='Wait until pause is cleared on all nodes', is_flag=True) -@click.pass_obj -def resume(obj: Dict[str, Any], cluster_name: str, group: Optional[int], wait: bool) -> None: +def resume(cluster_name: str, group: Optional[int], wait: bool) -> None: """Process ``unpause`` command of ``patronictl`` utility. Put the cluster out of maintenance mode. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter which Citus group we should unpause. Refer to the module note for more details. :param wait: ``True`` if it should block until the operation is finished or ``false`` for returning immediately. """ - return toggle_pause(obj, cluster_name, group, False, wait) + return toggle_pause(cluster_name, group, False, wait) @contextmanager @@ -2081,15 +2062,12 @@ def invoke_editor(before_editing: str, cluster_name: str) -> Tuple[str, Dict[str @click.option('--replace', 'replace_filename', help='Apply configuration from file, replacing existing configuration.' ' Use - for stdin.') @option_force -@click.pass_obj -def edit_config(obj: Dict[str, Any], cluster_name: str, group: Optional[int], - force: bool, quiet: bool, kvpairs: List[str], pgkvpairs: List[str], - apply_filename: Optional[str], replace_filename: Optional[str]) -> None: +def edit_config(cluster_name: str, group: Optional[int], force: bool, quiet: bool, kvpairs: List[str], + pgkvpairs: List[str], apply_filename: Optional[str], replace_filename: Optional[str]) -> None: """Process ``edit-config`` command of ``patronictl`` utility. Update or replace Patroni configuration in the DCS. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter which Citus group configuration we should edit. Refer to the module note for more details. :param force: if ``True`` apply config changes without asking for confirmations. @@ -2106,7 +2084,7 @@ def edit_config(obj: Dict[str, Any], cluster_name: str, group: Optional[int], * Configuration is absent from DCS; or * Detected a concurrent modification of the configuration in the DCS. """ - dcs = get_dcs(obj, cluster_name, group) + dcs = get_dcs(cluster_name, group) cluster = dcs.get_cluster() if not cluster.config: @@ -2152,17 +2130,15 @@ def edit_config(obj: Dict[str, Any], cluster_name: str, group: Optional[int], @ctl.command('show-config', help="Show cluster configuration") @arg_cluster_name @option_default_citus_group -@click.pass_obj -def show_config(obj: Dict[str, Any], cluster_name: str, group: Optional[int]) -> None: +def show_config(cluster_name: str, group: Optional[int]) -> None: """Process ``show-config`` command of ``patronictl`` utility. Show Patroni configuration stored in the DCS. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter which Citus group configuration we should show. Refer to the module note for more details. """ - cluster = get_dcs(obj, cluster_name, group).get_cluster() + cluster = get_dcs(cluster_name, group).get_cluster() if cluster.config: click.echo(format_config_for_editing(cluster.config.data)) @@ -2171,8 +2147,7 @@ def show_config(obj: Dict[str, Any], cluster_name: str, group: Optional[int]) -> @click.argument('cluster_name', required=False) @click.argument('member_names', nargs=-1) @option_citus_group -@click.pass_obj -def version(obj: Dict[str, Any], cluster_name: str, group: Optional[int], member_names: List[str]) -> None: +def version(cluster_name: str, group: Optional[int], member_names: List[str]) -> None: """Process ``version`` command of ``patronictl`` utility. Show version of: @@ -2180,7 +2155,6 @@ def version(obj: Dict[str, Any], cluster_name: str, group: Optional[int], member * ``patroni`` on all members of the cluster; * ``PostgreSQL`` on all members of the cluster. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter which Citus group we should get members from. Refer to the module note for more details. :param member_names: filter which members we should get version information from. @@ -2191,8 +2165,8 @@ def version(obj: Dict[str, Any], cluster_name: str, group: Optional[int], member return click.echo("") - cluster = get_dcs(obj, cluster_name, group).get_cluster() - for m in get_all_members(obj, cluster, group, 'any'): + cluster = get_dcs(cluster_name, group).get_cluster() + for m in get_all_members(cluster, group, 'any'): if m.api_url: if not member_names or m.name in member_names: try: @@ -2210,8 +2184,7 @@ def version(obj: Dict[str, Any], cluster_name: str, group: Optional[int], member @arg_cluster_name @option_default_citus_group @option_format -@click.pass_obj -def history(obj: Dict[str, Any], cluster_name: str, group: Optional[int], fmt: str) -> None: +def history(cluster_name: str, group: Optional[int], fmt: str) -> None: """Process ``history`` command of ``patronictl`` utility. Show the history of failover/switchover events in the cluster. @@ -2223,12 +2196,11 @@ def history(obj: Dict[str, Any], cluster_name: str, group: Optional[int], fmt: s * ``Timestamp``: timestamp when the event occurred; * ``New Leader``: the Postgres node that was promoted during the event. - :param obj: Patroni configuration. :param cluster_name: name of the Patroni cluster. :param group: filter which Citus group we should get events from. Refer to the module note for more details. :param fmt: the output table printing format. See :func:`print_output` for available options. """ - cluster = get_dcs(obj, cluster_name, group).get_cluster() + cluster = get_dcs(cluster_name, group).get_cluster() cluster_history = cluster.history.lines if cluster.history else [] history: List[List[Any]] = list(map(list, cluster_history)) table_header_row = ['TL', 'LSN', 'Reason', 'Timestamp', 'New Leader'] diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 85691812..bacd2d60 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -1,3 +1,4 @@ +import click import etcd import mock import os @@ -9,7 +10,7 @@ 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 -from patroni.dcs.etcd import AbstractEtcdClientWithFailover, Cluster, Failover +from patroni.dcs import Cluster, Failover from patroni.psycopg import OperationalError from patroni.utils import tzutc from prettytable import PrettyTable, ALL @@ -21,26 +22,26 @@ from .test_ha import get_cluster_initialized_without_leader, get_cluster_initial get_cluster_initialized_with_only_leader, get_cluster_not_initialized_without_leader, get_cluster, Member -DEFAULT_CONFIG = { - 'scope': 'alpha', - 'restapi': {'listen': '::', 'certfile': 'a'}, - 'ctl': {'certfile': 'a'}, - 'etcd': {'host': 'localhost:2379'}, - 'citus': {'database': 'citus', 'group': 0}, - 'postgresql': {'data_dir': '.', 'pgpass': './pgpass', 'parameters': {}, 'retry_timeout': 5} -} +def get_default_config(*args): + return { + 'scope': 'alpha', + 'restapi': {'listen': '::', 'certfile': 'a'}, + 'ctl': {'certfile': 'a'}, + 'etcd': {'host': 'localhost:2379', 'retry_timeout': 10, 'ttl': 30}, + 'citus': {'database': 'citus', 'group': 0}, + 'postgresql': {'data_dir': '.', 'pgpass': './pgpass', 'parameters': {}, 'retry_timeout': 5} + } -@patch('patroni.ctl.load_config', Mock(return_value=DEFAULT_CONFIG)) +@patch.object(PoolManager, 'request', Mock(return_value=MockResponse())) +@patch('patroni.ctl.load_config', get_default_config) +@patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) class TestCtl(unittest.TestCase): TEST_ROLES = ('master', 'primary', 'leader') @patch('socket.getaddrinfo', socket_getaddrinfo) - @patch.object(AbstractEtcdClientWithFailover, '_get_machines_list', Mock(return_value=['http://remotehost:2379'])) def setUp(self): self.runner = CliRunner() - self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'retry_timeout': 10}, - 'citus': {'group': 0}}, 'foo', None) @patch('patroni.ctl.logging.debug') def test_load_config(self, mock_logger_debug): @@ -66,29 +67,31 @@ class TestCtl(unittest.TestCase): @patch('patroni.psycopg.connect', psycopg_connect) def test_get_cursor(self): - for role in self.TEST_ROLES: - self.assertIsNone(get_cursor({}, get_cluster_initialized_without_leader(), None, {}, role=role)) - self.assertIsNotNone(get_cursor({}, get_cluster_initialized_with_leader(), None, {}, role=role)) + with click.Context(click.Command('query')) as ctx: + ctx.obj = {'__config': {}} + for role in self.TEST_ROLES: + self.assertIsNone(get_cursor(get_cluster_initialized_without_leader(), None, {}, role=role)) + self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), None, {}, role=role)) - # MockCursor returns pg_is_in_recovery as false - self.assertIsNone(get_cursor({}, get_cluster_initialized_with_leader(), None, {}, role='replica')) + # MockCursor returns pg_is_in_recovery as false + self.assertIsNone(get_cursor(get_cluster_initialized_with_leader(), None, {}, role='replica')) - self.assertIsNotNone(get_cursor({}, get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, role='any')) + self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, role='any')) - # Mutually exclusive options - with self.assertRaises(PatroniCtlException) as e: - get_cursor({}, get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, member_name='other', - role='replica') + # Mutually exclusive options + with self.assertRaises(PatroniCtlException) as e: + get_cursor(get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, member_name='other', + role='replica') - self.assertEqual(str(e.exception), '--role and --member are mutually exclusive options') + self.assertEqual(str(e.exception), '--role and --member are mutually exclusive options') - # Invalid member provided - self.assertIsNone(get_cursor({}, get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, - member_name='invalid')) + # Invalid member provided + self.assertIsNone(get_cursor(get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, + member_name='invalid')) - # Valid member provided - self.assertIsNotNone(get_cursor({}, get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, - member_name='other')) + # Valid member provided + self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), None, {'dbname': 'foo'}, + member_name='other')) def test_parse_dcs(self): assert parse_dcs(None) is None @@ -102,23 +105,20 @@ class TestCtl(unittest.TestCase): self.assertRaises(PatroniCtlException, parse_dcs, 'invalid://test') def test_output_members(self): - scheduled_at = datetime.now(tzutc) + timedelta(seconds=600) - cluster = get_cluster_initialized_with_leader(Failover(1, 'foo', 'bar', scheduled_at)) - del cluster.members[1].data['conn_url'] - for fmt in ('pretty', 'json', 'yaml', 'topology'): - self.assertIsNone(output_members({}, cluster, name='abc', fmt=fmt)) + with click.Context(click.Command('list')) as ctx: + ctx.obj = {'__config': {}} + scheduled_at = datetime.now(tzutc) + timedelta(seconds=600) + cluster = get_cluster_initialized_with_leader(Failover(1, 'foo', 'bar', scheduled_at)) + del cluster.members[1].data['conn_url'] + for fmt in ('pretty', 'json', 'yaml', 'topology'): + self.assertIsNone(output_members(cluster, name='abc', fmt=fmt)) - with patch('click.echo') as mock_echo: - self.assertIsNone(output_members({}, cluster, name='abc', fmt='tsv')) - self.assertEqual(mock_echo.call_args[0][0], 'abc\tother\t\tReplica\trunning\t\tunknown') - - @patch('patroni.ctl.get_dcs') - @patch.object(PoolManager, 'request', Mock(return_value=MockResponse())) - def test_switchover(self, mock_get_dcs): - mock_get_dcs.return_value = self.e - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader - mock_get_dcs.return_value.set_failover_value = Mock() + with patch('click.echo') as mock_echo: + self.assertIsNone(output_members(cluster, name='abc', fmt='tsv')) + self.assertEqual(mock_echo.call_args[0][0], 'abc\tother\t\tReplica\trunning\t\tunknown') + @patch('patroni.dcs.AbstractDCS.set_failover_value', Mock()) + def test_switchover(self): # Confirm result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny') self.assertEqual(result.exit_code, 0) @@ -180,12 +180,12 @@ class TestCtl(unittest.TestCase): 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)): + with patch('patroni.ctl.request_patroni', 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: + with patch('patroni.ctl.request_patroni') 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) @@ -196,64 +196,58 @@ class TestCtl(unittest.TestCase): 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') - self.assertEqual(result.exit_code, 1) - self.assertIn('No candidates found to switchover to', result.output) + with patch('patroni.dcs.AbstractDCS.get_cluster', + Mock(return_value=get_cluster_initialized_with_only_leader())): + result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny') + self.assertEqual(result.exit_code, 1) + self.assertIn('No candidates found to switchover to', result.output) # No leader available - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_without_leader - result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny') - self.assertEqual(result.exit_code, 1) - self.assertIn('This cluster has no leader', result.output) + with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_without_leader())): + result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nother\n\ny') + self.assertEqual(result.exit_code, 1) + self.assertIn('This cluster has no leader', result.output) # Citus cluster, no group number specified result = self.runner.invoke(ctl, ['switchover', 'dummy', '--force'], input='\n') self.assertEqual(result.exit_code, 1) self.assertIn('For Citus clusters the --group must me specified', result.output) - @patch('patroni.ctl.get_dcs') - @patch.object(PoolManager, 'request', Mock(return_value=MockResponse())) - @patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse())) - def test_failover(self, mock_get_dcs): - mock_get_dcs.return_value.set_failover_value = Mock() - + @patch('patroni.dcs.AbstractDCS.set_failover_value', Mock()) + def test_failover(self): # 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) - cluster = get_cluster_initialized_with_leader(sync=('leader', 'other')) - # Temp test to check a fallback to switchover if leader is specified with patch('patroni.ctl._do_failover_or_switchover') as failover_func_mock: result = self.runner.invoke(ctl, ['failover', '--leader', 'leader', 'dummy'], input='0\n') self.assertIn('Supplying a leader name using this command is deprecated', result.output) - failover_func_mock.assert_called_once_with( - DEFAULT_CONFIG, 'switchover', 'dummy', None, 'leader', None, False) + failover_func_mock.assert_called_once_with('switchover', 'dummy', None, 'leader', None, False) # Failover to an async member in sync mode (confirm) + cluster = get_cluster_initialized_with_leader(sync=('leader', 'other')) cluster.members.append(Member(0, 'async', 28, {'api_url': 'http://127.0.0.1:8012/patroni'})) cluster.config.data['synchronous_mode'] = True - mock_get_dcs.return_value.get_cluster = Mock(return_value=cluster) - result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='y\ny') - self.assertIn('Are you sure you want to failover to the asynchronous node async', result.output) + with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=cluster)): + result = self.runner.invoke(ctl, + ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='y\ny') + self.assertIn('Are you sure you want to failover to the asynchronous node async', result.output) # Failover to an async member in sync mode (abort) - mock_get_dcs.return_value.get_cluster = Mock(return_value=cluster) result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='N') self.assertEqual(result.exit_code, 1) @patch('patroni.dcs.dcs_modules', Mock(return_value=['patroni.dcs.dummy', 'patroni.dcs.etcd'])) def test_get_dcs(self): - self.assertRaises(PatroniCtlException, get_dcs, {'dummy': {}}, 'dummy', 0) + with click.Context(click.Command('list')) as ctx: + ctx.obj = {'__config': {'dummy': {}}} + self.assertRaises(PatroniCtlException, get_dcs, 'dummy', 0) @patch('patroni.psycopg.connect', psycopg_connect) @patch('patroni.ctl.query_member', Mock(return_value=([['mock column']], None))) - @patch('patroni.ctl.get_dcs') @patch.object(etcd.Client, 'read', etcd_read) - def test_query(self, mock_get_dcs): - mock_get_dcs.return_value = self.e + def test_query(self): # Mutually exclusive for role in self.TEST_ROLES: result = self.runner.invoke(ctl, ['query', 'alpha', '--member', 'abc', '--role', role]) @@ -286,31 +280,29 @@ class TestCtl(unittest.TestCase): def test_query_member(self): with patch('patroni.ctl.get_cursor', Mock(return_value=MockConnect().cursor())): for role in self.TEST_ROLES: - rows = query_member({}, None, None, None, None, role, 'SELECT pg_catalog.pg_is_in_recovery()', {}) + rows = query_member(None, None, None, None, role, 'SELECT pg_catalog.pg_is_in_recovery()', {}) self.assertTrue('False' in str(rows)) with patch.object(MockCursor, 'execute', Mock(side_effect=OperationalError('bla'))): - rows = query_member({}, None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {}) + rows = query_member(None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {}) with patch('patroni.ctl.get_cursor', Mock(return_value=None)): # No role nor member given -- generic message - rows = query_member({}, None, None, None, None, None, 'SELECT pg_catalog.pg_is_in_recovery()', {}) + rows = query_member(None, None, None, None, None, 'SELECT pg_catalog.pg_is_in_recovery()', {}) self.assertTrue('No connection is available' in str(rows)) # Member given -- message pointing to member - rows = query_member({}, None, None, None, 'foo', None, 'SELECT pg_catalog.pg_is_in_recovery()', {}) + rows = query_member(None, None, None, 'foo', None, 'SELECT pg_catalog.pg_is_in_recovery()', {}) self.assertTrue('No connection to member foo' in str(rows)) # Role given -- message pointing to role - rows = query_member({}, None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {}) + rows = query_member(None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {}) self.assertTrue('No connection to role replica' in str(rows)) with patch('patroni.ctl.get_cursor', Mock(side_effect=OperationalError('bla'))): - rows = query_member({}, None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {}) + rows = query_member(None, None, None, None, 'replica', 'SELECT pg_catalog.pg_is_in_recovery()', {}) - @patch('patroni.ctl.get_dcs') - def test_dsn(self, mock_get_dcs): - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader + def test_dsn(self): result = self.runner.invoke(ctl, ['dsn', 'alpha']) assert 'host=127.0.0.1 port=5435' in result.output @@ -323,11 +315,8 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['dsn', 'alpha', '--member', 'dummy']) assert result.exit_code == 1 - @patch.object(PoolManager, 'request') - @patch('patroni.ctl.get_dcs') - def test_reload(self, mock_get_dcs, mock_post): - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader - + @patch('patroni.ctl.request_patroni') + def test_reload(self, mock_post): result = self.runner.invoke(ctl, ['reload', 'alpha'], input='y') assert 'Failed: reload for member' in result.output @@ -339,10 +328,8 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['reload', 'alpha'], input='y') assert 'Reload request received for member' in result.output - @patch.object(PoolManager, 'request') - @patch('patroni.ctl.get_dcs') - def test_restart_reinit(self, mock_get_dcs, mock_post): - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader + @patch('patroni.ctl.request_patroni') + def test_restart_reinit(self, mock_post): mock_post.return_value.status = 503 result = self.runner.invoke(ctl, ['restart', 'alpha'], input='now\ny\n') assert 'Failed: restart for' in result.output @@ -417,12 +404,10 @@ class TestCtl(unittest.TestCase): assert 'Failed: another restart is already' in result.output assert result.exit_code == 0 - @patch('patroni.ctl.get_dcs') - def test_remove(self, mock_get_dcs): - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader + def test_remove(self): result = self.runner.invoke(ctl, ['remove', 'dummy'], input='\n') assert 'For Citus clusters the --group must me specified' in result.output - result = self.runner.invoke(ctl, ['-k', 'remove', 'alpha', '--group', '0'], input='alpha\nstandby') + result = self.runner.invoke(ctl, ['remove', 'alpha', '--group', '0'], input='alpha\nstandby') assert 'Please confirm' in result.output assert 'You are about to remove all' in result.output # Not typing an exact confirmation @@ -440,37 +425,36 @@ class TestCtl(unittest.TestCase): assert result.exit_code == 0 def test_ctl(self): - self.runner.invoke(ctl, ['list']) - result = self.runner.invoke(ctl, ['--help']) assert 'Usage:' in result.output def test_get_any_member(self): - for role in self.TEST_ROLES: - self.assertIsNone(get_any_member({}, get_cluster_initialized_without_leader(), None, role=role)) + with click.Context(click.Command('list')) as ctx: + ctx.obj = {'__config': {}} + for role in self.TEST_ROLES: + self.assertIsNone(get_any_member(get_cluster_initialized_without_leader(), None, role=role)) - m = get_any_member({}, get_cluster_initialized_with_leader(), None, role=role) - self.assertEqual(m.name, 'leader') + m = get_any_member(get_cluster_initialized_with_leader(), None, role=role) + self.assertEqual(m.name, 'leader') def test_get_all_members(self): - for role in self.TEST_ROLES: - self.assertEqual(list(get_all_members({}, get_cluster_initialized_without_leader(), None, role=role)), []) + with click.Context(click.Command('list')) as ctx: + ctx.obj = {'__config': {}} + for role in self.TEST_ROLES: + self.assertEqual(list(get_all_members(get_cluster_initialized_without_leader(), None, role=role)), []) - r = list(get_all_members({}, get_cluster_initialized_with_leader(), None, role=role)) + r = list(get_all_members(get_cluster_initialized_with_leader(), None, role=role)) + self.assertEqual(len(r), 1) + self.assertEqual(r[0].name, 'leader') + + r = list(get_all_members(get_cluster_initialized_with_leader(), None, role='replica')) self.assertEqual(len(r), 1) - self.assertEqual(r[0].name, 'leader') + self.assertEqual(r[0].name, 'other') - r = list(get_all_members({}, get_cluster_initialized_with_leader(), None, role='replica')) - self.assertEqual(len(r), 1) - self.assertEqual(r[0].name, 'other') - - self.assertEqual(len(list(get_all_members({}, get_cluster_initialized_without_leader(), - None, role='replica'))), 2) - - @patch('patroni.ctl.get_dcs') - def test_members(self, mock_get_dcs): - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader + self.assertEqual(len(list(get_all_members(get_cluster_initialized_without_leader(), + None, role='replica'))), 2) + def test_members(self): result = self.runner.invoke(ctl, ['list']) assert '127.0.0.1' in result.output assert result.exit_code == 0 @@ -479,121 +463,94 @@ class TestCtl(unittest.TestCase): result = self.runner.invoke(ctl, ['list', '--group', '0']) assert 'Citus cluster: alpha (group: 0, 12345678901) -' in result.output - with patch('patroni.ctl.load_config', Mock(return_value={'scope': 'alpha'})): + config = get_default_config() + del config['citus'] + with patch('patroni.ctl.load_config', Mock(return_value=config)): result = self.runner.invoke(ctl, ['list']) assert 'Cluster: alpha (12345678901) -' in result.output with patch('patroni.ctl.load_config', Mock(return_value={})): self.runner.invoke(ctl, ['list']) - @patch('patroni.ctl.get_dcs') - def test_list_extended(self, mock_get_dcs): - mock_get_dcs.return_value = self.e - cluster = get_cluster_initialized_with_leader(sync=('leader', 'other')) - mock_get_dcs.return_value.get_cluster = Mock(return_value=cluster) - + def test_list_extended(self): result = self.runner.invoke(ctl, ['list', 'dummy', '--extended', '--timestamp']) assert '2100' in result.output assert 'Scheduled restart' in result.output - @patch('patroni.ctl.get_dcs') - def test_topology(self, mock_get_dcs): - mock_get_dcs.return_value = self.e + def test_topology(self): cluster = get_cluster_initialized_with_leader() - cascade_member = Member(0, 'cascade', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5437/postgres', - 'api_url': 'http://127.0.0.1:8012/patroni', - 'state': 'running', - 'tags': {'replicatefrom': 'other'}, - }) - cascade_member_wrong_tags = Member(0, 'wrong_cascade', 28, - {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5438/postgres', - 'api_url': 'http://127.0.0.1:8013/patroni', - 'state': 'running', - 'tags': {'replicatefrom': 'nonexistinghost'}, - }) - cluster.members.append(cascade_member) - cluster.members.append(cascade_member_wrong_tags) - mock_get_dcs.return_value.get_cluster = Mock(return_value=cluster) - result = self.runner.invoke(ctl, ['topology', 'dummy']) - assert '+\n| 0 | leader | 127.0.0.1:5435 | Leader |' in result.output - assert '|\n| 0 | + other | 127.0.0.1:5436 | Replica |' in result.output - assert '|\n| 0 | + cascade | 127.0.0.1:5437 | Replica |' in result.output - assert '|\n| 0 | + wrong_cascade | 127.0.0.1:5438 | Replica |' in result.output + cluster.members.append(Member(0, 'cascade', 28, + {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5437/postgres', + 'api_url': 'http://127.0.0.1:8012/patroni', 'state': 'running', + 'tags': {'replicatefrom': 'other'}})) + cluster.members.append(Member(0, 'wrong_cascade', 28, + {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5438/postgres', + 'api_url': 'http://127.0.0.1:8013/patroni', 'state': 'running', + 'tags': {'replicatefrom': 'nonexistinghost'}})) + with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=cluster)): + result = self.runner.invoke(ctl, ['topology', 'dummy']) + assert '+\n| 0 | leader | 127.0.0.1:5435 | Leader |' in result.output + assert '|\n| 0 | + other | 127.0.0.1:5436 | Replica |' in result.output + assert '|\n| 0 | + cascade | 127.0.0.1:5437 | Replica |' in result.output + assert '|\n| 0 | + wrong_cascade | 127.0.0.1:5438 | Replica |' in result.output - cluster = get_cluster_initialized_without_leader() - mock_get_dcs.return_value.get_cluster = Mock(return_value=cluster) - result = self.runner.invoke(ctl, ['topology', 'dummy']) - assert '+\n| 0 | + leader | 127.0.0.1:5435 | Replica |' in result.output - assert '|\n| 0 | + other | 127.0.0.1:5436 | Replica |' in result.output - - @patch('patroni.ctl.get_dcs') - @patch.object(PoolManager, 'request', Mock(return_value=MockResponse())) - def test_flush_restart(self, mock_get_dcs): - mock_get_dcs.return_value = self.e - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader + with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_without_leader())): + result = self.runner.invoke(ctl, ['topology', 'dummy']) + assert '+\n| 0 | + leader | 127.0.0.1:5435 | Replica |' in result.output + assert '|\n| 0 | + other | 127.0.0.1:5436 | Replica |' in result.output + @patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) + def test_flush_restart(self): for role in self.TEST_ROLES: - result = self.runner.invoke(ctl, ['-k', 'flush', 'dummy', 'restart', '-r', role], input='y') + result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '-r', role], input='y') assert 'No scheduled restart' in result.output result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '--force']) assert 'Success: flush scheduled restart' in result.output - with patch.object(PoolManager, 'request', return_value=MockResponse(404)): + with patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse(404))): result = self.runner.invoke(ctl, ['flush', 'dummy', 'restart', '--force']) assert 'Failed: flush scheduled restart' in result.output - @patch('patroni.ctl.get_dcs') - @patch.object(PoolManager, 'request', Mock(return_value=MockResponse())) - def test_flush_switchover(self, mock_get_dcs): - mock_get_dcs.return_value = self.e - - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader - result = self.runner.invoke(ctl, ['flush', 'dummy', 'switchover']) - assert 'No pending scheduled switchover' in result.output + def test_flush_switchover(self): + with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())): + result = self.runner.invoke(ctl, ['flush', 'dummy', 'switchover']) + assert 'No pending scheduled switchover' in result.output scheduled_at = datetime.now(tzutc) + timedelta(seconds=600) - mock_get_dcs.return_value.get_cluster = Mock( - return_value=get_cluster_initialized_with_leader(Failover(1, 'a', 'b', scheduled_at))) - result = self.runner.invoke(ctl, ['flush', 'dummy', 'switchover']) - assert result.output.startswith('Success: ') + with patch('patroni.dcs.AbstractDCS.get_cluster', + Mock(return_value=get_cluster_initialized_with_leader(Failover(1, 'a', 'b', scheduled_at)))): + result = self.runner.invoke(ctl, ['-k', 'flush', 'dummy', 'switchover']) + assert result.output.startswith('Success: ') - mock_get_dcs.return_value.manual_failover = Mock() - with patch.object(PoolManager, 'request', side_effect=[MockResponse(409), Exception]): - result = self.runner.invoke(ctl, ['flush', 'dummy', 'switchover']) - assert 'Could not find any accessible member of cluster' in result.output + with patch('patroni.ctl.request_patroni', side_effect=[MockResponse(409), Exception]), \ + patch('patroni.dcs.AbstractDCS.manual_failover', Mock()): + result = self.runner.invoke(ctl, ['flush', 'dummy', 'switchover']) + assert 'Could not find any accessible member of cluster' in result.output - @patch.object(PoolManager, 'request') - @patch('patroni.ctl.get_dcs') @patch('patroni.ctl.polling_loop', Mock(return_value=[1])) - def test_pause_cluster(self, mock_get_dcs, mock_post): - mock_get_dcs.return_value = self.e - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader + def test_pause_cluster(self): + with patch('patroni.ctl.request_patroni', Mock(return_value=MockResponse(500))): + result = self.runner.invoke(ctl, ['pause', 'dummy']) + assert 'Failed' in result.output - mock_post.return_value.status = 500 - result = self.runner.invoke(ctl, ['pause', 'dummy']) - assert 'Failed' in result.output - - mock_post.return_value.status = 200 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 result = self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) assert "'pause' request sent" in result.output - mock_get_dcs.return_value.get_cluster = Mock(side_effect=[get_cluster_initialized_with_leader(), - get_cluster(None, None, [], None, None)]) - self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) - member = Member(1, 'other', 28, {}) - mock_get_dcs.return_value.get_cluster = Mock(side_effect=[get_cluster_initialized_with_leader(), - get_cluster(None, None, [member], None, None)]) - self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) - @patch.object(PoolManager, 'request') - @patch('patroni.ctl.get_dcs') - def test_resume_cluster(self, mock_get_dcs, mock_post): - mock_get_dcs.return_value = self.e - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader + with patch('patroni.dcs.AbstractDCS.get_cluster', + Mock(side_effect=[get_cluster_initialized_with_leader(), get_cluster(None, None, [], None, None)])): + self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) + with patch('patroni.dcs.AbstractDCS.get_cluster', + Mock(side_effect=[get_cluster_initialized_with_leader(), + get_cluster(None, None, [Member(1, 'other', 28, {})], None, None)])): + self.runner.invoke(ctl, ['pause', 'dummy', '--wait']) + @patch('patroni.ctl.request_patroni') + @patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) + def test_resume_cluster(self, mock_post): mock_post.return_value.status = 200 with patch('patroni.config.GlobalConfig.is_paused', PropertyMock(return_value=False)): result = self.runner.invoke(ctl, ['resume', 'dummy']) @@ -701,67 +658,53 @@ class TestCtl(unittest.TestCase): with patch('shutil.which', Mock(return_value=e)): self.assertRaises(PatroniCtlException, invoke_editor, 'foo: bar\n', 'test') - @patch('patroni.ctl.get_dcs') - def test_show_config(self, mock_get_dcs): - mock_get_dcs.return_value = self.e - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader + def test_show_config(self): self.runner.invoke(ctl, ['show-config', 'dummy']) - @patch('patroni.ctl.get_dcs') @patch('subprocess.call', Mock(return_value=0)) - def test_edit_config(self, mock_get_dcs): - mock_get_dcs.return_value = self.e - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader - mock_get_dcs.return_value.set_config_value = Mock(return_value=False) + def test_edit_config(self): os.environ['EDITOR'] = 'true' self.runner.invoke(ctl, ['edit-config', 'dummy']) self.runner.invoke(ctl, ['edit-config', 'dummy', '-s', 'foo=bar']) self.runner.invoke(ctl, ['edit-config', 'dummy', '--replace', 'postgres0.yml']) self.runner.invoke(ctl, ['edit-config', 'dummy', '--apply', '-'], input='foo: bar') self.runner.invoke(ctl, ['edit-config', 'dummy', '--force', '--apply', '-'], input='foo: bar') - mock_get_dcs.return_value.set_config_value.return_value = True - self.runner.invoke(ctl, ['edit-config', 'dummy', '--force', '--apply', '-'], input='foo: bar') - mock_get_dcs.return_value.get_cluster = Mock(return_value=Cluster.empty()) - result = self.runner.invoke(ctl, ['edit-config', 'dummy']) - assert result.exit_code == 1 - assert 'The config key does not exist in the cluster dummy' in result.output + with patch('patroni.dcs.etcd.Etcd.set_config_value', Mock(return_value=True)): + self.runner.invoke(ctl, ['edit-config', 'dummy', '--force', '--apply', '-'], input='foo: bar') + with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=Cluster.empty())): + result = self.runner.invoke(ctl, ['edit-config', 'dummy']) + assert result.exit_code == 1 + assert 'The config key does not exist in the cluster dummy' in result.output - @patch('patroni.ctl.get_dcs') - def test_version(self, mock_get_dcs): - mock_get_dcs.return_value = self.e - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader - with patch.object(PoolManager, 'request') as mocked: - result = self.runner.invoke(ctl, ['version']) - assert 'patronictl version' in result.output - mocked.return_value.data = b'{"patroni":{"version":"1.2.3"},"server_version": 100001}' - result = self.runner.invoke(ctl, ['version', 'dummy']) - assert '1.2.3' in result.output - with patch.object(PoolManager, 'request', Mock(side_effect=Exception)): - result = self.runner.invoke(ctl, ['version', 'dummy']) - assert 'failed to get version' in result.output + @patch('patroni.ctl.request_patroni') + def test_version(self, mock_request): + result = self.runner.invoke(ctl, ['version']) + assert 'patronictl version' in result.output + mock_request.return_value.data = b'{"patroni":{"version":"1.2.3"},"server_version": 100001}' + result = self.runner.invoke(ctl, ['version', 'dummy']) + assert '1.2.3' in result.output + mock_request.side_effect = Exception + result = self.runner.invoke(ctl, ['version', 'dummy']) + assert 'failed to get version' in result.output - @patch('patroni.ctl.get_dcs') - def test_history(self, mock_get_dcs): - mock_get_dcs.return_value.get_cluster = Mock() - mock_get_dcs.return_value.get_cluster.return_value.history.lines = [[1, 67176, 'no recovery target specified']] - result = self.runner.invoke(ctl, ['history']) - assert 'Reason' in result.output + def test_history(self): + with patch('patroni.dcs.AbstractDCS.get_cluster') as mock_get_cluster: + mock_get_cluster.return_value.history.lines = [[1, 67176, 'no recovery target specified']] + result = self.runner.invoke(ctl, ['history']) + assert 'Reason' in result.output def test_format_pg_version(self): self.assertEqual(format_pg_version(100001), '10.1') self.assertEqual(format_pg_version(90605), '9.6.5') - @patch('patroni.ctl.get_dcs') - def test_get_members(self, mock_get_dcs): - mock_get_dcs.return_value = self.e - mock_get_dcs.return_value.get_cluster = get_cluster_not_initialized_without_leader - result = self.runner.invoke(ctl, ['reinit', 'dummy']) - assert "cluster doesn\'t have any members" in result.output + def test_get_members(self): + with patch('patroni.dcs.AbstractDCS.get_cluster', + Mock(return_value=get_cluster_not_initialized_without_leader())): + result = self.runner.invoke(ctl, ['reinit', 'dummy']) + assert "cluster doesn\'t have any members" in result.output @patch('time.sleep', Mock()) - @patch('patroni.ctl.get_dcs') - def test_reinit_wait(self, mock_get_dcs): - mock_get_dcs.return_value.get_cluster = get_cluster_initialized_with_leader + def test_reinit_wait(self): with patch.object(PoolManager, 'request') as mocked: mocked.side_effect = [Mock(data=s, status=200) for s in [b"reinitialize", b'{"state":"creating replica"}', b'{"state":"running"}']]