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/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/api.py b/patroni/api.py index 466d7e32..47416b10 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -242,14 +242,17 @@ class RestApiHandler(BaseHTTPRequestHandler): } except (psycopg2.Error, RetryFailedError, PostgresConnectionException): state = self.server.patroni.postgresql.state - if state in ['stopped', 'starting', 'stopping', 'restarting', 'running']: + if state == 'running': logger.exception('get_postgresql_status') - state = 'unknown' if state == 'running' else state + state = 'unknown' return {'state': state} def get_tags(self): return {'tags': self.server.patroni.tags} + def log_message(self, format, *args): + logger.debug("API thread: " + format % args) + class RestApiServer(ThreadingMixIn, HTTPServer, Thread): diff --git a/patroni/ctl.py b/patroni/ctl.py new file mode 100644 index 00000000..a3c8aa73 --- /dev/null +++ b/patroni/ctl.py @@ -0,0 +1,613 @@ +''' +Patroni Control +''' + +import click +import os +import yaml +import json +import time +import psycopg2 +import random +import requests +import datetime +from prettytable import PrettyTable +from six.moves.urllib_parse import urlparse +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 = 'WARNING' + + +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') +option_force = click.option('--force', is_flag=True, help='Do not ask for confirmation at any point') + + +@click.group() +@click.pass_context +def ctl(ctx): + global LOGLEVEL + LOGLEVEL = os.environ.get('LOGLEVEL', LOGLEVEL) + + 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 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) + return requests.post('{}://{}/{}'.format(url.scheme, url.netloc, endpoint), headers=headers, + data=json.dumps(content), timeout=5) + + +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) + click.echo(t) + return + + if format == 'json': + elements = list() + for r in rows: + elements.append(dict(zip(columns, r))) + + 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, clear=True): + """ + >>> len(list(watching(True, 1, 0))) + 1 + >>> len(list(watching(True, 1, 1))) + 2 + >>> len(list(watching(True, None, 0))) + 1 + """ + + if w and not watch: + watch = 2 + if watch and clear: + 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 + 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': + if cluster.leader is not None: + yield cluster.leader + 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 + + +@ctl.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, 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) + if confirm != cluster_name: + 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 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') + + dcs.client.delete(dcs._base_path, recursive=True) + + +def wait_for_leader(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: + return cluster + + raise PatroniCtlException('Timeout occured') + + +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 {} [{}]?'.format(endpoint, + ', '.join(candidates.keys())), type=str, default='')] + + for mn in member_names: + if mn not in candidates.keys(): + 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, ', '.join(member_names))) + if not confirm: + raise PatroniCtlException('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)) + + +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) +@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, 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') + + +@ctl.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): + 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') +@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, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs) + + if cluster.leader is None: + raise PatroniCtlException('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 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 client + candidate_names.sort() + + if len(candidate_names) == 0: + 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 == 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)) + + # 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 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: + 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_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))) + 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 = '*' + + host = build_connect_parameters(m.conn_url)['host'] + + xlog_location = m.data.get('xlog_location') + lag = '' + if xlog_location is not None: + lag = round(((cluster.last_leader_operation or 0) - 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', 'Host': 'l'} + + print_output(columns, rows, alignment, format) + + +@ctl.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(precision=6): + return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:precision - 7] + + +@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/') +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..5f4fa0fb 100644 --- a/patroni/etcd.py +++ b/patroni/etcd.py @@ -143,6 +143,10 @@ def catch_etcd_errors(func): return not func(*args, **kwargs) is None except (RetryFailedError, etcd.EtcdException): return False + except: + logger.exception("") + raise EtcdError("unexpected error") + return wrapper @@ -150,7 +154,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/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/patroni/version.py b/patroni/version.py index 27fda16b..e220fa9e 100644 --- a/patroni/version.py +++ b/patroni/version.py @@ -1 +1 @@ -__version__ = '0.6' +__version__ = '0.7' diff --git a/patronictl.py b/patronictl.py new file mode 100755 index 00000000..5b06c153 --- /dev/null +++ b/patronictl.py @@ -0,0 +1,5 @@ +#!/usr/bin/env python +from patroni.ctl import ctl + +if __name__ == '__main__': + ctl() diff --git a/requirements-py2.txt b/requirements-py2.txt index fde9c79a..f8eacc12 100644 --- a/requirements-py2.txt +++ b/requirements-py2.txt @@ -7,3 +7,5 @@ requests 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 cc00965b..30b5ce96 100644 --- a/requirements-py3.txt +++ b/requirements-py3.txt @@ -7,3 +7,5 @@ requests six kazoo>=2.2.1 python-etcd>=0.4.1 +click>=4.1 +prettytable>=0.7 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): diff --git a/tests/test_ctl.py b/tests/test_ctl.py new file mode 100644 index 00000000..0d65a05c --- /dev/null +++ b/tests/test_ctl.py @@ -0,0 +1,376 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +import os +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, 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' + +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): + + @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') + + @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.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.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.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() + + 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, ['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.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) + + 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 = 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() + + runner.invoke(ctl, ['list']) + + 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 + + @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() + + result = runner.invoke(members, ['alpha']) + assert result.exit_code == 0 + + 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 915d5f9d..e9a334c8 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -8,7 +8,7 @@ import unittest from dns.exception import DNSException from mock import Mock, patch from patroni.dcs import Cluster, DCSError, Leader -from patroni.etcd import Client, Etcd +from patroni.etcd import Client, Etcd, EtcdError class MockResponse: @@ -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) @@ -266,3 +267,7 @@ class TestEtcd(unittest.TestCase): self.etcd.watch(4.5) self.etcd.watch(9.5) self.etcd.watch(100) + + @patch('patroni.etcd.Etcd.retry', Mock(side_effect=AttributeError("foo"))) + def test_other_exceptions(self): + self.assertRaises(EtcdError, self.etcd.cancel_initialization) diff --git a/tests/test_ha.py b/tests/test_ha.py index 43d4cded..d9a408a4 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 1eabcf8d..ba108c28 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