mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-08-25 14:53:37 +00:00
Merge branch 'master' into pgexperts-restore/movebasebackup
This commit is contained in:
Oleksii Kliukin
+3
-2
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
+5
-2
@@ -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):
|
||||
|
||||
|
||||
+613
@@ -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)
|
||||
+5
-1
@@ -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,
|
||||
|
||||
@@ -13,6 +13,10 @@ class PatroniException(Exception):
|
||||
return repr(self.value)
|
||||
|
||||
|
||||
class PatroniCtlException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class PostgresException(PatroniException):
|
||||
pass
|
||||
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
__version__ = '0.6'
|
||||
__version__ = '0.7'
|
||||
|
||||
Executable
+5
@@ -0,0 +1,5 @@
|
||||
#!/usr/bin/env python
|
||||
from patroni.ctl import ctl
|
||||
|
||||
if __name__ == '__main__':
|
||||
ctl()
|
||||
@@ -7,3 +7,5 @@ requests
|
||||
six >= 1.7
|
||||
kazoo>=2.2.1
|
||||
python-etcd>=0.4.1
|
||||
click>=4.1
|
||||
prettytable>=0.7
|
||||
|
||||
@@ -7,3 +7,5 @@ requests
|
||||
six
|
||||
kazoo>=2.2.1
|
||||
python-etcd>=0.4.1
|
||||
click>=4.1
|
||||
prettytable>=0.7
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
+6
-1
@@ -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)
|
||||
|
||||
+5
-1
@@ -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:[email protected]: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:[email protected]: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):
|
||||
|
||||
|
||||
@@ -56,6 +56,9 @@ class MockCursor:
|
||||
def fetchone(self):
|
||||
return self.results[0]
|
||||
|
||||
def fetchall(self):
|
||||
return self.results
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
|
||||
Reference in New Issue
Block a user