From e2aff13d3e846d36f4608346cbc04cdf0d295545 Mon Sep 17 00:00:00 2001 From: Feike Steenbergen Date: Mon, 26 Oct 2015 13:57:16 +0100 Subject: [PATCH] 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)