From e2aff13d3e846d36f4608346cbc04cdf0d295545 Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Mon, 26 Oct 2015 13:57:16 +0100 Subject: [PATCH 01/14] Patronictl: Create commandline tool that can manage clusters. For managing Patroni clusters, the Patroni api can be used. For many tasks, a command line interface for this api would be a useful addition. This commit adds patroncli (The name is still under debate). The command line interface needs access to the DCS; this is required for any operation. For some tasks it is required to have access to the Patroni api. A small summary of the additions to get the cli/ctl started: * Updated Docker image to use 'true' as the archive_command, to ensure disk not filling up during failover testing. * The cli currently can list members, failover a master and remove a given cluster from DCS. * The cli can be configured with a command, for repeated access to the same DCS * Added some simple tests for the cli, code coverage is very low --- docker/entrypoint.sh | 2 +- patroni/cli.py | 323 +++++++++++++++++++++++++++++++++++++++++++ patroni/etcd.py | 2 +- patronicli.py | 5 + tests/test_cli.py | 51 +++++++ tests/test_ha.py | 4 +- 6 files changed, 383 insertions(+), 4 deletions(-) create mode 100644 patroni/cli.py create mode 100755 patronicli.py create mode 100644 tests/test_cli.py diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index 7afdf9c5..1718fb31 100755 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -116,7 +116,7 @@ postgresql: parameters: archive_mode: "on" wal_level: hot_standby - archive_command: mkdir -p ../wal_archive && cp %p ../wal_archive/%f + archive_command: 'true' max_wal_senders: 20 listen_addresses: 0.0.0.0 wal_keep_segments: 8 diff --git a/patroni/cli.py b/patroni/cli.py new file mode 100644 index 00000000..f5aa1166 --- /dev/null +++ b/patroni/cli.py @@ -0,0 +1,323 @@ +#!/usr/bin/env python3 +''' +Patroni Command Line Client +''' + +import click +import os +import yaml +import json +import time +import requests +import datetime +from prettytable import PrettyTable +from six.moves.urllib_parse import urlparse +import logging + +from .etcd import Etcd + +CONFIG_DIR_PATH = click.get_app_dir('patroni') +CONFIG_FILE_PATH = os.path.join(CONFIG_DIR_PATH, 'patronicli.yaml') +LOGLEVEL = 'DEBUG' + + +def parse_dcs(dcs): + """ + Break up the provided dcs string + >>> parse_dcs('localhost') == {'scheme': 'etcd', 'hostname': 'localhost', 'port': 4001} + True + >>> parse_dcs('localhost:8500') == {'scheme': 'consul', 'hostname': 'localhost', 'port': 8500} + True + >>> parse_dcs('zookeeper://localhost') == {'scheme': 'zookeeper', 'hostname': 'localhost', 'port': 2181} + True + """ + + if not dcs: + return {} + + parsed = urlparse(dcs) + scheme = parsed.scheme + if scheme == '' and parsed.netloc == '': + parsed = urlparse('//'+dcs) + + if scheme == '': + default_schemes = {'2181': 'zookeeper', '8500': 'consul'} + scheme = default_schemes.get(str(parsed.port), 'etcd') + + port = parsed.port + if port is None: + default_ports = {'consul': 8500, 'zookeeper': 2181} + port = default_ports.get(str(scheme), 4001) + + return {'scheme': str(scheme), 'hostname': str(parsed.hostname), 'port': int(port)} + + +def load_config(path, dcs): + logging.debug('Loading configuration from file {}'.format(path)) + config = dict() + try: + with open(path, 'rb') as fd: + config = yaml.safe_load(fd) + except: + logging.exception('Could not load configuration file') + + if dcs: + config['dcs'] = parse_dcs(dcs) + else: + config['dcs'] = parse_dcs(config.get('dcs_api')) + + return config + + +def store_config(config, path): + dir_path = os.path.dirname(path) + if dir_path: + if not os.path.isdir(dir_path): + os.makedirs(dir_path) + with open(path, 'w') as fd: + yaml.dump(config, fd) + +option_config_file = click.option('--config-file', '-c', help='Configuration file', default=CONFIG_FILE_PATH) +option_format = click.option('--format', '-f', help='Output format (pretty, json)', default='pretty') +option_dcs = click.option('--dcs', '-d', help='Use this DCS', envvar='DCS') +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') + + +@click.group() +@click.pass_context +def cli(ctx): + logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=LOGLEVEL) + + +def get_dcs(config, scope): + scheme, hostname, port = map(config.get('dcs', {}).get, ('scheme', 'hostname', 'port')) + + if scheme == 'etcd': + return Etcd(name=scope, config={'scope': scope, 'host': '{}:{}'.format(hostname, port)}) + + raise Exception('Can not find suitable configuration of distributed configuration store') + + +def post_patroni(member, endpoint, content, headers={'Content-Type': 'application/json'}): + url = urlparse(member.api_url) + logging.debug(url) + r = requests.post('{}://{}/{}'.format(url.scheme, url.netloc, endpoint), headers=headers, data=json.dumps(content)) + return r + + +def print_output(columns, rows=[], alignment=None, format='pretty'): + if format == 'pretty': + t = PrettyTable(columns) + for k, v in (alignment or {}).items(): + t.align[k] = v + for r in rows: + t.add_row(r) + print(t) + return + + if format == 'json': + elements = list() + for r in rows: + elements.append(dict(zip(columns, r))) + + print(json.dumps(elements)) + + +def watching(w, watch): + if w and not watch: + watch = 2 + if watch: + click.clear() + yield 0 + if watch: + while True: + time.sleep(watch) + click.clear() + yield 0 + + +@cli.command('remove', help='Remove cluster from DCS') +@click.argument('cluster_name') +@option_config_file +@option_format +@option_dcs +def remove(config_file, cluster_name, format, dcs): + config = load_config(config_file, dcs) + dcs = get_dcs(config, cluster_name) + cluster = dcs.get_cluster() + + output_members(cluster, format=format) + + if cluster.name is None: + raise Exception("This does not seem to be a valid Patroni cluster") + + confirm = click.prompt('Please confirm the cluster name to remove', type=str) + if confirm != cluster_name: + raise Exception("Cluster names specified do not match") + + message = 'Yes I am aware' + confirm = click.prompt('You are about to remove all information in DCS for {}, please type: "{}"'.format( + cluster_name, message), type=str) + if message != confirm: + raise Exception('You did not exactly type "{}"'.format(message)) + + if cluster.leader: + confirm = click.prompt('This cluster currently is healthy. Please specify the master name to continue') + if confirm != cluster.leader.name: + raise Exception("You did not specify the current master of the cluster") + + if isinstance(dcs, Etcd): + dcs.client.delete(dcs._base_path, recursive=True) + else: + raise Exception("We have not implemented this for DCS of type {}", type(dcs)) + + +def wait_for_master(dcs, timeout=30): + t_stop = time.time() + timeout + timeout /= 2 + + while time.time() < t_stop: + dcs.watch(timeout) + cluster = dcs.get_cluster() + + if cluster.leader and cluster.leader.member.data['role'] == 'master': + return cluster + + raise Exception('Timeout occured') + + +@cli.command('failover', help='Failover to a replica') +@click.argument('cluster_name') +@click.option('--master', help='The name of the current master', default=None) +@click.option('--candidate', help='The name of the candidate', default=None) +@click.option('--force', is_flag=True) +@option_config_file +@option_dcs +def failover(config_file, cluster_name, master, candidate, force, dcs): + """ + We want to trigger a failover for the specified cluster name. + + We verify that the cluster name, master name and candidate name are correct. + If so, we trigger a failover and keep the client up to date. + """ + config = load_config(config_file, dcs) + dcs = get_dcs(config, cluster_name) + cluster = dcs.get_cluster() + + if cluster.leader is None: + raise Exception('This cluster has no master') + + if master is None: + if force: + master = cluster.leader.member.name + else: + master = click.prompt('Master', type=str, default=cluster.leader.member.name) + + if cluster.leader.member.name != master: + raise Exception('Member {} is not the leader of cluster {}'.format(master, cluster_name)) + + candidate_names = [str(m.name) for m in cluster.members if m.name != master] + candidate_names.sort() + + if candidate is None and not force: + candidate = click.prompt('Candidate '+str(candidate_names), type=str, default='') + + if candidate and candidate not in candidate_names: + raise Exception('Member {} does not exist in cluster {}'.format(candidate, cluster_name)) + + # By now we have established that the leader exists and the candidate exists + click.echo('Current cluster topology') + output_members(dcs.get_cluster(), name=cluster_name) + + if not force: + a = click.confirm('Are you sure you want to failover cluster {}, demoting current master {}?'.format( + cluster_name, master)) + if not a: + raise Exception('Aborting failover') + + failover_value = '{}:{}'.format(master, (candidate or '')) + + t_started = time.time() + try: + r = post_patroni(cluster.leader.member, 'failover', {'leader': master, 'candidate': (candidate or '')}) + if r.status_code == 200: + logging.debug(r) + logging.debug(r.text) + cluster = dcs.get_cluster() + click.echo(timestamp()+' Failing over to new leader: {}'.format(cluster.leader.member.name)) + else: + click.echo('Failover failed, details: {}, {}'.format(r.status_code, r.text)) + return + except: + logging.exception(r) + logging.warning('Failing over to DCS') + click.echo(timestamp()+' Could not failover using Patroni api, falling back to DCS') + dcs.set_failover_value(failover_value) + click.echo(timestamp()+' Initialized failover from master {}'.format(master)) + + # The failover process should within a minute update the failover key, we will keep watching it until it changes + # or we timeout + cluster = wait_for_master(dcs, timeout=60) + click.echo(timestamp()+' Failover completed in {:0.1f} seconds, new leader is {}'.format( + time.time() - t_started, str(cluster.leader.member.name))) + output_members(cluster, name=cluster_name) + + +def output_members(cluster, name=None, format='pretty'): + rows = [] + logging.debug(cluster) + leader_name = None + if cluster.leader: + leader_name = cluster.leader.member.name + + # Mainly for consistent pretty printing and watching we sort the output + cluster.members.sort(key=lambda x: x.name) + for m in cluster.members: + logging.debug(m) + + leader = '' + if m.name == leader_name: + leader = '*' + role = m.data['role'] + else: + role = 'replica' + + rows.append([name, m.name, role, leader]) + + print_output(['Cluster', 'Member', 'Role', 'Leader'], rows, {'Cluster': 'l', 'Member': 'l', 'Role': 'l'}, format) + + +@cli.command('list', help='List the Patroni members for a given Patroni') +@click.argument('cluster_names', nargs=-1) +@option_config_file +@option_format +@option_watch +@option_watchrefresh +@option_dcs +def members(config_file, cluster_names, format, watch, w, dcs): + if len(cluster_names) == 0: + logging.warning('Listing members: No cluster names were provided') + return + + config = load_config(config_file, dcs) + for cn in cluster_names: + dcs = get_dcs(config, cn) + + for _ in watching(w, watch): + output_members(dcs.get_cluster(), name=cn, format=format) + + +def timestamp(): + return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f") + + +@cli.command('configure', help='Create configuration file') +@click.option('--config-file', '-c', help='Configuration file', prompt='Configuration file', default=CONFIG_FILE_PATH) +@click.option('--dcs', '-d', help='The DCS connect url', prompt='DCS connect url', default='etcd://localhost:4001') +@click.option('--namespace', '-n', help='The namespace', prompt='Namespace', default='/service/') +def configure(config_file, dcs, namespace): + config = dict() + config['dcs_api'] = str(dcs) + config['namespace'] = str(namespace) + store_config(config, config_file) diff --git a/patroni/etcd.py b/patroni/etcd.py index 4a82f2d7..d12cbf93 100644 --- a/patroni/etcd.py +++ b/patroni/etcd.py @@ -150,7 +150,7 @@ class Etcd(AbstractDCS): def __init__(self, name, config): super(Etcd, self).__init__(name, config) - self.ttl = config['ttl'] + self.ttl = config.get('ttl', 30) self._retry = Retry(deadline=10, max_delay=1, max_tries=-1, retry_exceptions=(etcd.EtcdConnectionFailed, etcd.EtcdLeaderElectionInProgress, diff --git a/patronicli.py b/patronicli.py new file mode 100755 index 00000000..3e7afb05 --- /dev/null +++ b/patronicli.py @@ -0,0 +1,5 @@ +#!/usr/bin/env python +from patroni.cli import cli + +if __name__ == '__main__': + cli() diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 00000000..1f5f77c2 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,51 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +import os +import pytest + +from click.testing import CliRunner +from patroni.cli import cli, members, store_config, load_config, output_members +from test_ha import get_cluster_initialized_with_leader + +CONFIG_FILE_PATH = './test-cli.yaml' + + +def test_output_members(): + cluster = get_cluster_initialized_with_leader() + output_members(cluster, name='abc', format='pretty') + output_members(cluster, name='abc', format='json') + + +def test_rw_config(): + runner = CliRunner() + config = 'a:b' + with runner.isolated_filesystem(): + os.mkdir(CONFIG_FILE_PATH) + with pytest.raises(Exception): + result = load_config(CONFIG_FILE_PATH, None) + assert 'Could not load configuration file' in result.output + + with pytest.raises(Exception): + store_config(config, CONFIG_FILE_PATH) + os.rmdir(CONFIG_FILE_PATH) + + store_config(config, 'abc/CONFIG_FILE_PATH') + load_config(CONFIG_FILE_PATH, None) + + +def test_cli(): + runner = CliRunner() + + runner.invoke(cli, ['list']) + + result = runner.invoke(cli, ['--help']) + assert 'Usage:' in result.output + + +def test_members(): + runner = CliRunner() + + runner.invoke(members) + + diff --git a/tests/test_ha.py b/tests/test_ha.py index e34f9b8e..b775f4ee 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -27,10 +27,10 @@ def get_cluster_not_initialized_without_leader(): def get_cluster_initialized_without_leader(leader=False, failover=None): m = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', - 'api_url': 'http://127.0.0.1:8008/patroni'}) + 'api_url': 'http://127.0.0.1:8008/patroni', 'role':'replica'}) l = Leader(0, 0, m) if leader else None o = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres', - 'api_url': 'http://127.0.0.1:8011/patroni'}) + 'api_url': 'http://127.0.0.1:8011/patroni', 'role':'replica'}) return get_cluster(True, l, [m, o], failover) From a2cb3f18898cf44d7a38d70aa93c607ba6d3a2c7 Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Mon, 26 Oct 2015 15:04:23 +0100 Subject: [PATCH 02/14] Include Click as requirement for patroncli --- requirements-py2.txt | 1 + requirements-py3.txt | 1 + 2 files changed, 2 insertions(+) diff --git a/requirements-py2.txt b/requirements-py2.txt index fde9c79a..517ec064 100644 --- a/requirements-py2.txt +++ b/requirements-py2.txt @@ -7,3 +7,4 @@ requests six >= 1.7 kazoo>=2.2.1 python-etcd>=0.4.1 +click>=4.1 diff --git a/requirements-py3.txt b/requirements-py3.txt index cc00965b..b04a0f74 100644 --- a/requirements-py3.txt +++ b/requirements-py3.txt @@ -7,3 +7,4 @@ requests six kazoo>=2.2.1 python-etcd>=0.4.1 +click>=4.1 From 98a0d8381b5f8d4a72d834b8e4c66770e7e54e64 Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Mon, 26 Oct 2015 15:07:08 +0100 Subject: [PATCH 03/14] Add prettytable to requirements --- requirements-py2.txt | 1 + requirements-py3.txt | 1 + 2 files changed, 2 insertions(+) diff --git a/requirements-py2.txt b/requirements-py2.txt index 517ec064..f8eacc12 100644 --- a/requirements-py2.txt +++ b/requirements-py2.txt @@ -8,3 +8,4 @@ six >= 1.7 kazoo>=2.2.1 python-etcd>=0.4.1 click>=4.1 +prettytable>=0.7 diff --git a/requirements-py3.txt b/requirements-py3.txt index b04a0f74..30b5ce96 100644 --- a/requirements-py3.txt +++ b/requirements-py3.txt @@ -8,3 +8,4 @@ six kazoo>=2.2.1 python-etcd>=0.4.1 click>=4.1 +prettytable>=0.7 From 39383598287e427e58a98ccf97016bee2a9b41eb Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Tue, 27 Oct 2015 12:13:26 +0100 Subject: [PATCH 04/14] Command Line: Add reinit and restart commands. --- patroni/cli.py | 51 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 51 insertions(+) diff --git a/patroni/cli.py b/patroni/cli.py index f5aa1166..abd3b815 100644 --- a/patroni/cli.py +++ b/patroni/cli.py @@ -82,6 +82,7 @@ option_format = click.option('--format', '-f', help='Output format (pretty, json option_dcs = click.option('--dcs', '-d', help='Use this DCS', envvar='DCS') 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') @click.group() @@ -187,6 +188,56 @@ def wait_for_master(dcs, timeout=30): raise Exception('Timeout occured') +def empty_post_to_members(cluster_name, member_names, config_file, dcs, force, endpoint): + config = load_config(config_file, dcs) + dcs = get_dcs(config, cluster_name) + cluster = dcs.get_cluster() + + candidates = dict() + for m in cluster.members: + candidates[m.name] = m + + if len(member_names) == 0: + member_names = [ + click.prompt('Which member do you want to '+endpoint + ' '+str(candidates.keys()), type=str, default='')] + + for mn in member_names: + if mn not in candidates.keys(): + raise Exception('{} is not a member of cluster {}'.format(mn, cluster_name)) + + if not force: + confirm = click.confirm('Are you sure you want to {} members {}?'.format(endpoint, str(candidates.keys()))) + if not confirm: + raise Exception('Aborted {}'.format(endpoint)) + + for mn in member_names: + r = post_patroni(candidates[mn], endpoint, '') + if r.status_code != 200: + click.echo('{} failed for member {}, status code={}, ({})'.format(endpoint, mn, r.status_code, r.text)) + else: + click.echo('Succesful {} on member {}'.format(endpoint, mn)) + + +@cli.command('restart', help='Restart cluster member') +@click.argument('cluster_name') +@click.argument('member_names', nargs=-1) +@option_config_file +@option_force +@option_dcs +def restart(cluster_name, member_names, config_file, dcs, force): + empty_post_to_members(cluster_name, member_names, config_file, dcs, force, 'restart') + + +@cli.command('reinit', help='Reinitialize cluster member') +@click.argument('cluster_name') +@click.argument('member_names', nargs=-1) +@option_config_file +@option_force +@option_dcs +def reinit(cluster_name, member_names, config_file, dcs, force): + empty_post_to_members(cluster_name, member_names, config_file, dcs, force, 'reinitialize') + + @cli.command('failover', help='Failover to a replica') @click.argument('cluster_name') @click.option('--master', help='The name of the current master', default=None) From da23dd12f3ec7fd9ee26eaaa3dd7dee525dd5965 Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Fri, 6 Nov 2015 14:59:48 +0100 Subject: [PATCH 05/14] Do not wait for leader key to change after failover. Previously, the leader key was watched for changes after a failover. This resulted in a delay of up to 10 seconds to report a healthy failover back to the client. With this patch, we are not relying on the role of a member registered in the dcs anymore. --- patroni/cli.py | 21 +++++++++++---------- tests/test_ha.py | 4 ++-- 2 files changed, 13 insertions(+), 12 deletions(-) diff --git a/patroni/cli.py b/patroni/cli.py index abd3b815..f11d0587 100644 --- a/patroni/cli.py +++ b/patroni/cli.py @@ -1,4 +1,3 @@ -#!/usr/bin/env python3 ''' Patroni Command Line Client ''' @@ -182,7 +181,7 @@ def wait_for_master(dcs, timeout=30): dcs.watch(timeout) cluster = dcs.get_cluster() - if cluster.leader and cluster.leader.member.data['role'] == 'master': + if cluster.leader: return cluster raise Exception('Timeout occured') @@ -269,8 +268,13 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): raise Exception('Member {} is not the leader of cluster {}'.format(master, cluster_name)) candidate_names = [str(m.name) for m in cluster.members if m.name != master] + ## We sort the names for consistent output to the client candidate_names.sort() + if len(candidate_names) == 0: + raise Exception('No candidates found to failover to') + + if candidate is None and not force: candidate = click.prompt('Candidate '+str(candidate_names), type=str, default='') @@ -306,10 +310,10 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): click.echo(timestamp()+' Could not failover using Patroni api, falling back to DCS') dcs.set_failover_value(failover_value) click.echo(timestamp()+' Initialized failover from master {}'.format(master)) + # The failover process should within a minute update the failover key, we will keep watching it until it changes + # or we timeout + cluster = wait_for_master(dcs, timeout=60) - # The failover process should within a minute update the failover key, we will keep watching it until it changes - # or we timeout - cluster = wait_for_master(dcs, timeout=60) click.echo(timestamp()+' Failover completed in {:0.1f} seconds, new leader is {}'.format( time.time() - t_started, str(cluster.leader.member.name))) output_members(cluster, name=cluster_name) @@ -330,13 +334,10 @@ def output_members(cluster, name=None, format='pretty'): leader = '' if m.name == leader_name: leader = '*' - role = m.data['role'] - else: - role = 'replica' - rows.append([name, m.name, role, leader]) + rows.append([name, m.name, leader]) - print_output(['Cluster', 'Member', 'Role', 'Leader'], rows, {'Cluster': 'l', 'Member': 'l', 'Role': 'l'}, format) + print_output(['Cluster', 'Member', 'Leader'], rows, {'Cluster': 'l', 'Member': 'l'}, format) @cli.command('list', help='List the Patroni members for a given Patroni') diff --git a/tests/test_ha.py b/tests/test_ha.py index b775f4ee..e34f9b8e 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -27,10 +27,10 @@ def get_cluster_not_initialized_without_leader(): def get_cluster_initialized_without_leader(leader=False, failover=None): m = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', - 'api_url': 'http://127.0.0.1:8008/patroni', 'role':'replica'}) + 'api_url': 'http://127.0.0.1:8008/patroni'}) l = Leader(0, 0, m) if leader else None o = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres', - 'api_url': 'http://127.0.0.1:8011/patroni', 'role':'replica'}) + 'api_url': 'http://127.0.0.1:8011/patroni'}) return get_cluster(True, l, [m, o], failover) From dcb5113f9d937db498847cea2f0ad92b301d43cc Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Fri, 13 Nov 2015 12:55:38 +0100 Subject: [PATCH 06/14] Refactoring from patronicli to patronictl --- Dockerfile | 5 ++-- patroni/{cli.py => ctl.py} | 38 +++++++++++++++++------------- patronicli.py => patronictl.py | 4 ++-- tests/{test_cli.py => test_ctl.py} | 10 ++++---- 4 files changed, 32 insertions(+), 25 deletions(-) rename patroni/{cli.py => ctl.py} (92%) rename patronicli.py => patronictl.py (56%) rename tests/{test_cli.py => test_ctl.py} (84%) diff --git a/Dockerfile b/Dockerfile index 61d85a74..84b9cdf6 100644 --- a/Dockerfile +++ b/Dockerfile @@ -13,8 +13,9 @@ RUN apt-get update -y RUN apt-get upgrade -y ENV PGVERSION 9.4 -RUN apt-get install python python-psycopg2 python-yaml python-requests python-boto postgresql-${PGVERSION} python-dnspython python-kazoo python-pip -y -RUN pip install python-etcd +RUN apt-get install python python-yaml python-requests python-boto postgresql-${PGVERSION} python-dnspython python-kazoo python-pip -y +RUN apt-get install python-dev postgresql-server-dev-${PGVERSION} -y +RUN pip install python-etcd psycopg2 ENV PATH /usr/lib/postgresql/${PGVERSION}/bin:$PATH diff --git a/patroni/cli.py b/patroni/ctl.py similarity index 92% rename from patroni/cli.py rename to patroni/ctl.py index f11d0587..ed24ef93 100644 --- a/patroni/cli.py +++ b/patroni/ctl.py @@ -1,5 +1,5 @@ ''' -Patroni Command Line Client +Patroni Control ''' import click @@ -16,8 +16,8 @@ import logging from .etcd import Etcd CONFIG_DIR_PATH = click.get_app_dir('patroni') -CONFIG_FILE_PATH = os.path.join(CONFIG_DIR_PATH, 'patronicli.yaml') -LOGLEVEL = 'DEBUG' +CONFIG_FILE_PATH = os.path.join(CONFIG_DIR_PATH, 'patronictl.yaml') +LOGLEVEL = 'INFO' def parse_dcs(dcs): @@ -86,7 +86,7 @@ option_force = click.option('--force', is_flag=True, help='Do not ask for confir @click.group() @click.pass_context -def cli(ctx): +def ctl(ctx): logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=LOGLEVEL) @@ -99,6 +99,13 @@ def get_dcs(config, scope): raise Exception('Can not find suitable configuration of distributed configuration store') +def get_patroni(member, endpoint): + url = urlparse(member.api_url) + logging.debug(url) + r = requests.get('{}://{}/{}'.format(url.scheme, url.netloc, endpoint)) + return r + + def post_patroni(member, endpoint, content, headers={'Content-Type': 'application/json'}): url = urlparse(member.api_url) logging.debug(url) @@ -137,7 +144,7 @@ def watching(w, watch): yield 0 -@cli.command('remove', help='Remove cluster from DCS') +@ctl.command('remove', help='Remove cluster from DCS') @click.argument('cluster_name') @option_config_file @option_format @@ -168,12 +175,12 @@ def remove(config_file, cluster_name, format, dcs): raise Exception("You did not specify the current master of the cluster") if isinstance(dcs, Etcd): - dcs.client.delete(dcs._base_path, recursive=True) + dcs.ctlent.delete(dcs._base_path, recursive=True) else: raise Exception("We have not implemented this for DCS of type {}", type(dcs)) -def wait_for_master(dcs, timeout=30): +def wait_for_leader(dcs, timeout=30): t_stop = time.time() + timeout timeout /= 2 @@ -217,7 +224,7 @@ def empty_post_to_members(cluster_name, member_names, config_file, dcs, force, e click.echo('Succesful {} on member {}'.format(endpoint, mn)) -@cli.command('restart', help='Restart cluster member') +@ctl.command('restart', help='Restart cluster member') @click.argument('cluster_name') @click.argument('member_names', nargs=-1) @option_config_file @@ -227,7 +234,7 @@ def restart(cluster_name, member_names, config_file, dcs, force): empty_post_to_members(cluster_name, member_names, config_file, dcs, force, 'restart') -@cli.command('reinit', help='Reinitialize cluster member') +@ctl.command('reinit', help='Reinitialize cluster member') @click.argument('cluster_name') @click.argument('member_names', nargs=-1) @option_config_file @@ -237,7 +244,7 @@ def reinit(cluster_name, member_names, config_file, dcs, force): empty_post_to_members(cluster_name, member_names, config_file, dcs, force, 'reinitialize') -@cli.command('failover', help='Failover to a replica') +@ctl.command('failover', help='Failover to a replica') @click.argument('cluster_name') @click.option('--master', help='The name of the current master', default=None) @click.option('--candidate', help='The name of the candidate', default=None) @@ -249,7 +256,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): We want to trigger a failover for the specified cluster name. We verify that the cluster name, master name and candidate name are correct. - If so, we trigger a failover and keep the client up to date. + If so, we trigger a failover and keep the ctlent up to date. """ config = load_config(config_file, dcs) dcs = get_dcs(config, cluster_name) @@ -268,13 +275,12 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): raise Exception('Member {} is not the leader of cluster {}'.format(master, cluster_name)) candidate_names = [str(m.name) for m in cluster.members if m.name != master] - ## We sort the names for consistent output to the client + # We sort the names for consistent output to the ctlent candidate_names.sort() if len(candidate_names) == 0: raise Exception('No candidates found to failover to') - if candidate is None and not force: candidate = click.prompt('Candidate '+str(candidate_names), type=str, default='') @@ -312,7 +318,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): click.echo(timestamp()+' Initialized failover from master {}'.format(master)) # The failover process should within a minute update the failover key, we will keep watching it until it changes # or we timeout - cluster = wait_for_master(dcs, timeout=60) + cluster = wait_for_leader(dcs, timeout=60) click.echo(timestamp()+' Failover completed in {:0.1f} seconds, new leader is {}'.format( time.time() - t_started, str(cluster.leader.member.name))) @@ -340,7 +346,7 @@ def output_members(cluster, name=None, format='pretty'): print_output(['Cluster', 'Member', 'Leader'], rows, {'Cluster': 'l', 'Member': 'l'}, format) -@cli.command('list', help='List the Patroni members for a given Patroni') +@ctl.command('list', help='List the Patroni members for a given Patroni') @click.argument('cluster_names', nargs=-1) @option_config_file @option_format @@ -364,7 +370,7 @@ def timestamp(): return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f") -@cli.command('configure', help='Create configuration file') +@ctl.command('configure', help='Create configuration file') @click.option('--config-file', '-c', help='Configuration file', prompt='Configuration file', default=CONFIG_FILE_PATH) @click.option('--dcs', '-d', help='The DCS connect url', prompt='DCS connect url', default='etcd://localhost:4001') @click.option('--namespace', '-n', help='The namespace', prompt='Namespace', default='/service/') diff --git a/patronicli.py b/patronictl.py similarity index 56% rename from patronicli.py rename to patronictl.py index 3e7afb05..5b06c153 100755 --- a/patronicli.py +++ b/patronictl.py @@ -1,5 +1,5 @@ #!/usr/bin/env python -from patroni.cli import cli +from patroni.ctl import ctl if __name__ == '__main__': - cli() + ctl() diff --git a/tests/test_cli.py b/tests/test_ctl.py similarity index 84% rename from tests/test_cli.py rename to tests/test_ctl.py index 1f5f77c2..2c732de7 100644 --- a/tests/test_cli.py +++ b/tests/test_ctl.py @@ -5,10 +5,10 @@ import os import pytest from click.testing import CliRunner -from patroni.cli import cli, members, store_config, load_config, output_members +from patroni.ctl import ctl, members, store_config, load_config, output_members from test_ha import get_cluster_initialized_with_leader -CONFIG_FILE_PATH = './test-cli.yaml' +CONFIG_FILE_PATH = './test-ctl.yaml' def test_output_members(): @@ -34,12 +34,12 @@ def test_rw_config(): load_config(CONFIG_FILE_PATH, None) -def test_cli(): +def test_ctl(): runner = CliRunner() - runner.invoke(cli, ['list']) + runner.invoke(ctl, ['list']) - result = runner.invoke(cli, ['--help']) + result = runner.invoke(ctl, ['--help']) assert 'Usage:' in result.output From 990276c2140b85829a006ce3bc9f2f4d91043401 Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Fri, 13 Nov 2015 13:02:40 +0100 Subject: [PATCH 07/14] Install patronictl as a script --- setup.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/setup.py b/setup.py index 724f3b4a..afd7a498 100644 --- a/setup.py +++ b/setup.py @@ -56,7 +56,7 @@ CLASSIFIERS = [ 'Programming Language :: Python :: Implementation :: CPython', ] -CONSOLE_SCRIPTS = ['patroni = patroni:main'] +CONSOLE_SCRIPTS = ['patroni = patroni:main', 'patronictl = patroni.ctl:ctl'] class PyTest(TestCommand): From e92041611460247f4c27b5cef5bc15cfdc3ede1f Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Mon, 16 Nov 2015 12:32:01 +0100 Subject: [PATCH 08/14] Patronictl: Extend test cases and create own Exception class. --- patroni/ctl.py | 113 ++++++++++++++++++++--------------- patroni/exceptions.py | 4 ++ tests/test_ctl.py | 136 ++++++++++++++++++++++++++++++++---------- 3 files changed, 174 insertions(+), 79 deletions(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index ed24ef93..490ee607 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -14,6 +14,7 @@ from six.moves.urllib_parse import urlparse import logging from .etcd import Etcd +from .exceptions import PatroniCtlException CONFIG_DIR_PATH = click.get_app_dir('patroni') CONFIG_FILE_PATH = os.path.join(CONFIG_DIR_PATH, 'patronictl.yaml') @@ -91,26 +92,25 @@ def ctl(ctx): def get_dcs(config, scope): + """ + >>> get_dcs({'scheme':'redis', 'hostname':'a', 'port':'1'}, 'testing') + Traceback (most recent call last): + ... + patroni.exceptions.PatroniCtlException: Can not find suitable configuration of distributed configuration store + """ scheme, hostname, port = map(config.get('dcs', {}).get, ('scheme', 'hostname', 'port')) if scheme == 'etcd': return Etcd(name=scope, config={'scope': scope, 'host': '{}:{}'.format(hostname, port)}) - raise Exception('Can not find suitable configuration of distributed configuration store') - - -def get_patroni(member, endpoint): - url = urlparse(member.api_url) - logging.debug(url) - r = requests.get('{}://{}/{}'.format(url.scheme, url.netloc, endpoint)) - return r + raise PatroniCtlException('Can not find suitable configuration of distributed configuration store') def post_patroni(member, endpoint, content, headers={'Content-Type': 'application/json'}): url = urlparse(member.api_url) logging.debug(url) - r = requests.post('{}://{}/{}'.format(url.scheme, url.netloc, endpoint), headers=headers, data=json.dumps(content)) - return r + return requests.post('{}://{}/{}'.format( + url.scheme, url.netloc, endpoint), headers=headers, data=json.dumps(content)) def print_output(columns, rows=[], alignment=None, format='pretty'): @@ -131,17 +131,28 @@ def print_output(columns, rows=[], alignment=None, format='pretty'): print(json.dumps(elements)) -def watching(w, watch): +def watching(w, watch, max_count=None): + """ + >>> len(list(watching(True, 1, 0))) + 1 + >>> len(list(watching(True, 1, 1))) + 2 + """ if w and not watch: watch = 2 if watch: click.clear() yield 0 - if watch: - while True: - time.sleep(watch) - click.clear() - yield 0 + + if max_count is not None and max_count < 1: + return + + counter = 1 + while watch and counter <= (max_count or counter): + time.sleep(watch) + counter += 1 + click.clear() + yield 0 @ctl.command('remove', help='Remove cluster from DCS') @@ -150,34 +161,29 @@ def watching(w, watch): @option_format @option_dcs def remove(config_file, cluster_name, format, dcs): - config = load_config(config_file, dcs) - dcs = get_dcs(config, cluster_name) - cluster = dcs.get_cluster() + config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) output_members(cluster, format=format) - if cluster.name is None: - raise Exception("This does not seem to be a valid Patroni cluster") - confirm = click.prompt('Please confirm the cluster name to remove', type=str) if confirm != cluster_name: - raise Exception("Cluster names specified do not match") + raise PatroniCtlException("Cluster names specified do not match") message = 'Yes I am aware' confirm = click.prompt('You are about to remove all information in DCS for {}, please type: "{}"'.format( cluster_name, message), type=str) if message != confirm: - raise Exception('You did not exactly type "{}"'.format(message)) + raise PatroniCtlException('You did not exactly type "{}"'.format(message)) if cluster.leader: confirm = click.prompt('This cluster currently is healthy. Please specify the master name to continue') if confirm != cluster.leader.name: - raise Exception("You did not specify the current master of the cluster") + raise PatroniCtlException("You did not specify the current master of the cluster") if isinstance(dcs, Etcd): - dcs.ctlent.delete(dcs._base_path, recursive=True) + dcs.client.delete(dcs._base_path, recursive=True) else: - raise Exception("We have not implemented this for DCS of type {}", type(dcs)) + raise PatroniCtlException("We have not implemented this for DCS of type {}", type(dcs)) def wait_for_leader(dcs, timeout=30): @@ -191,30 +197,27 @@ def wait_for_leader(dcs, timeout=30): if cluster.leader: return cluster - raise Exception('Timeout occured') + raise PatroniCtlException('Timeout occured') -def empty_post_to_members(cluster_name, member_names, config_file, dcs, force, endpoint): - config = load_config(config_file, dcs) - dcs = get_dcs(config, cluster_name) - cluster = dcs.get_cluster() - +def empty_post_to_members(cluster, member_names, force, endpoint): candidates = dict() for m in cluster.members: candidates[m.name] = m if len(member_names) == 0: member_names = [ - click.prompt('Which member do you want to '+endpoint + ' '+str(candidates.keys()), type=str, default='')] + click.prompt('Which member do you want to {} [{}]?'.format( + endpoint, ', '.join(candidates.keys())), type=str, default='')] for mn in member_names: if mn not in candidates.keys(): - raise Exception('{} is not a member of cluster {}'.format(mn, cluster_name)) + raise PatroniCtlException('{} is not a member of cluster'.format(mn)) if not force: - confirm = click.confirm('Are you sure you want to {} members {}?'.format(endpoint, str(candidates.keys()))) + confirm = click.confirm('Are you sure you want to {} members {}?'.format(endpoint, ', '.join(member_names))) if not confirm: - raise Exception('Aborted {}'.format(endpoint)) + raise PatroniCtlException('Aborted {}'.format(endpoint)) for mn in member_names: r = post_patroni(candidates[mn], endpoint, '') @@ -224,6 +227,14 @@ def empty_post_to_members(cluster_name, member_names, config_file, dcs, force, e click.echo('Succesful {} on member {}'.format(endpoint, mn)) +def ctl_load_config(cluster_name, config_file, dcs): + config = load_config(config_file, dcs) + dcs = get_dcs(config, cluster_name) + cluster = dcs.get_cluster() + + return (config, dcs, cluster) + + @ctl.command('restart', help='Restart cluster member') @click.argument('cluster_name') @click.argument('member_names', nargs=-1) @@ -231,7 +242,8 @@ def empty_post_to_members(cluster_name, member_names, config_file, dcs, force, e @option_force @option_dcs def restart(cluster_name, member_names, config_file, dcs, force): - empty_post_to_members(cluster_name, member_names, config_file, dcs, force, 'restart') + config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) + empty_post_to_members(cluster, member_names, force, 'restart') @ctl.command('reinit', help='Reinitialize cluster member') @@ -241,7 +253,8 @@ def restart(cluster_name, member_names, config_file, dcs, force): @option_force @option_dcs def reinit(cluster_name, member_names, config_file, dcs, force): - empty_post_to_members(cluster_name, member_names, config_file, dcs, force, 'reinitialize') + config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) + empty_post_to_members(cluster, member_names, force, 'reinitialize') @ctl.command('failover', help='Failover to a replica') @@ -256,14 +269,12 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): We want to trigger a failover for the specified cluster name. We verify that the cluster name, master name and candidate name are correct. - If so, we trigger a failover and keep the ctlent up to date. + If so, we trigger a failover and keep the client up to date. """ - config = load_config(config_file, dcs) - dcs = get_dcs(config, cluster_name) - cluster = dcs.get_cluster() + config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) if cluster.leader is None: - raise Exception('This cluster has no master') + raise PatroniCtlException('This cluster has no master') if master is None: if force: @@ -272,20 +283,20 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): master = click.prompt('Master', type=str, default=cluster.leader.member.name) if cluster.leader.member.name != master: - raise Exception('Member {} is not the leader of cluster {}'.format(master, cluster_name)) + raise PatroniCtlException('Member {} is not the leader of cluster {}'.format(master, cluster_name)) candidate_names = [str(m.name) for m in cluster.members if m.name != master] - # We sort the names for consistent output to the ctlent + # We sort the names for consistent output to the client candidate_names.sort() if len(candidate_names) == 0: - raise Exception('No candidates found to failover to') + raise PatroniCtlException('No candidates found to failover to') if candidate is None and not force: candidate = click.prompt('Candidate '+str(candidate_names), type=str, default='') if candidate and candidate not in candidate_names: - raise Exception('Member {} does not exist in cluster {}'.format(candidate, cluster_name)) + raise PatroniCtlException('Member {} does not exist in cluster {}'.format(candidate, cluster_name)) # By now we have established that the leader exists and the candidate exists click.echo('Current cluster topology') @@ -295,11 +306,12 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): a = click.confirm('Are you sure you want to failover cluster {}, demoting current master {}?'.format( cluster_name, master)) if not a: - raise Exception('Aborting failover') + raise PatroniCtlException('Aborting failover') failover_value = '{}:{}'.format(master, (candidate or '')) t_started = time.time() + r = None try: r = post_patroni(cluster.leader.member, 'failover', {'leader': master, 'candidate': (candidate or '')}) if r.status_code == 200: @@ -319,6 +331,9 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): # The failover process should within a minute update the failover key, we will keep watching it until it changes # or we timeout cluster = wait_for_leader(dcs, timeout=60) + if cluster.leader.member.name == master: + click.echo('Failover failed, master did not change after {:0.1f} seconds'.format(time.time() - t_started)) + return click.echo(timestamp()+' Failover completed in {:0.1f} seconds, new leader is {}'.format( time.time() - t_started, str(cluster.leader.member.name))) diff --git a/patroni/exceptions.py b/patroni/exceptions.py index 97985696..43f54e7f 100644 --- a/patroni/exceptions.py +++ b/patroni/exceptions.py @@ -13,6 +13,10 @@ class PatroniException(Exception): return repr(self.value) +class PatroniCtlException(Exception): + pass + + class PostgresException(PatroniException): pass diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 2c732de7..b0715118 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -1,51 +1,127 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - import os +import mock import pytest +import unittest +import requests +import patroni.exceptions +from mock import patch, Mock from click.testing import CliRunner -from patroni.ctl import ctl, members, store_config, load_config, output_members -from test_ha import get_cluster_initialized_with_leader +from patroni.ctl import ctl, members, store_config, load_config, output_members, post_patroni, get_dcs, wait_for_leader +from patroni.dcs import AbstractDCS, Member +from patroni.etcd import Etcd +from test_ha import get_cluster_initialized_without_leader, get_cluster_initialized_with_leader, get_cluster CONFIG_FILE_PATH = './test-ctl.yaml' +class TestCtl(unittest.TestCase): -def test_output_members(): - cluster = get_cluster_initialized_with_leader() - output_members(cluster, name='abc', format='pretty') - output_members(cluster, name='abc', format='json') + def test_output_members(self): + cluster = get_cluster_initialized_with_leader() + output_members(cluster, name='abc', format='pretty') + output_members(cluster, name='abc', format='json') -def test_rw_config(): - runner = CliRunner() - config = 'a:b' - with runner.isolated_filesystem(): - os.mkdir(CONFIG_FILE_PATH) - with pytest.raises(Exception): - result = load_config(CONFIG_FILE_PATH, None) - assert 'Could not load configuration file' in result.output - - with pytest.raises(Exception): + def test_rw_config(self): + runner = CliRunner() + config = 'a:b' + with runner.isolated_filesystem(): store_config(config, CONFIG_FILE_PATH) - os.rmdir(CONFIG_FILE_PATH) + os.remove(CONFIG_FILE_PATH) + os.mkdir(CONFIG_FILE_PATH) + with pytest.raises(Exception): + result = load_config(CONFIG_FILE_PATH, None) + assert 'Could not load configuration file' in result.output - store_config(config, 'abc/CONFIG_FILE_PATH') - load_config(CONFIG_FILE_PATH, None) + with pytest.raises(Exception): + store_config(config, CONFIG_FILE_PATH) + + os.rmdir(CONFIG_FILE_PATH) + + store_config(config, 'abc/CONFIG_FILE_PATH') + load_config(CONFIG_FILE_PATH, None) + + @patch('patroni.ctl.get_dcs', Mock(return_value=AbstractDCS('dummy',{'namespace':'dummy', 'scope':'dummy'}))) + @patch('patroni.ctl.post_patroni', Mock(return_value=None)) + def test_failover(self): + runner = CliRunner() + + with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_without_leader())): + result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8']) + assert 'This cluster has no master' in str(result.exception) + + with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())): + result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='nonsense') + assert 'is not the leader of cluster' in str(result.exception) + + result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8', '--master', 'nonsense']) + assert 'is not the leader of cluster' in str(result.exception) + + result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nn') + assert 'Aborting failover' in str(result.exception) + + with patch('patroni.ctl.wait_for_leader', Mock(return_value = get_cluster_initialized_with_leader())): + result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nY') + assert 'master did not change after' in result.output + + result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nY') + assert 'Failover failed' in result.output + + def test_get_dcs(self): + self.assertRaises(patroni.exceptions.PatroniCtlException, get_dcs, {'scheme':'dummy'}, 'dummy') -def test_ctl(): - runner = CliRunner() + @patch('patroni.ctl.get_dcs', Mock(return_value=AbstractDCS('dummy',{'namespace':'dummy', 'scope':'dummy'}))) + @patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) + @patch('patroni.etcd.Etcd.get_etcd_client', Mock(return_value=None)) + def test_remove(self): + runner = CliRunner() - runner.invoke(ctl, ['list']) + result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='alpha\nslave') + assert 'Please confirm' in result.output + assert 'You are about to remove all' in result.output + assert 'You did not exactly type' in str(result.exception) - result = runner.invoke(ctl, ['--help']) - assert 'Usage:' in result.output + result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='alpha\nYes I am aware\nleader') + assert 'We have not implemented this for DCS of type' in str(result.exception) + + result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='alpha\nYes I am aware\nslave') + assert 'You did not specify the current master of the cluster' in str(result.exception) + + result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='beta\nleader') + assert 'Cluster names specified do not match' in str(result.exception) + + with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=Etcd('dummy', {'namespace':'dummy', 'scope':'dummy'}))): + result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='alpha\nYes I am aware\nleader') + assert 'object has no attribute' in str(result.exception) + + @patch('patroni.dcs.AbstractDCS.watch', Mock(return_value=None)) + @patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) + def test_wait_for_leader(self): + dcs = AbstractDCS('dummy',{'namespace':'dummy', 'scope':'dummy'}) + self.assertRaises(patroni.exceptions.PatroniCtlException, wait_for_leader, dcs, 0) + + cluster = wait_for_leader(dcs=dcs, timeout=2) + assert cluster.leader.member.name == 'leader' + + + def test_post_patroni(self): + member = get_cluster_initialized_with_leader().leader.member + self.assertRaises(requests.exceptions.ConnectionError, post_patroni, member, 'dummy', {}) + + + def test_ctl(self): + runner = CliRunner() + + runner.invoke(ctl, ['list']) + + result = runner.invoke(ctl, ['--help']) + assert 'Usage:' in result.output -def test_members(): - runner = CliRunner() + def test_members(self): + runner = CliRunner() - runner.invoke(members) + runner.invoke(members) From 2d9f5d9e4b48cd03fc7f4353ed7b6abc6cc66c8c Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Tue, 17 Nov 2015 13:55:09 +0100 Subject: [PATCH 09/14] Refactoring and adding a dsn option to patronictl. Some refactoring to reuse some codepaths. A dsn option is now added, it is useful in scripts like so: psql -d "$(patronictl dsn alpha)" Restarting has been extended to allow restarting based on role. --- patroni/ctl.py | 298 ++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 258 insertions(+), 40 deletions(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index 490ee607..cf57cec6 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -1,3 +1,6 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + ''' Patroni Control ''' @@ -7,6 +10,8 @@ import os import yaml import json import time +import psycopg2 +import random import requests import datetime from prettytable import PrettyTable @@ -15,10 +20,11 @@ import logging from .etcd import Etcd from .exceptions import PatroniCtlException +from .postgresql import parseurl CONFIG_DIR_PATH = click.get_app_dir('patroni') CONFIG_FILE_PATH = os.path.join(CONFIG_DIR_PATH, 'patronictl.yaml') -LOGLEVEL = 'INFO' +LOGLEVEL = 'WARNING' def parse_dcs(dcs): @@ -38,7 +44,7 @@ def parse_dcs(dcs): parsed = urlparse(dcs) scheme = parsed.scheme if scheme == '' and parsed.netloc == '': - parsed = urlparse('//'+dcs) + parsed = urlparse('//' + dcs) if scheme == '': default_schemes = {'2181': 'zookeeper', '8500': 'consul'} @@ -77,6 +83,7 @@ def store_config(config, path): with open(path, 'w') as fd: yaml.dump(config, fd) + option_config_file = click.option('--config-file', '-c', help='Configuration file', default=CONFIG_FILE_PATH) option_format = click.option('--format', '-f', help='Output format (pretty, json)', default='pretty') option_dcs = click.option('--dcs', '-d', help='Use this DCS', envvar='DCS') @@ -88,16 +95,15 @@ option_force = click.option('--force', is_flag=True, help='Do not ask for confir @click.group() @click.pass_context def ctl(ctx): + global LOGLEVEL + if 'DEBUG' in os.environ: + LOGLEVEL = os.environ.get('DEBUG') + if LOGLEVEL == '': + LOGLEVEL = 'DEBUG' logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=LOGLEVEL) def get_dcs(config, scope): - """ - >>> get_dcs({'scheme':'redis', 'hostname':'a', 'port':'1'}, 'testing') - Traceback (most recent call last): - ... - patroni.exceptions.PatroniCtlException: Can not find suitable configuration of distributed configuration store - """ scheme, hostname, port = map(config.get('dcs', {}).get, ('scheme', 'hostname', 'port')) if scheme == 'etcd': @@ -109,18 +115,18 @@ def get_dcs(config, scope): def post_patroni(member, endpoint, content, headers={'Content-Type': 'application/json'}): url = urlparse(member.api_url) logging.debug(url) - return requests.post('{}://{}/{}'.format( - url.scheme, url.netloc, endpoint), headers=headers, data=json.dumps(content)) + return requests.post('{}://{}/{}'.format(url.scheme, url.netloc, endpoint), headers=headers, + data=json.dumps(content)) -def print_output(columns, rows=[], alignment=None, format='pretty'): +def print_output(columns, rows=[], alignment=None, format='pretty', header=True, delimiter='\t'): if format == 'pretty': t = PrettyTable(columns) for k, v in (alignment or {}).items(): t.align[k] = v for r in rows: t.add_row(r) - print(t) + click.echo(t) return if format == 'json': @@ -128,19 +134,28 @@ def print_output(columns, rows=[], alignment=None, format='pretty'): for r in rows: elements.append(dict(zip(columns, r))) - print(json.dumps(elements)) + click.echo(json.dumps(elements)) + + if format == 'tsv': + if columns is not None and header: + click.echo(delimiter.join(columns) + '\n') + + for r in rows or []: + c = [str(c) for c in r] + click.echo(delimiter.join(c)) -def watching(w, watch, max_count=None): +def watching(w, watch, max_count=None, clear=True): """ >>> len(list(watching(True, 1, 0))) 1 >>> len(list(watching(True, 1, 1))) 2 """ + if w and not watch: watch = 2 - if watch: + if watch and clear: click.clear() yield 0 @@ -151,10 +166,171 @@ def watching(w, watch, max_count=None): while watch and counter <= (max_count or counter): time.sleep(watch) counter += 1 - click.clear() + if clear: + click.clear() yield 0 +def build_connect_parameters(conn_url, connect_parameters={}): + params = connect_parameters.copy() + parsed = parseurl(conn_url) + params['host'] = parsed['host'] + params['port'] = parsed['port'] + params['fallback_application_name'] = 'Patroni ctl' + params['connect_timeout'] = '5' + + return params + + +def get_all_members(cluster, role='master'): + if role == 'master': + yield (None if cluster.leader is None else cluster.leader.member) + return + + leader_name = (cluster.leader.member.name if cluster.leader else None) + for m in cluster.members: + if role == 'any' or role == 'replica' and m.name != leader_name: + yield m + + +def get_any_member(cluster, role='master', member=None): + members = get_all_members(cluster=cluster, role=role) + for m in members: + if member is None or m.name == member: + return m + + return None + + +def get_cursor(cluster, role='master', member=None, connect_parameters={}): + member = get_any_member(cluster=cluster, role=role, member=member) + if member is None: + return None + + params = build_connect_parameters(member.conn_url, connect_parameters=connect_parameters) + + conn = psycopg2.connect(**params) + conn.autocommit = True + cursor = conn.cursor() + if role == 'any': + return cursor + + cursor.execute('SELECT pg_is_in_recovery()') + in_recovery = cursor.fetchone()[0] + + if in_recovery and role == 'replica' or not in_recovery and role == 'master': + return cursor + + conn.close() + + return None + + +@ctl.command('dsn', help='Generate a dsn for the provided member, defaults to a dsn of the master') +@click.option('--role', '-r', help='Give a dsn of any member with this role', type=click.Choice(['master', 'replica', + 'any']), default=None) +@click.option('--member', '-m', help='Generate a dsn for this member', type=str) +@option_dcs +@option_config_file +@click.argument('cluster_name') +def dsn(cluster_name, config_file, dcs, role, member): + if role is not None and member is not None: + raise PatroniCtlException('--role and --member are mutually exclusive options') + if member is None and role is None: + role = 'master' + + config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) + m = get_any_member(cluster=cluster, role=role, member=member) + if m is None: + raise PatroniCtlException('Can not find a suitable member') + + params = build_connect_parameters(m.conn_url) + click.echo('host={} port={}'.format(params['host'], params['port'])) + + +@ctl.command('query', help='Query a Patroni PostgreSQL member') +@click.argument('cluster_name') +@option_config_file +@option_format +@click.option('--format', help='Output format (pretty, json)', default='tsv') +@click.option('--file', '-f', help='Execute the SQL commands from this file', type=click.File('rb')) +@option_dcs +@option_watch +@option_watchrefresh +@click.option('--role', '-r', help='The role of the query', type=click.Choice(['master', 'replica', 'any']), + default=None) +@click.option('--member', '-m', help='Query a specific member', type=str) +@click.option('--delimiter', help='The column delimiter', default='\t') +@click.option('--command', '-c', help='The SQL commands to execute') +def query( + cluster_name, + config_file, + dcs, + role, + member, + w, + watch, + delimiter, + command, + file, + format='tsv', +): + if role is not None and member is not None: + raise PatroniCtlException('--role and --member are mutually exclusive options') + if member is None and role is None: + role = 'master' + + if file is not None and command is not None: + raise PatroniCtlException('--file and --command are mutually exclusive options') + + if file is not None: + command = file.read() + + config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) + + cursor = None + for _ in watching(w, watch, clear=False): + + output, cursor = query_member(cluster=cluster, cursor=cursor, member=member, role=role, command=command) + print_output(None, output, format=format, delimiter=delimiter) + + if cursor is None: + cluster = dcs.get_cluster() + + +def query_member(cluster, cursor, member, role, command): + try: + if cursor is None: + cursor = get_cursor(cluster, role=role, member=member) + + if cursor is None: + if role is None: + message = 'No connection to member {} is available'.format(member) + else: + message = 'No connection to role={} is available'.format(role) + logging.debug(message) + return [[timestamp(0), message]], None + + cursor.execute('SELECT pg_is_in_recovery()') + in_recovery = cursor.fetchone()[0] + + if in_recovery and role == 'master' or not in_recovery and role == 'replica': + cursor.connection.close() + return None, None + + cursor.execute(command) + return cursor.fetchall(), cursor + except (psycopg2.OperationalError, psycopg2.DatabaseError) as oe: + logging.debug(oe) + if cursor is not None and not cursor.connection.closed: + cursor.connection.close() + message = oe.pgcode or oe.pgerror or str(oe) + message = message.replace('\n', ' ') + return [[timestamp(0), 'ERROR, SQLSTATE: {}'.format(message)]], None + + return None, None + + @ctl.command('remove', help='Remove cluster from DCS') @click.argument('cluster_name') @option_config_file @@ -167,23 +343,24 @@ def remove(config_file, cluster_name, format, dcs): confirm = click.prompt('Please confirm the cluster name to remove', type=str) if confirm != cluster_name: - raise PatroniCtlException("Cluster names specified do not match") + raise PatroniCtlException('Cluster names specified do not match') message = 'Yes I am aware' - confirm = click.prompt('You are about to remove all information in DCS for {}, please type: "{}"'.format( - cluster_name, message), type=str) + confirm = \ + click.prompt('You are about to remove all information in DCS for {}, please type: "{}"'.format(cluster_name, + message), type=str) if message != confirm: raise PatroniCtlException('You did not exactly type "{}"'.format(message)) if cluster.leader: confirm = click.prompt('This cluster currently is healthy. Please specify the master name to continue') if confirm != cluster.leader.name: - raise PatroniCtlException("You did not specify the current master of the cluster") + raise PatroniCtlException('You did not specify the current master of the cluster') if isinstance(dcs, Etcd): dcs.client.delete(dcs._base_path, recursive=True) else: - raise PatroniCtlException("We have not implemented this for DCS of type {}", type(dcs)) + raise PatroniCtlException('We have not implemented this for DCS of type {}', type(dcs)) def wait_for_leader(dcs, timeout=30): @@ -206,9 +383,8 @@ def empty_post_to_members(cluster, member_names, force, endpoint): candidates[m.name] = m if len(member_names) == 0: - member_names = [ - click.prompt('Which member do you want to {} [{}]?'.format( - endpoint, ', '.join(candidates.keys())), type=str, default='')] + member_names = [click.prompt('Which member do you want to {} [{}]?'.format(endpoint, + ', '.join(candidates.keys())), type=str, default='')] for mn in member_names: if mn not in candidates.keys(): @@ -232,17 +408,33 @@ def ctl_load_config(cluster_name, config_file, dcs): dcs = get_dcs(config, cluster_name) cluster = dcs.get_cluster() - return (config, dcs, cluster) + return config, dcs, cluster @ctl.command('restart', help='Restart cluster member') @click.argument('cluster_name') @click.argument('member_names', nargs=-1) +@click.option('--role', '-r', help='Restart only members with this role', default='any', + type=click.Choice(['master', 'replica', 'any'])) +@click.option('--any', help='Restart a single member only', is_flag=True) @option_config_file @option_force @option_dcs -def restart(cluster_name, member_names, config_file, dcs, force): +def restart(cluster_name, member_names, config_file, dcs, force, role, any): config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) + + role_names = [m.name for m in get_all_members(cluster=cluster, role=role)] + + if len(member_names) > 0: + member_names = list(set(member_names) & set(role_names)) + else: + member_names = role_names + + if any: + random.shuffle(member_names) + member_names = member_names[:1] + + output_members(cluster) empty_post_to_members(cluster, member_names, force, 'restart') @@ -271,6 +463,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): We verify that the cluster name, master name and candidate name are correct. If so, we trigger a failover and keep the client up to date. """ + config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) if cluster.leader is None: @@ -293,7 +486,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): raise PatroniCtlException('No candidates found to failover to') if candidate is None and not force: - candidate = click.prompt('Candidate '+str(candidate_names), type=str, default='') + candidate = click.prompt('Candidate ' + str(candidate_names), type=str, default='') if candidate and candidate not in candidate_names: raise PatroniCtlException('Member {} does not exist in cluster {}'.format(candidate, cluster_name)) @@ -303,31 +496,32 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): output_members(dcs.get_cluster(), name=cluster_name) if not force: - a = click.confirm('Are you sure you want to failover cluster {}, demoting current master {}?'.format( - cluster_name, master)) + a = \ + click.confirm('Are you sure you want to failover cluster {}, demoting current master {}?'.format( + cluster_name, master)) if not a: raise PatroniCtlException('Aborting failover') - failover_value = '{}:{}'.format(master, (candidate or '')) + failover_value = '{}:{}'.format(master, candidate or '') t_started = time.time() r = None try: - r = post_patroni(cluster.leader.member, 'failover', {'leader': master, 'candidate': (candidate or '')}) + r = post_patroni(cluster.leader.member, 'failover', {'leader': master, 'candidate': candidate or ''}) if r.status_code == 200: logging.debug(r) logging.debug(r.text) cluster = dcs.get_cluster() - click.echo(timestamp()+' Failing over to new leader: {}'.format(cluster.leader.member.name)) + click.echo(timestamp() + ' Failing over to new leader: {}'.format(cluster.leader.member.name)) else: click.echo('Failover failed, details: {}, {}'.format(r.status_code, r.text)) return except: logging.exception(r) logging.warning('Failing over to DCS') - click.echo(timestamp()+' Could not failover using Patroni api, falling back to DCS') + click.echo(timestamp() + ' Could not failover using Patroni api, falling back to DCS') dcs.set_failover_value(failover_value) - click.echo(timestamp()+' Initialized failover from master {}'.format(master)) + click.echo(timestamp() + ' Initialized failover from master {}'.format(master)) # The failover process should within a minute update the failover key, we will keep watching it until it changes # or we timeout cluster = wait_for_leader(dcs, timeout=60) @@ -335,8 +529,8 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): click.echo('Failover failed, master did not change after {:0.1f} seconds'.format(time.time() - t_started)) return - click.echo(timestamp()+' Failover completed in {:0.1f} seconds, new leader is {}'.format( - time.time() - t_started, str(cluster.leader.member.name))) + click.echo(timestamp() + ' Failover completed in {:0.1f} seconds, new leader is {}'.format(time.time() - t_started, + str(cluster.leader.member.name))) output_members(cluster, name=cluster_name) @@ -356,9 +550,33 @@ def output_members(cluster, name=None, format='pretty'): if m.name == leader_name: leader = '*' - rows.append([name, m.name, leader]) + host = build_connect_parameters(m.conn_url)['host'] - print_output(['Cluster', 'Member', 'Leader'], rows, {'Cluster': 'l', 'Member': 'l'}, format) + xlog_location = m.data.get('xlog_location') + lag = '' + if xlog_location is not None: + lag = round((cluster.last_leader_operation - m.data.get('xlog_location', 0)) / 1024 / 1024) + + rows.append([ + name, + m.name, + host, + leader, + m.data.get('state', ''), + lag, + ]) + + columns = [ + 'Cluster', + 'Member', + 'Host', + 'Leader', + 'State', + 'Lag in MB', + ] + alignment = {'Cluster': 'l', 'Member': 'l'} + + print_output(columns, rows, alignment, format) @ctl.command('list', help='List the Patroni members for a given Patroni') @@ -381,8 +599,8 @@ def members(config_file, cluster_names, format, watch, w, dcs): output_members(dcs.get_cluster(), name=cn, format=format) -def timestamp(): - return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f") +def timestamp(precision=6): + return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:precision - 7] @ctl.command('configure', help='Create configuration file') From ca4d9eaaf99b65df4ed98cac338e3afa9db94873 Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Wed, 18 Nov 2015 07:54:08 +0100 Subject: [PATCH 10/14] Patronictl: Expand tests to increase coverage --- patroni/ctl.py | 29 ++-- tests/test_ctl.py | 336 +++++++++++++++++++++++++++++++++------ tests/test_etcd.py | 1 + tests/test_ha.py | 6 +- tests/test_postgresql.py | 3 + 5 files changed, 311 insertions(+), 64 deletions(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index cf57cec6..5192ab38 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -1,6 +1,3 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - ''' Patroni Control ''' @@ -96,10 +93,8 @@ option_force = click.option('--force', is_flag=True, help='Do not ask for confir @click.pass_context def ctl(ctx): global LOGLEVEL - if 'DEBUG' in os.environ: - LOGLEVEL = os.environ.get('DEBUG') - if LOGLEVEL == '': - LOGLEVEL = 'DEBUG' + LOGLEVEL = os.environ.get('LOGLEVEL', LOGLEVEL) + logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=LOGLEVEL) @@ -151,6 +146,8 @@ def watching(w, watch, max_count=None, clear=True): 1 >>> len(list(watching(True, 1, 1))) 2 + >>> len(list(watching(True, None, 0))) + 1 """ if w and not watch: @@ -184,7 +181,8 @@ def build_connect_parameters(conn_url, connect_parameters={}): def get_all_members(cluster, role='master'): if role == 'master': - yield (None if cluster.leader is None else cluster.leader.member) + if cluster.leader is not None: + yield cluster.leader return leader_name = (cluster.leader.member.name if cluster.leader else None) @@ -328,8 +326,6 @@ def query_member(cluster, cursor, member, role, command): message = message.replace('\n', ' ') return [[timestamp(0), 'ERROR, SQLSTATE: {}'.format(message)]], None - return None, None - @ctl.command('remove', help='Remove cluster from DCS') @click.argument('cluster_name') @@ -339,6 +335,9 @@ def query_member(cluster, cursor, member, role, command): def remove(config_file, cluster_name, format, dcs): config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) + if not isinstance(dcs, Etcd): + raise PatroniCtlException('We have not implemented this for DCS of type {}'.format(type(dcs))) + output_members(cluster, format=format) confirm = click.prompt('Please confirm the cluster name to remove', type=str) @@ -357,10 +356,7 @@ def remove(config_file, cluster_name, format, dcs): if confirm != cluster.leader.name: raise PatroniCtlException('You did not specify the current master of the cluster') - if isinstance(dcs, Etcd): - dcs.client.delete(dcs._base_path, recursive=True) - else: - raise PatroniCtlException('We have not implemented this for DCS of type {}', type(dcs)) + dcs.client.delete(dcs._base_path, recursive=True) def wait_for_leader(dcs, timeout=30): @@ -488,6 +484,9 @@ def failover(config_file, cluster_name, master, candidate, force, dcs): if candidate is None and not force: candidate = click.prompt('Candidate ' + str(candidate_names), type=str, default='') + if candidate == master: + raise PatroniCtlException('Failover target and source are the same.') + if candidate and candidate not in candidate_names: raise PatroniCtlException('Member {} does not exist in cluster {}'.format(candidate, cluster_name)) @@ -555,7 +554,7 @@ def output_members(cluster, name=None, format='pretty'): xlog_location = m.data.get('xlog_location') lag = '' if xlog_location is not None: - lag = round((cluster.last_leader_operation - m.data.get('xlog_location', 0)) / 1024 / 1024) + lag = round(((cluster.last_leader_operation or 0) - m.data.get('xlog_location', 0)) / 1024 / 1024) rows.append([ name, diff --git a/tests/test_ctl.py b/tests/test_ctl.py index b0715118..2263ec18 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -1,34 +1,72 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + import os -import mock import pytest import unittest +import psycopg2 import requests import patroni.exceptions +import etcd from mock import patch, Mock from click.testing import CliRunner -from patroni.ctl import ctl, members, store_config, load_config, output_members, post_patroni, get_dcs, wait_for_leader -from patroni.dcs import AbstractDCS, Member -from patroni.etcd import Etcd -from test_ha import get_cluster_initialized_without_leader, get_cluster_initialized_with_leader, get_cluster +from patroni.ctl import ctl, members, store_config, load_config, output_members, post_patroni, get_dcs, \ + wait_for_leader, get_all_members, get_any_member, get_cursor, query_member, configure +from patroni.ha import Ha +from patroni.etcd import Etcd, Client +from test_ha import get_cluster_initialized_without_leader, get_cluster_initialized_with_leader, \ + get_cluster_initialized_with_only_leader, MockPostgresql, MockPatroni, run_async, \ + get_cluster_not_initialized_without_leader +from test_etcd import etcd_read, etcd_write, requests_get, MockResponse +from test_postgresql import MockConnect, psycopg2_connect CONFIG_FILE_PATH = './test-ctl.yaml' + class TestCtl(unittest.TestCase): + @patch.object(Client, 'machines') + def setUp(self, mock_machines): + mock_machines.__get__ = Mock(return_value=['http://remotehost:2379']) + self.p = MockPostgresql() + self.e = Etcd('foo', {'ttl': 30, 'host': 'ok:2379', 'scope': 'test'}) + self.e.client.read = etcd_read + self.e.client.write = etcd_write + self.e.client.delete = Mock(side_effect=etcd.EtcdException()) + self.ha = Ha(MockPatroni(self.p, self.e)) + self.ha._async_executor.run_async = run_async + self.ha.old_cluster = self.e.get_cluster() + self.ha.cluster = get_cluster_not_initialized_without_leader() + self.ha.load_cluster_from_dcs = Mock() + + @patch('psycopg2.connect', psycopg2_connect) + def test_get_cursor(self): + c = get_cursor(get_cluster_initialized_without_leader(), role='master') + assert c is None + + c = get_cursor(get_cluster_initialized_with_leader(), role='master') + assert c is not None + + c = get_cursor(get_cluster_initialized_with_leader(), role='replica') + # # MockCursor returns pg_is_in_recovery as false + assert c is None + + c = get_cursor(get_cluster_initialized_with_leader(), role='any') + assert c is not None + def test_output_members(self): cluster = get_cluster_initialized_with_leader() output_members(cluster, name='abc', format='pretty') output_members(cluster, name='abc', format='json') - + output_members(cluster, name='abc', format='tsv') def test_rw_config(self): runner = CliRunner() config = 'a:b' with runner.isolated_filesystem(): - store_config(config, CONFIG_FILE_PATH) - os.remove(CONFIG_FILE_PATH) - os.mkdir(CONFIG_FILE_PATH) + store_config(config, CONFIG_FILE_PATH + '/dummy') + os.remove(CONFIG_FILE_PATH + '/dummy') with pytest.raises(Exception): result = load_config(CONFIG_FILE_PATH, None) assert 'Could not load configuration file' in result.output @@ -41,38 +79,198 @@ class TestCtl(unittest.TestCase): store_config(config, 'abc/CONFIG_FILE_PATH') load_config(CONFIG_FILE_PATH, None) - @patch('patroni.ctl.get_dcs', Mock(return_value=AbstractDCS('dummy',{'namespace':'dummy', 'scope':'dummy'}))) - @patch('patroni.ctl.post_patroni', Mock(return_value=None)) + @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) + @patch('patroni.etcd.Etcd.get_etcd_client', Mock(return_value=None)) + @patch('patroni.etcd.Etcd.set_failover_value', Mock(return_value=None)) + @patch('patroni.ctl.wait_for_leader', Mock(return_value=get_cluster_initialized_with_leader())) + @patch('requests.get', requests_get) + @patch('requests.post', requests_get) + @patch('patroni.ctl.post_patroni', Mock(return_value=MockResponse())) def test_failover(self): runner = CliRunner() - - with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_without_leader())): - result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8']) + + with patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())): + result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader +other +y''') + assert 'Failing over to new leader' in result.output + + result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader +other +N''') + assert 'Aborting failover' in str(result.exception) + + result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader +leader +y''') + assert 'target and source are the same' in str(result.exception) + + result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader +Reality +y''') + assert 'Reality does not exist' in str(result.exception) + + result = runner.invoke(ctl, ['failover', 'dummy', '--force']) + assert 'Failing over to new leader' in result.output + + result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='dummy') + assert 'is not the leader of cluster' in str(result.exception) + + with patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_only_leader())): + result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader +other +y''') + assert 'No candidates found to failover to' in str(result.exception) + + with patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_without_leader())): + result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader +other +y''') assert 'This cluster has no master' in str(result.exception) - with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())): - result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='nonsense') - assert 'is not the leader of cluster' in str(result.exception) - - result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8', '--master', 'nonsense']) - assert 'is not the leader of cluster' in str(result.exception) - - result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nn') - assert 'Aborting failover' in str(result.exception) - - with patch('patroni.ctl.wait_for_leader', Mock(return_value = get_cluster_initialized_with_leader())): - result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nY') - assert 'master did not change after' in result.output - - result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nY') - assert 'Failover failed' in result.output - - def test_get_dcs(self): - self.assertRaises(patroni.exceptions.PatroniCtlException, get_dcs, {'scheme':'dummy'}, 'dummy') + with patch('patroni.ctl.post_patroni', Mock(side_effect=Exception())): + result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader +other +y''') + assert 'falling back to DCS' in result.output + assert 'Failover failed' in result.output + mocked = Mock() + mocked.return_value.status_code = 500 + with patch('patroni.ctl.post_patroni', Mock(return_value=mocked)): + result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader +other +y''') + assert 'Failover failed, details' in result.output + +# with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())): +# result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='nonsense') +# assert 'is not the leader of cluster' in str(result.exception) + + # result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8', '--master', 'nonsense']) + # assert 'is not the leader of cluster' in str(result.exception) + + # result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nn') + # assert 'Aborting failover' in str(result.exception) + + # with patch('patroni.ctl.wait_for_leader', Mock(return_value = get_cluster_initialized_with_leader())): + # result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nY') + # assert 'master did not change after' in result.output + + # result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nY') + # assert 'Failover failed' in result.output + + def test_(self): + self.assertRaises(patroni.exceptions.PatroniCtlException, get_dcs, {'scheme': 'dummy'}, 'dummy') + + @patch('psycopg2.connect', psycopg2_connect) + @patch('patroni.ctl.query_member', Mock(return_value=([['mock column']], None))) + def test_query(self): + runner = CliRunner() + + with patch('patroni.ctl.get_dcs', Mock(return_value=self.e)): + result = runner.invoke(ctl, [ + 'query', + 'alpha', + '--member', + 'abc', + '--role', + 'master', + ]) + assert 'mutually exclusive' in str(result.exception) + + with runner.isolated_filesystem(): + dummy_file = open('dummy', 'w') + dummy_file.write('SELECT 1') + dummy_file.close() + + result = runner.invoke(ctl, [ + 'query', + 'alpha', + '--file', + 'dummy', + '--command', + 'dummy', + ]) + assert 'mutually exclusive' in str(result.exception) + + result = runner.invoke(ctl, ['query', 'alpha', '--file', 'dummy']) + + os.remove('dummy') + + result = runner.invoke(ctl, ['query', 'alpha', '--command', 'SELECT 1']) + assert 'mock column' in result.output + + @patch('patroni.ctl.get_cursor', Mock(return_value=MockConnect().cursor())) + def test_query_member(self): + rows = query_member(None, None, None, 'master', 'SELECT pg_is_in_recovery()') + assert 'False' in str(rows) + + rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()') + assert rows == (None, None) + + with patch('patroni.ctl.get_cursor', Mock(return_value=None)): + rows = query_member(None, None, None, None, 'SELECT pg_is_in_recovery()') + assert 'No connection to' in str(rows) + + rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()') + assert 'No connection to' in str(rows) + + with patch('patroni.ctl.get_cursor', Mock(side_effect=psycopg2.OperationalError('bla'))): + rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()') + + with patch('test_postgresql.MockCursor.execute', Mock(side_effect=psycopg2.OperationalError('bla'))): + rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()') - @patch('patroni.ctl.get_dcs', Mock(return_value=AbstractDCS('dummy',{'namespace':'dummy', 'scope':'dummy'}))) @patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) + def test_dsn(self): + runner = CliRunner() + + with patch('patroni.ctl.get_dcs', Mock(return_value=self.e)): + result = runner.invoke(ctl, ['dsn', 'alpha', '--dcs', '8.8.8.8']) + assert 'host=127.0.0.1 port=5435' in result.output + + result = runner.invoke(ctl, [ + 'dsn', + 'alpha', + '--role', + 'master', + '--member', + 'dummy', + ]) + assert 'mutually exclusive' in str(result.exception) + + result = runner.invoke(ctl, ['dsn', 'alpha', '--member', 'dummy']) + assert 'Can not find' in str(result.exception) + + # result = runner.invoke(ctl, ['dsn', 'alpha', '--dcs', '8.8.8.8', '--role', 'replica']) + # assert 'host=127.0.0.1 port=5436' in result.output + + @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) + @patch('patroni.etcd.Etcd.get_etcd_client', Mock(return_value=None)) + @patch('requests.get', requests_get) + @patch('requests.post', requests_get) + def test_restart_reinit(self): + runner = CliRunner() + + result = runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='y') + result = runner.invoke(ctl, ['reinit', 'alpha', '--dcs', '8.8.8.8'], input='y') + + result = runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='N') + result = runner.invoke(ctl, [ + 'restart', + 'alpha', + '--dcs', + '8.8.8.8', + 'dummy', + '--any', + ], input='y') + assert 'not a member' in str(result.exception) + + with patch('requests.post', Mock(return_value=MockResponse())): + result = runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='y') + + @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) @patch('patroni.etcd.Etcd.get_etcd_client', Mock(return_value=None)) def test_remove(self): runner = CliRunner() @@ -82,33 +280,40 @@ class TestCtl(unittest.TestCase): assert 'You are about to remove all' in result.output assert 'You did not exactly type' in str(result.exception) - result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='alpha\nYes I am aware\nleader') - assert 'We have not implemented this for DCS of type' in str(result.exception) - - result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='alpha\nYes I am aware\nslave') + result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='''alpha +Yes I am aware +slave''') assert 'You did not specify the current master of the cluster' in str(result.exception) result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='beta\nleader') assert 'Cluster names specified do not match' in str(result.exception) - - with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=Etcd('dummy', {'namespace':'dummy', 'scope':'dummy'}))): - result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='alpha\nYes I am aware\nleader') + + with patch('patroni.etcd.Etcd.get_cluster', get_cluster_initialized_with_leader): + result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], + input='''alpha +Yes I am aware +leader''') assert 'object has no attribute' in str(result.exception) - @patch('patroni.dcs.AbstractDCS.watch', Mock(return_value=None)) - @patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) + with patch('patroni.ctl.get_dcs', Mock(return_value=Mock())): + result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], + input='''alpha +Yes I am aware +leader''') + assert 'We have not implemented this for DCS of type' in str(result.exception) + + @patch('patroni.etcd.Etcd.watch', Mock(return_value=None)) + @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) def test_wait_for_leader(self): - dcs = AbstractDCS('dummy',{'namespace':'dummy', 'scope':'dummy'}) + dcs = self.e self.assertRaises(patroni.exceptions.PatroniCtlException, wait_for_leader, dcs, 0) - + cluster = wait_for_leader(dcs=dcs, timeout=2) assert cluster.leader.member.name == 'leader' - def test_post_patroni(self): member = get_cluster_initialized_with_leader().leader.member self.assertRaises(requests.exceptions.ConnectionError, post_patroni, member, 'dummy', {}) - def test_ctl(self): runner = CliRunner() @@ -118,10 +323,45 @@ class TestCtl(unittest.TestCase): result = runner.invoke(ctl, ['--help']) assert 'Usage:' in result.output + def test_get_any_member(self): + m = get_any_member(get_cluster_initialized_without_leader(), role='master') + assert m is None + + m = get_any_member(get_cluster_initialized_with_leader(), role='master') + assert m.name == 'leader' + + def test_get_all_members(self): + r = list(get_all_members(get_cluster_initialized_without_leader(), role='master')) + assert len(r) == 0 + + r = list(get_all_members(get_cluster_initialized_with_leader(), role='master')) + assert len(r) == 1 + assert r[0].name == 'leader' + + r = list(get_all_members(get_cluster_initialized_with_leader(), role='replica')) + assert len(r) == 1 + assert r[0].name == 'other' + + r = list(get_all_members(get_cluster_initialized_without_leader(), role='replica')) + assert len(r) == 2 def test_members(self): runner = CliRunner() - runner.invoke(members) + runner.invoke(members, ['alpha']) + + def test_configure(self): + runner = CliRunner() + + result = runner.invoke(configure, [ + '--dcs', + 'abc', + '-c', + 'dummy', + '-n', + 'bla', + ]) + + assert result.exit_code == 0 diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 6ffa59af..10b12bd9 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -17,6 +17,7 @@ class MockResponse: self.status_code = 200 self.content = '{}' self.ok = True + self.text = '' def json(self): return json.loads(self.content) diff --git a/tests/test_ha.py b/tests/test_ha.py index e34f9b8e..3671fea7 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -27,7 +27,7 @@ def get_cluster_not_initialized_without_leader(): def get_cluster_initialized_without_leader(leader=False, failover=None): m = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', - 'api_url': 'http://127.0.0.1:8008/patroni'}) + 'api_url': 'http://127.0.0.1:8008/patroni', 'xlog_location':4}) l = Leader(0, 0, m) if leader else None o = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres', 'api_url': 'http://127.0.0.1:8011/patroni'}) @@ -37,6 +37,10 @@ def get_cluster_initialized_without_leader(leader=False, failover=None): def get_cluster_initialized_with_leader(failover=None): return get_cluster_initialized_without_leader(leader=True, failover=failover) +def get_cluster_initialized_with_only_leader(failover=None): + l = get_cluster_initialized_without_leader(leader=True, failover=failover).leader + return get_cluster(True, l, [l], failover) + class MockPostgresql(Mock): diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 0ed04a15..09d6fb91 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -56,6 +56,9 @@ class MockCursor: def fetchone(self): return self.results[0] + def fetchall(self): + return self.results + def close(self): pass From 4bb1e060c0f09438724826a44ae3e8541bdf7a78 Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Wed, 18 Nov 2015 12:00:51 +0100 Subject: [PATCH 11/14] Bugfix for patronictl tests --- tests/test_ctl.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 2263ec18..093882a7 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -23,7 +23,7 @@ from test_postgresql import MockConnect, psycopg2_connect CONFIG_FILE_PATH = './test-ctl.yaml' - +@patch('patroni.ctl.load_config', Mock(return_value={'dcs': {'scheme': 'etcd', 'hostname': 'localhost', 'port': 4001}})) class TestCtl(unittest.TestCase): @patch.object(Client, 'machines') From cae025c3fb645c4a4405ff8887b1f03ca7621db3 Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Wed, 18 Nov 2015 12:50:36 +0100 Subject: [PATCH 12/14] Testing patronictl: Mock configuration and bugfix --- tests/test_ctl.py | 43 ++++++++++++++++++++++++------------------- 1 file changed, 24 insertions(+), 19 deletions(-) diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 093882a7..128f4a16 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -23,6 +23,28 @@ from test_postgresql import MockConnect, psycopg2_connect CONFIG_FILE_PATH = './test-ctl.yaml' +def test_rw_config(): + runner = CliRunner() + config = {'a':'b'} + with runner.isolated_filesystem(): + store_config(config, CONFIG_FILE_PATH + '/dummy') + os.remove(CONFIG_FILE_PATH + '/dummy') + os.rmdir(CONFIG_FILE_PATH) + + with pytest.raises(Exception): + result = load_config(CONFIG_FILE_PATH, None) + assert 'Could not load configuration file' in result.output + + os.mkdir(CONFIG_FILE_PATH) + with pytest.raises(Exception): + store_config(config, CONFIG_FILE_PATH) + + os.rmdir(CONFIG_FILE_PATH) + + store_config(config, CONFIG_FILE_PATH) + load_config(CONFIG_FILE_PATH, None) + load_config(CONFIG_FILE_PATH, '0.0.0.0') + @patch('patroni.ctl.load_config', Mock(return_value={'dcs': {'scheme': 'etcd', 'hostname': 'localhost', 'port': 4001}})) class TestCtl(unittest.TestCase): @@ -61,24 +83,6 @@ class TestCtl(unittest.TestCase): output_members(cluster, name='abc', format='json') output_members(cluster, name='abc', format='tsv') - def test_rw_config(self): - runner = CliRunner() - config = 'a:b' - with runner.isolated_filesystem(): - store_config(config, CONFIG_FILE_PATH + '/dummy') - os.remove(CONFIG_FILE_PATH + '/dummy') - with pytest.raises(Exception): - result = load_config(CONFIG_FILE_PATH, None) - assert 'Could not load configuration file' in result.output - - with pytest.raises(Exception): - store_config(config, CONFIG_FILE_PATH) - - os.rmdir(CONFIG_FILE_PATH) - - store_config(config, 'abc/CONFIG_FILE_PATH') - load_config(CONFIG_FILE_PATH, None) - @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) @patch('patroni.etcd.Etcd.get_etcd_client', Mock(return_value=None)) @patch('patroni.etcd.Etcd.set_failover_value', Mock(return_value=None)) @@ -348,7 +352,8 @@ leader''') def test_members(self): runner = CliRunner() - runner.invoke(members, ['alpha']) + result = runner.invoke(members, ['alpha']) + assert result.exit_code == 0 def test_configure(self): runner = CliRunner() From f081f9d67eaad5a782e719ad29e7aff2ec49be6a Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Wed, 18 Nov 2015 12:55:11 +0100 Subject: [PATCH 13/14] Bugfix for Patronictl tests --- tests/test_ctl.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 128f4a16..0d65a05c 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -349,6 +349,10 @@ leader''') r = list(get_all_members(get_cluster_initialized_without_leader(), role='replica')) assert len(r) == 2 + @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())) + @patch('patroni.etcd.Etcd.get_etcd_client', Mock(return_value=None)) + @patch('requests.get', requests_get) + @patch('requests.post', requests_get) def test_members(self): runner = CliRunner() From 897024a297221e7b6413ef2e72167f3f8cdd51ed Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Wed, 18 Nov 2015 15:01:35 +0100 Subject: [PATCH 14/14] PatroniCtl: Bugfixes for formatting and timeouts --- patroni/ctl.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/patroni/ctl.py b/patroni/ctl.py index 5192ab38..a3c8aa73 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -111,7 +111,7 @@ def post_patroni(member, endpoint, content, headers={'Content-Type': 'applicatio url = urlparse(member.api_url) logging.debug(url) return requests.post('{}://{}/{}'.format(url.scheme, url.netloc, endpoint), headers=headers, - data=json.dumps(content)) + data=json.dumps(content), timeout=5) def print_output(columns, rows=[], alignment=None, format='pretty', header=True, delimiter='\t'): @@ -573,7 +573,7 @@ def output_members(cluster, name=None, format='pretty'): 'State', 'Lag in MB', ] - alignment = {'Cluster': 'l', 'Member': 'l'} + alignment = {'Cluster': 'l', 'Member': 'l', 'Host': 'l'} print_output(columns, rows, alignment, format)