gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of github.com:zalando/patroni into feature/xlog_lag_interval

Browse Source
This commit is contained in:
Alexander Kukushkin
2016-02-17 12:41:09 +01:00
parent c2d46a084e f131961d8b
commit 7928a3a0d6
26 changed files with 476 additions and 446 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.travis.yml
+2 -1
View File
@@ -6,9 +6,10 @@ python:
install:
- if [[ $TRAVIS_PYTHON_VERSION == 2* ]]; then pip install -r requirements-py2.txt --use-mirrors; fi
- if [[ $TRAVIS_PYTHON_VERSION == 3* ]]; then pip install -r requirements-py3.txt; fi
- pip install coveralls
- pip install coveralls codacy-coverage
script:
- python setup.py test
- python setup.py flake8
after_success:
- coveralls
- python-codacy-coverage -r coverage.xml
patroni/__init__.py
+2 -2
View File
@@ -15,7 +15,7 @@ from .version import __version__
logger = logging.getLogger(__name__)
class Patroni:
class Patroni(object):
def __init__(self, config):
self.nap_time = config['loop_wait']
@@ -68,7 +68,7 @@ def main():
setup_signal_handlers()
if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]):
print('Usage: {} config.yml'.format(sys.argv[0]))
print('Usage: {0} config.yml'.format(sys.argv[0]))
return
with open(sys.argv[1], 'r') as f:
patroni/api.py
+6 -6
View File
@@ -172,8 +172,8 @@ class RestApiHandler(BaseHTTPRequestHandler):
def do_POST_failover(self):
content_length = int(self.headers.get('content-length', 0))
request = json.loads(self.rfile.read(content_length).decode('utf-8'))
leader = request.get('leader', None)
member = request.get('member', None)
leader = request.get('leader')
member = request.get('member')
cluster = self.server.patroni.ha.dcs.get_cluster()
status_code = 503
data = self.is_failover_possible(cluster, leader, member)
@@ -254,8 +254,8 @@ class RestApiHandler(BaseHTTPRequestHandler):
def get_tags(self):
return {'tags': self.server.patroni.tags}
def log_message(self, format, *args):
logger.debug("API thread: " + format % args)
def log_message(self, fmt, *args):
logger.debug("API thread: %s - - [%s] %s", self.client_address[0], self.log_date_time_string(), fmt % args)
class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
@@ -272,12 +272,12 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
# wrap socket with ssl if 'certfile' is defined in a config.yaml
# Sometime it's also needed to pass reference to a 'keyfile'.
options = {option: config[option] for option in ['certfile', 'keyfile'] if option in config}
if options.get('certfile', None):
if options.get('certfile'):
import ssl
self.socket = ssl.wrap_socket(self.socket, server_side=True, **options)
protocol = 'https'
self.connection_string = '{}://{}/patroni'.format(protocol, config.get('connect_address', config['listen']))
self.connection_string = '{0}://{1}/patroni'.format(protocol, config.get('connect_address', config['listen']))
self.patroni = patroni
self.daemon = True
patroni/async_executor.py
+2 -3
View File
@@ -4,10 +4,9 @@ from threading import Lock, Thread
logger = logging.getLogger(__name__)
class AsyncExecutor:
class AsyncExecutor(object):
def __init__(self):
Lock.__init__(self)
self._busy = False
self._thread_lock = Lock()
self._scheduled_action = None
@@ -51,5 +50,5 @@ class AsyncExecutor:
def __enter__(self):
self._thread_lock.acquire()
def __exit__(self, type, value, traceback):
def __exit__(self, *args):
self._thread_lock.release()
patroni/ctl.py
+60 -65
View File
@@ -56,12 +56,12 @@ def parse_dcs(dcs):
def load_config(path, dcs):
logging.debug('Loading configuration from file {}'.format(path))
logging.debug('Loading configuration from file %s', path)
config = dict()
try:
with open(path, 'rb') as fd:
config = yaml.safe_load(fd)
except:
except (IOError, yaml.YAMLError):
logging.exception('Could not load configuration file')
if dcs:
@@ -74,15 +74,14 @@ def load_config(path, dcs):
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)
if dir_path and 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_format = click.option('--format', '-f', 'fmt', 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')
@@ -102,20 +101,22 @@ 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)})
return Etcd(name=scope, config={'scope': scope, 'host': '{0}:{1}'.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'}):
def post_patroni(member, endpoint, content, headers=None):
url = urlparse(member.api_url)
logging.debug(url)
return requests.post('{}://{}/{}'.format(url.scheme, url.netloc, endpoint), headers=headers,
return requests.post('{0}://{1}/{2}'.format(url.scheme, url.netloc, endpoint),
headers=headers or {'Content-Type': 'application/json'},
data=json.dumps(content), timeout=60)
def print_output(columns, rows=[], alignment=None, format='pretty', header=True, delimiter='\t'):
if format == 'pretty':
def print_output(columns, rows=None, alignment=None, fmt='pretty', header=True, delimiter='\t'):
rows = rows or []
if fmt == 'pretty':
t = PrettyTable(columns)
for k, v in (alignment or {}).items():
t.align[k] = v
@@ -124,18 +125,18 @@ def print_output(columns, rows=[], alignment=None, format='pretty', header=True,
click.echo(t)
return
if format == 'json':
if fmt == 'json':
elements = list()
for r in rows:
elements.append(dict(zip(columns, r)))
click.echo(json.dumps(elements))
if format == 'tsv':
if fmt == 'tsv':
if columns is not None and header:
click.echo(delimiter.join(columns) + '\n')
for r in rows or []:
for r in rows:
c = [str(c) for c in r]
click.echo(delimiter.join(c))
@@ -169,9 +170,7 @@ def watching(w, watch, max_count=None, clear=True):
def build_connect_parameters(conn_url, connect_parameters=None):
if connect_parameters is None:
connect_parameters = {}
params = connect_parameters.copy()
params = (connect_parameters or {}).copy()
parsed = parseurl(conn_url)
params['host'] = parsed['host']
params['port'] = parsed['port']
@@ -203,13 +202,11 @@ def get_any_member(cluster, role='master', member=None):
def get_cursor(cluster, role='master', member=None, connect_parameters=None):
if connect_parameters is 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)
params = build_connect_parameters(member.conn_url, connect_parameters)
conn = psycopg2.connect(**params)
conn.autocommit = True
@@ -247,15 +244,15 @@ def dsn(cluster_name, config_file, dcs, role, member):
raise PatroniCtlException('Can not find a suitable member')
params = build_connect_parameters(m.conn_url)
click.echo('host={} port={}'.format(params['host'], params['port']))
click.echo('host={host} port={port}'.format(**params))
@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'))
@click.option('--format', 'fmt', help='Output format (pretty, json)', default='tsv')
@click.option('--file', '-f', 'p_file', help='Execute the SQL commands from this file', type=click.File('rb'))
@click.option('--password', help='force password prompt', is_flag=True)
@click.option('-U', '--username', help='database user name', type=str)
@option_dcs
@@ -277,21 +274,21 @@ def query(
watch,
delimiter,
command,
file,
p_file,
password,
username,
dbname,
format='tsv',
fmt='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:
if p_file is not None and command is not None:
raise PatroniCtlException('--file and --command are mutually exclusive options')
if file is None and command is None:
if p_file is None and command is None:
raise PatroniCtlException('You need to specify either --command or --file')
connect_parameters = dict()
@@ -302,8 +299,8 @@ def query(
if dbname:
connect_parameters['database'] = dbname
if file is not None:
command = file.read()
if p_file is not None:
command = p_file.read()
config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs)
@@ -312,24 +309,22 @@ def query(
output, cursor = query_member(cluster=cluster, cursor=cursor, member=member, role=role, command=command,
connect_parameters=connect_parameters)
print_output(None, output, format=format, delimiter=delimiter)
print_output(None, output, fmt=fmt, delimiter=delimiter)
if cursor is None:
cluster = dcs.get_cluster()
def query_member(cluster, cursor, member, role, command, connect_parameters=None):
if connect_parameters is None:
connect_parameters = {}
try:
if cursor is None:
cursor = get_cursor(cluster, role=role, member=member, connect_parameters=connect_parameters)
if cursor is None:
if role is None:
message = 'No connection to member {} is available'.format(member)
message = 'No connection to member {0} is available'.format(member)
else:
message = 'No connection to role={} is available'.format(role)
message = 'No connection to role={0} is available'.format(role)
logging.debug(message)
return [[timestamp(0), message]], None
@@ -348,7 +343,7 @@ def query_member(cluster, cursor, member, role, command, connect_parameters=None
cursor.connection.close()
message = oe.pgcode or oe.pgerror or str(oe)
message = message.replace('\n', ' ')
return [[timestamp(0), 'ERROR, SQLSTATE: {}'.format(message)]], None
return [[timestamp(0), 'ERROR, SQLSTATE: {0}'.format(message)]], None
@ctl.command('remove', help='Remove cluster from DCS')
@@ -356,13 +351,13 @@ def query_member(cluster, cursor, member, role, command, connect_parameters=None
@option_config_file
@option_format
@option_dcs
def remove(config_file, cluster_name, format, dcs):
def remove(config_file, cluster_name, fmt, 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)))
raise PatroniCtlException('We have not implemented this for DCS of type {0}'.format(type(dcs)))
output_members(cluster, format=format)
output_members(cluster, fmt=fmt)
confirm = click.prompt('Please confirm the cluster name to remove', type=str)
if confirm != cluster_name:
@@ -370,17 +365,17 @@ def remove(config_file, cluster_name, format, dcs):
message = 'Yes I am aware'
confirm = \
click.prompt('You are about to remove all information in DCS for {}, please type: "{}"'.format(cluster_name,
click.prompt('You are about to remove all information in DCS for {0}, please type: "{1}"'.format(cluster_name,
message), type=str)
if message != confirm:
raise PatroniCtlException('You did not exactly type "{}"'.format(message))
raise PatroniCtlException('You did not exactly type "{0}"'.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)
dcs.client.delete(dcs.client_path(''), recursive=True)
def wait_for_leader(dcs, timeout=30):
@@ -402,25 +397,25 @@ def empty_post_to_members(cluster, member_names, force, endpoint):
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,
if not member_names:
member_names = [click.prompt('Which member do you want to {0} [{1}]?'.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))
raise PatroniCtlException('{0} 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)))
confirm = click.confirm('Are you sure you want to {0} members {1}?'.format(endpoint, ', '.join(member_names)))
if not confirm:
raise PatroniCtlException('Aborted {}'.format(endpoint))
raise PatroniCtlException('Aborted {0}'.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))
click.echo('{0} failed for member {1}, status code={2}, ({3})'.format(endpoint, mn, r.status_code, r.text))
else:
click.echo('Succesful {} on member {}'.format(endpoint, mn))
click.echo('Succesful {0} on member {1}'.format(endpoint, mn))
def ctl_load_config(cluster_name, config_file, dcs):
@@ -436,21 +431,21 @@ def ctl_load_config(cluster_name, config_file, dcs):
@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)
@click.option('--any', 'p_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):
def restart(cluster_name, member_names, config_file, dcs, force, role, p_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:
if member_names:
member_names = list(set(member_names) & set(role_names))
else:
member_names = role_names
if any:
if p_any:
random.shuffle(member_names)
member_names = member_names[:1]
@@ -496,13 +491,13 @@ def failover(config_file, cluster_name, master, candidate, force, dcs):
master = click.prompt('Master', type=str, default=cluster.leader.member.name)
if cluster.leader.member.name != master:
raise PatroniCtlException('Member {} is not the leader of cluster {}'.format(master, cluster_name))
raise PatroniCtlException('Member {0} is not the leader of cluster {1}'.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:
if not candidate_names:
raise PatroniCtlException('No candidates found to failover to')
if candidate is None and not force:
@@ -512,7 +507,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs):
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))
raise PatroniCtlException('Member {0} does not exist in cluster {1}'.format(candidate, cluster_name))
# By now we have established that the leader exists and the candidate exists
click.echo('Current cluster topology')
@@ -520,12 +515,12 @@ def failover(config_file, cluster_name, master, candidate, force, dcs):
if not force:
a = \
click.confirm('Are you sure you want to failover cluster {}, demoting current master {}?'.format(
click.confirm('Are you sure you want to failover cluster {0}, demoting current master {1}?'.format(
cluster_name, master))
if not a:
raise PatroniCtlException('Aborting failover')
failover_value = '{}:{}'.format(master, candidate or '')
failover_value = '{0}:{1}'.format(master, candidate or '')
t_started = time.time()
r = None
@@ -535,16 +530,16 @@ def failover(config_file, cluster_name, master, candidate, force, dcs):
logging.debug(r)
logging.debug(r.text)
cluster = dcs.get_cluster()
click.echo(timestamp() + ' Failing over to new leader: {}'.format(cluster.leader.member.name))
click.echo(timestamp() + ' Failing over to new leader: {0}'.format(cluster.leader.member.name))
else:
click.echo('Failover failed, details: {}, {}'.format(r.status_code, r.text))
click.echo('Failover failed, details: {0}, {1}'.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))
click.echo(timestamp() + ' Initialized failover from master {0}'.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)
@@ -557,7 +552,7 @@ def failover(config_file, cluster_name, master, candidate, force, dcs):
output_members(cluster, name=cluster_name)
def output_members(cluster, name=None, format='pretty'):
def output_members(cluster, name=None, fmt='pretty'):
rows = []
logging.debug(cluster)
leader_name = None
@@ -602,7 +597,7 @@ def output_members(cluster, name=None, format='pretty'):
]
alignment = {'Cluster': 'l', 'Member': 'l', 'Host': 'l', 'Lag in MB': 'r'}
print_output(columns, rows, alignment, format)
print_output(columns, rows, alignment, fmt)
@ctl.command('list', help='List the Patroni members for a given Patroni')
@@ -612,8 +607,8 @@ def output_members(cluster, name=None, format='pretty'):
@option_watch
@option_watchrefresh
@option_dcs
def members(config_file, cluster_names, format, watch, w, dcs):
if len(cluster_names) == 0:
def members(config_file, cluster_names, fmt, watch, w, dcs):
if not cluster_names:
logging.warning('Listing members: No cluster names were provided')
return
@@ -622,7 +617,7 @@ def members(config_file, cluster_names, format, watch, w, dcs):
dcs = get_dcs(config, cn)
for _ in watching(w, watch):
output_members(dcs.get_cluster(), name=cn, format=format)
output_members(dcs.get_cluster(), name=cn, fmt=fmt)
def timestamp(precision=6):
patroni/dcs.py
+5 -5
View File
@@ -51,17 +51,17 @@ class Member(namedtuple('Member', 'index,name,session,data')):
else:
try:
data = json.loads(data)
except:
except (TypeError, ValueError):
data = {}
return Member(index, name, session, data)
@property
def conn_url(self):
return self.data.get('conn_url', None)
return self.data.get('conn_url')
@property
def api_url(self):
return self.data.get('api_url', None)
return self.data.get('api_url')
@property
def nofailover(self):
@@ -115,7 +115,7 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem
return any(m for m in self.members if m.name == member_name)
class AbstractDCS:
class AbstractDCS(object):
__metaclass__ = abc.ABCMeta
@@ -133,7 +133,7 @@ class AbstractDCS:
i.e.: `zookeeper` for zookeeper, `etcd` for etcd, etc...
"""
self._name = name
self._namespace = '/{}'.format(config.get('namespace', '/service/').strip('/'))
self._namespace = '/{0}'.format(config.get('namespace', '/service/').strip('/'))
self._base_path = '/'.join([self._namespace, config['scope']])
self._cluster = None
patroni/etcd.py
+16 -12
View File
@@ -51,7 +51,8 @@ class Client(etcd.Client):
def api_execute(self, path, method, **kwargs):
# Update machines_cache if previous attempt of update has failed
self._update_machines_cache and self._load_machines_cache()
if self._update_machines_cache:
self._load_machines_cache()
try:
return super(Client, self).api_execute(path, method, **kwargs)
except etcd.EtcdConnectionFailed:
@@ -73,7 +74,7 @@ class Client(etcd.Client):
except urllib3.exceptions.TimeoutError:
raise
except Exception as e:
raise etcd.EtcdException('Unable to decode server response: %s' % e)
raise etcd.EtcdException('Unable to decode server response: {0}'.format(e))
return super(Client, self)._result_from_response(response)
def _get_machines_cache_from_srv(self, discovery_srv):
@@ -83,7 +84,7 @@ class Client(etcd.Client):
ret = []
for host, port in self.get_srv_record(discovery_srv):
url = '{}://{}:{}/members'.format(self._protocol, host, port)
url = '{0}://{1}:{2}/members'.format(self._protocol, host, port)
try:
response = requests.get(url, timeout=5)
if response.ok:
@@ -101,10 +102,10 @@ class Client(etcd.Client):
host, port = addr.split(':')
try:
for r in set(socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)):
ret.append('{}://{}:{}'.format(self._protocol, r[4][0], r[4][1]))
ret.append('{0}://{1}:{2}'.format(self._protocol, r[4][0], r[4][1]))
except socket.error:
logger.exception('Can not resolve %s', host)
return list(set(ret)) if ret else ['{}://{}:{}'.format(self._protocol, host, port)]
return list(set(ret)) if ret else ['{0}://{1}:{2}'.format(self._protocol, host, port)]
def _load_machines_cache(self):
"""This method should fill up `_machines_cache` from scratch.
@@ -132,7 +133,9 @@ class Client(etcd.Client):
# After filling up initial list of machines_cache we should ask etcd-cluster about actual list
self._base_uri = self._machines_cache.pop(0)
self._machines_cache = self.machines
self._base_uri in self._machines_cache and self._machines_cache.remove(self._base_uri)
if self._base_uri in self._machines_cache:
self._machines_cache.remove(self._base_uri)
self._update_machines_cache = False
@@ -140,7 +143,7 @@ class Client(etcd.Client):
def catch_etcd_errors(func):
def wrapper(*args, **kwargs):
try:
return not func(*args, **kwargs) is None
return func(*args, **kwargs) is not None
except (RetryFailedError, etcd.EtcdException):
return False
except:
@@ -165,7 +168,8 @@ class Etcd(AbstractDCS):
def retry(self, *args, **kwargs):
return self._retry.copy()(*args, **kwargs)
def get_etcd_client(self, config):
@staticmethod
def get_etcd_client(config):
client = None
while not client:
try:
@@ -185,25 +189,25 @@ class Etcd(AbstractDCS):
nodes = {os.path.relpath(node.key, result.key): node for node in result.leaves}
# get initialize flag
initialize = nodes.get(self._INITIALIZE, None)
initialize = nodes.get(self._INITIALIZE)
initialize = initialize and initialize.value
# get last leader operation
last_leader_operation = nodes.get(self._LEADER_OPTIME, None)
last_leader_operation = nodes.get(self._LEADER_OPTIME)
last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation.value)
# get list of members
members = [self.member(n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1]
# get leader
leader = nodes.get(self._LEADER, None)
leader = nodes.get(self._LEADER)
if leader:
member = Member(-1, leader.value, None, {})
member = ([m for m in members if m.name == leader.value] or [member])[0]
leader = Leader(leader.modifiedIndex, leader.ttl, member)
# failover key
failover = nodes.get(self._FAILOVER, None)
failover = nodes.get(self._FAILOVER)
if failover:
failover = Failover.from_node(failover.modifiedIndex, failover.value)
patroni/ha.py
+11 -5
View File
@@ -11,7 +11,7 @@ from multiprocessing.pool import ThreadPool
logger = logging.getLogger(__name__)
class Ha:
class Ha(object):
def __init__(self, patroni):
self.patroni = patroni
@@ -110,9 +110,14 @@ class Ha:
refresh=True, recovery=True)
def follow(self, demote_reason, follow_reason, refresh=True, recovery=False):
refresh and self.load_cluster_from_dcs()
ret = demote_reason if (not recovery and self.state_handler.is_leader() or
recovery and self.state_handler.role == 'master') else follow_reason
if refresh:
self.load_cluster_from_dcs()
if not recovery and self.state_handler.is_leader() or recovery and self.state_handler.role == 'master':
ret = demote_reason
else:
ret = follow_reason
# determine the node to follow. If replicatefrom tag is set,
# try to follow the node mentioned there, otherwise, follow the leader.
if self.patroni.replicatefrom:
@@ -375,7 +380,8 @@ class Ha:
else:
return self._async_executor.scheduled_action + ' in progress'
def sysid_valid(self, sysid):
@staticmethod
def sysid_valid(sysid):
# sysid does tv_sec << 32, where tv_sec is the number of seconds sine 1970,
# so even 1 << 32 would have 10 digits.
return str(sysid) and len(str(sysid)) >= 10 and str(sysid).isdigit()
patroni/postgresql.py
+88 -71
View File
@@ -39,7 +39,7 @@ def parseurl(url):
return ret
class Postgresql:
class Postgresql(object):
def __init__(self, config):
self.config = config
@@ -52,7 +52,7 @@ class Postgresql:
self.superuser = config['superuser']
self.admin = config['admin']
self.initdb_options = config.get('initdb', [])
self.pgpass = config.get('pgpass', None) or os.path.join(os.path.expanduser('~'), 'pgpass')
self.pgpass = config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass')
self.pg_rewind = config.get('pg_rewind', {})
self.callback = config.get('callbacks', {})
self.use_slots = config.get('use_slots', True)
@@ -61,13 +61,13 @@ class Postgresql:
self.configuration_to_save = (os.path.join(self.data_dir, 'pg_hba.conf'),
os.path.join(self.data_dir, 'postgresql.conf'))
self.postmaster_pid = os.path.join(self.data_dir, 'postmaster.pid')
self.trigger_file = config.get('recovery_conf', {}).get('trigger_file', None) or 'promote'
self.trigger_file = config.get('recovery_conf', {}).get('trigger_file') or 'promote'
self.trigger_file = os.path.abspath(os.path.join(self.data_dir, self.trigger_file))
self._pg_ctl = ['pg_ctl', '-w', '-D', self.data_dir]
self.local_address = self.get_local_address()
connect_address = config.get('connect_address', None) or self.local_address
connect_address = config.get('connect_address') or self.local_address
self.connection_string = 'postgres://{username}:{password}@{connect_address}/postgres'.format(
connect_address=connect_address, **self.replication)
@@ -128,11 +128,18 @@ class Postgresql:
break
return local_address + ':' + self.port
@property
def _connect_kwargs(self):
r = parseurl('postgres://{0}/postgres'.format(self.local_address))
if 'username' in self.superuser:
r['user'] = self.superuser['username']
if 'password' in self.superuser:
r['password'] = self.superuser['password']
return r
def connection(self):
if not self._connection or self._connection.closed != 0:
r = parseurl('postgres://{}/postgres'.format(self.local_address))
r.update(self.superuser)
self._connection = psycopg2.connect(**r)
self._connection = psycopg2.connect(**self._connect_kwargs)
self._connection.autocommit = True
self.server_version = self._connection.server_version
return self._connection
@@ -173,32 +180,36 @@ class Postgresql:
@staticmethod
def initdb_allowed_option(name):
if name in ['pgdata', 'nosync', 'pwfile', 'sync-only']:
raise Exception('{} option for initdb is not allowed'.format(name))
raise Exception('{0} option for initdb is not allowed'.format(name))
return True
def get_initdb_options(self):
options = []
for o in self.initdb_options:
if isinstance(o, string_types) and self.initdb_allowed_option(o):
options.append('--{}'.format(o))
options.append('--{0}'.format(o))
elif isinstance(o, dict):
keys = list(o.keys())
if len(keys) != 1 or not isinstance(keys[0], string_types) or not self.initdb_allowed_option(keys[0]):
raise Exception('Invalid option: {}'.format(o))
options.append('--{}={}'.format(keys[0], o[keys[0]]))
raise Exception('Invalid option: {0}'.format(o))
options.append('--{0}={1}'.format(keys[0], o[keys[0]]))
else:
raise Exception('Unknown type of initdb option: {}'.format(o))
raise Exception('Unknown type of initdb option: {0}'.format(o))
return options
def initialize(self):
self.set_state('initalizing new cluster')
options = self.get_initdb_options()
pwfile = None
if self.superuser and 'username' not in self.superuser and 'password' in self.superuser:
(fd, pwfile) = tempfile.mkstemp()
os.write(fd, self.superuser['password'].encode())
os.close(fd)
options.append('--pwfile={}'.format(pwfile))
if self.superuser:
if 'username' in self.superuser:
options.append('--username={0}'.format(self.superuser['username']))
if 'password' in self.superuser:
(fd, pwfile) = tempfile.mkstemp()
os.write(fd, self.superuser['password'].encode())
os.close(fd)
options.append('--pwfile={0}'.format(pwfile))
ret = subprocess.call(self._pg_ctl + ['initdb'] + (['-o', ' '.join(options)] if options else [])) == 0
if pwfile:
@@ -210,7 +221,8 @@ class Postgresql:
return ret
def delete_trigger_file(self):
os.path.exists(self.trigger_file) and os.unlink(self.trigger_file)
if os.path.exists(self.trigger_file):
os.unlink(self.trigger_file)
def write_pgpass(self, record):
with open(self.pgpass, 'w') as f:
@@ -222,12 +234,11 @@ class Postgresql:
return env
def sync_replica(self, leader):
if leader:
r = parseurl(leader.conn_url)
env = self.write_pgpass(r) if leader else os.environ.copy()
ret = self.create_replica(leader, env) == 0
ret and self.delete_trigger_file()
return ret
env = self.write_pgpass(parseurl(leader.conn_url)) if leader else os.environ.copy()
if self.create_replica(leader, env) == 0:
self.delete_trigger_file()
return True
return False
@staticmethod
def build_connstring(conn):
@@ -235,7 +246,7 @@ class Postgresql:
>>> Postgresql.build_connstring({'host': '127.0.0.1', 'port': '5432'}) == 'host=127.0.0.1 port=5432'
True
"""
return ' '.join('{}={}'.format(param, val) for param, val in sorted(conn.items()))
return ' '.join('{0}={1}'.format(param, val) for param, val in sorted(conn.items()))
def replica_method_can_work_without_leader(self, method):
return method != 'basebackup' and self.config and self.config.get(method, {}).get('no_master')
@@ -245,10 +256,7 @@ class Postgresql:
that does not require a running leader to create the replica.
"""
replica_methods = self.config.get('create_replica_method', [])
for replica_method in replica_methods:
if self.replica_method_can_work_without_leader(replica_method):
return True
return False
return any(self.replica_method_can_work_without_leader(replica_method) for replica_method in replica_methods)
def create_replica(self, leader, env):
# create the replica according to the replica_method
@@ -313,7 +321,7 @@ class Postgresql:
cmd = self.callback[cb_name]
try:
subprocess.Popen(shlex.split(cmd) + [cb_name, self.role, self.scope])
except:
except OSError:
logger.exception('callback %s %s %s %s failed', cmd, cb_name, self.role, self.scope)
return False
return True
@@ -349,7 +357,10 @@ class Postgresql:
if not block_callbacks:
self.set_state('starting')
ret = subprocess.call(self._pg_ctl + ['start', '-o', self.server_options()]) == 0
env = os.environ.copy()
if 'username' in self.superuser:
env['PGUSER'] = self.superuser['username']
ret = subprocess.call(self._pg_ctl + ['start', '-o', self.server_options()], env=env, preexec_fn=os.setsid) == 0
self.set_state('running' if ret else 'start failed')
@@ -357,18 +368,21 @@ class Postgresql:
self.save_configuration_files()
# block_callbacks is used during restart to avoid
# running start/stop callbacks in addition to restart ones
ret and not block_callbacks and self.call_nowait(ACTION_ON_START)
if ret and not block_callbacks:
self.call_nowait(ACTION_ON_START)
return ret
def checkpoint(self, connstring=None):
def checkpoint(self, connect_kwargs=None):
connect_kwargs = connect_kwargs or self._connect_kwargs
for p in ['connect_timeout', 'options']:
connect_kwargs.pop(p, None)
try:
connstring = connstring or 'postgres://{}/postgres'.format(self.local_address)
with psycopg2.connect(connstring) as conn:
with psycopg2.connect(**connect_kwargs) as conn:
conn.autocommit = True
with conn.cursor() as cur:
cur.execute("SET statement_timeout = 0")
cur.execute('CHECKPOINT')
except:
except psycopg2.Error:
logging.exception('Exception during CHECKPOINT')
def stop(self, mode='fast', block_callbacks=False):
@@ -400,7 +414,8 @@ class Postgresql:
def reload(self):
ret = subprocess.call(self._pg_ctl + ['reload']) == 0
ret and self.call_nowait(ACTION_ON_RELOAD)
if ret:
self.call_nowait(ACTION_ON_RELOAD)
return ret
def restart(self):
@@ -409,13 +424,13 @@ class Postgresql:
if ret:
self.call_nowait(ACTION_ON_RESTART)
else:
self.set_state('restart failed ({})'.format(self.state))
self.set_state('restart failed ({0})'.format(self.state))
return ret
def server_options(self):
options = "--listen_addresses='{}' --port={}".format(self.listen_addresses, self.port)
options = "--listen_addresses='{0}' --port={1}".format(self.listen_addresses, self.port)
for setting, value in self.server_parameters.items():
options += " --{}='{}'".format(setting, value)
options += " --{0}='{1}'".format(setting, value)
return options
def is_healthy(self):
@@ -425,8 +440,7 @@ class Postgresql:
return True
def check_replication_lag(self, last_leader_operation):
return (last_leader_operation if last_leader_operation else 0) - self.xlog_position() <=\
self.config.get('maximum_lag_on_failover', 0)
return (last_leader_operation or 0) - self.xlog_position() <= self.config.get('maximum_lag_on_failover', 0)
def write_pg_hba(self):
with open(os.path.join(self.data_dir, 'pg_hba.conf'), 'a') as f:
@@ -459,28 +473,28 @@ class Postgresql:
recovery_target_timeline = 'latest'
""")
if leader and leader.conn_url:
f.write("""primary_conninfo = '{}'\n""".format(self.primary_conninfo(leader.conn_url)))
f.write("""primary_conninfo = '{0}'\n""".format(self.primary_conninfo(leader.conn_url)))
if self.use_slots:
f.write("""primary_slot_name = '{}'\n""".format(self.name))
f.write("""primary_slot_name = '{0}'\n""".format(self.name))
if (leader and leader.conn_url) or bootstrap:
for name, value in self.config.get('recovery_conf', {}).items():
f.write("{} = '{}'\n".format(name, value))
f.write("{0} = '{1}'\n".format(name, value))
def rewind(self, leader):
# prepare pg_rewind connection
r = parseurl(leader.conn_url)
r.update(self.pg_rewind)
r['user'] = r['username']
r['user'] = r.pop('username')
env = self.write_pgpass(r)
pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r)
# first run a checkpoint on a promoted master in order
# to make it store the new timeline ([email protected])
self.checkpoint(pc)
logger.info("running pg_rewind from {}".format(pc))
self.checkpoint(r)
logger.info("running pg_rewind from %s", pc)
pg_rewind = ['pg_rewind', '-D', self.data_dir, '--source-server', pc]
try:
ret = (subprocess.call(pg_rewind, env=env) == 0)
except:
ret = subprocess.call(pg_rewind, env=env) == 0
except OSError:
ret = False
if ret:
self.write_recovery_conf(leader)
@@ -496,8 +510,7 @@ recovery_target_timeline = 'latest'
result = {l.split(':')[0].replace('Current ', '', 1): l.split(':')[1].strip() for l in data if l}
except subprocess.CalledProcessError:
logger.exception("Error when calling pg_controldata")
finally:
return result
return result
def read_postmaster_opts(self):
""" returns the list of option names/values from postgres.opts, Empty dict if read failed or no file """
@@ -516,23 +529,24 @@ recovery_target_timeline = 'latest'
finally:
return result
def single_user_mode(self, command=None, options={}):
def single_user_mode(self, command=None, options=None):
""" run a given command in a single-user mode. If the command is empty - then just start and stop """
cmd = ['postgres', '--single', '-D', self.data_dir]
for opt in sorted(options):
cmd.extend(['-c', '{0}={1}'.format(opt, options[opt])])
for opt, val in sorted((options or {}).items()):
cmd.extend(['-c', '{0}={1}'.format(opt, val)])
# need a database name to connect
cmd.append('postgres')
p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT)
if p:
command and p.communicate('{}\n'.format(command))
if command:
p.communicate('{0}\n'.format(command))
p.stdin.close()
return p.wait()
return 1
def cleanup_archive_status(self):
status_dir = os.path.join(self.data_dir, 'pg_xlog', 'archive_status')
if os.path.isdir(status_dir):
try:
for f in os.listdir(status_dir):
path = os.path.join(status_dir, f)
try:
@@ -540,8 +554,10 @@ recovery_target_timeline = 'latest'
os.unlink(path)
elif os.path.isfile(path):
os.remove(path)
except:
logger.exception("Unable to remove {}".format(path))
except OSError:
logger.exception("Unable to remove %s", path)
except OSError:
logger.exception("Unable to list %s", status_dir)
def follow(self, leader, recovery=False):
if not self.check_recovery_conf(leader) or recovery:
@@ -579,7 +595,8 @@ recovery_target_timeline = 'latest'
self.remove_data_directory()
ret = True
self._need_rewind = False
change_role and ret and self.call_nowait(ACTION_ON_ROLE_CHANGE)
if change_role and ret:
self.call_nowait(ACTION_ON_ROLE_CHANGE)
return ret
else:
return True
@@ -592,16 +609,18 @@ recovery_target_timeline = 'latest'
"""
try:
for f in self.configuration_to_save:
os.path.isfile(f) and shutil.copy(f, f + '.backup')
except:
if os.path.isfile(f):
shutil.copy(f, f + '.backup')
except IOError:
logger.exception('unable to create backup copies of configuration files')
def restore_configuration_files(self):
""" restore a previously saved postgresql.conf """
try:
for f in self.configuration_to_save:
not os.path.isfile(f) and os.path.isfile(f + '.backup') and shutil.copy(f + '.backup', f)
except:
if not os.path.isfile(f) and os.path.isfile(f + '.backup'):
shutil.copy(f + '.backup', f)
except IOError:
logger.exception('unable to restore configuration files from backup')
def promote(self):
@@ -631,9 +650,7 @@ $$""".format(name, options), name, password, password)
def create_replication_user(self):
self.create_or_update_role(self.replication['username'], self.replication['password'], 'REPLICATION')
def create_connection_users(self):
if 'username' in self.superuser:
self.create_or_update_role(self.superuser['username'], self.superuser['password'], 'SUPERUSER')
def create_connection_user(self):
if self.admin:
self.create_or_update_role(self.admin['username'], self.admin['password'], 'CREATEDB CREATEROLE')
@@ -678,7 +695,7 @@ $$""".format(name, options), name, password, password)
WHERE slot_name = %s)""", slot, slot)
self.replication_slots = slots
except:
except psycopg2.Error:
logger.exception('Exception when changing replication slots')
def last_operation(self):
@@ -709,7 +726,7 @@ $$""".format(name, options), name, password, password)
ret = self.initialize() and self.start()
if ret:
self.create_replication_user()
self.create_connection_users()
self.create_connection_user()
else:
raise PostgresException("Could not bootstrap master PostgreSQL")
else:
@@ -725,7 +742,7 @@ $$""".format(name, options), name, password, password)
new_name = '{0}_{1}'.format(self.data_dir, time.strftime('%Y-%m-%d-%H-%M-%S'))
logger.info('renaming data directory to %s', new_name)
os.rename(self.data_dir, new_name)
except:
except OSError:
logger.exception("Could not rename data directory %s", self.data_dir)
def remove_data_directory(self):
@@ -739,7 +756,7 @@ $$""".format(name, options), name, password, password)
os.remove(self.data_dir)
elif os.path.isdir(self.data_dir):
shutil.rmtree(self.data_dir)
except:
except (IOError, OSError):
logger.exception('Could not remove data directory %s', self.data_dir)
self.move_data_directory()
patroni/scripts/aws.py
+3 -2
View File
@@ -9,7 +9,7 @@ import boto.ec2
logger = logging.getLogger(__name__)
class AWSConnection:
class AWSConnection(object):
def __init__(self, cluster_name):
self.available = False
self.cluster_name = cluster_name if cluster_name is not None else 'unknown'
@@ -56,7 +56,7 @@ class AWSConnection:
conn = boto.ec2.connect_to_region(self.region)
conn.create_tags([self.instance_id], tags)
except Exception as e:
logger.info("could not set tags for EC2 instance {}: {}".format(self.instance_id, e))
logger.info("could not set tags for EC2 instance %s: %s", self.instance_id, e)
return False
return True
@@ -66,6 +66,7 @@ class AWSConnection:
def main():
logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
if len(sys.argv) == 4 and sys.argv[1] in ('on_start', 'on_stop', 'on_role_change'):
AWSConnection(cluster_name=sys.argv[3]).on_role_change(sys.argv[2])
else:
patroni/scripts/wale_restore.py
+9 -14
View File
@@ -1,4 +1,4 @@
#!/usr/bin/python
#!/usr/bin/env python
# sample script to clone new replicas using WAL-E restore
# falls back to pg_basebackup if WAL-E restore fails, or if
@@ -36,7 +36,6 @@ import argparse
if sys.hexversion >= 0x03000000:
long = int
logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)
@@ -105,25 +104,20 @@ class WALERestore(object):
lsn_offset = hex((long(backup_start_segment[16:32], 16) << 24) + long(backup_start_offset))[2:-1]
# construct the LSN from the segment and offset
backup_start_lsn = '{}/{}'.format(lsn_segment, lsn_offset)
backup_start_lsn = '{0}/{1}'.format(lsn_segment, lsn_offset)
conn = None
cursor = None
diff_in_bytes = long(backup_size)
if not self.no_master:
try:
# get the difference in bytes between the current WAL location and the backup start offset
conn = psycopg2.connect(self.master_connection)
conn.autocommit = True
cursor = conn.cursor()
cursor.execute("SELECT pg_xlog_location_diff(pg_current_xlog_location(), %s)", (backup_start_lsn,))
diff_in_bytes = long(cursor.fetchone()[0])
with psycopg2.connect(self.master_connection) as con:
con.autocommit = True
with con.cursor() as cur:
cur.execute("SELECT pg_xlog_location_diff(pg_current_xlog_location(), %s)", (backup_start_lsn,))
diff_in_bytes = long(cur.fetchone()[0])
except psycopg2.Error as e:
logger.error('could not determine difference with the master location: {}'.format(e))
logger.error('could not determine difference with the master location: %s', e)
return False
finally:
cursor and cursor.close()
conn and conn.close()
else:
# always try to use WAL-E if base backup is available
diff_in_bytes = 0
@@ -145,6 +139,7 @@ class WALERestore(object):
def main():
logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
parser = argparse.ArgumentParser(description='Script to image replicas using WAL-E')
parser.add_argument('--scope', required=True)
parser.add_argument('--role', required=False)
patroni/utils.py
+16 -16
View File
@@ -8,9 +8,9 @@ import time
from patroni.exceptions import PatroniException
ignore_sigterm = False
interrupted_sleep = False
reap_children = False
__ignore_sigterm = False
__interrupted_sleep = False
__reap_children = False
_DATE_TIME_RE = re.compile(r'''^
(?P<year>\d{4})\-(?P<month>\d{2})\-(?P<day>\d{2}) # date
@@ -49,28 +49,28 @@ def calculate_ttl(expiration):
def sigterm_handler(signo, stack_frame):
global ignore_sigterm
if not ignore_sigterm:
ignore_sigterm = True
global __ignore_sigterm
if not __ignore_sigterm:
__ignore_sigterm = True
sys.exit()
def sigchld_handler(signo, stack_frame):
global interrupted_sleep, reap_children
reap_children = interrupted_sleep = True
global __interrupted_sleep, __reap_children
__reap_children = __interrupted_sleep = True
def sleep(interval):
global interrupted_sleep
global __interrupted_sleep
current_time = time.time()
end_time = current_time + interval
while current_time < end_time:
interrupted_sleep = False
__interrupted_sleep = False
time.sleep(end_time - current_time)
if not interrupted_sleep: # we will ignore only sigchld
if not __interrupted_sleep: # we will ignore only sigchld
break
current_time = time.time()
interrupted_sleep = False
__interrupted_sleep = False
def setup_signal_handlers():
@@ -79,8 +79,8 @@ def setup_signal_handlers():
def reap_children():
global reap_children
if reap_children:
global __reap_children
if __reap_children:
try:
while True:
ret = os.waitpid(-1, os.WNOHANG)
@@ -89,7 +89,7 @@ def reap_children():
except OSError:
pass
finally:
reap_children = False
__reap_children = False
class RetryFailedError(PatroniException):
@@ -97,7 +97,7 @@ class RetryFailedError(PatroniException):
"""Raised when retrying an operation ultimately failed, after retrying the maximum number of attempts."""
class Retry:
class Retry(object):
"""Helper for retrying a method in the face of retry-able exceptions"""
patroni/zookeeper.py
+6 -5
View File
@@ -17,7 +17,7 @@ class ZooKeeperError(DCSError):
pass
class ExhibitorEnsembleProvider:
class ExhibitorEnsembleProvider(object):
TIMEOUT = 3.1
@@ -54,7 +54,7 @@ class ExhibitorEnsembleProvider:
def _query_exhibitors(self, exhibitors):
random.shuffle(exhibitors)
for host in exhibitors:
uri = 'http://{}:{}{}'.format(host, self._exhibitor_port, self._uri_path)
uri = 'http://{0}:{1}{2}'.format(host, self._exhibitor_port, self._uri_path)
try:
response = requests.get(uri, timeout=self.TIMEOUT)
return response.json()
@@ -84,9 +84,9 @@ class ZooKeeper(AbstractDCS):
hosts = self.exhibitor.zookeeper_hosts
self.client = KazooClient(hosts=hosts,
timeout=(config.get('session_timeout', None) or 30),
timeout=(config.get('session_timeout') or 30),
command_retry={
'deadline': (config.get('reconnect_timeout', None) or 10),
'deadline': (config.get('reconnect_timeout') or 10),
'max_delay': 1,
'max_tries': -1},
connection_retry={'max_delay': 1, 'max_tries': -1})
@@ -190,7 +190,8 @@ class ZooKeeper(AbstractDCS):
def attempt_to_acquire_leader(self):
ret = self._create(self.leader_path, self._name, makepath=True, ephemeral=True)
ret or logger.info('Could not take out TTL lock')
if ret:
logger.info('Could not take out TTL lock')
return ret
def set_failover_value(self, value, index=None):
patronictl.py
+1 -1
View File
@@ -2,4 +2,4 @@
from patroni.ctl import ctl
if __name__ == '__main__':
ctl()
ctl(None)
postgres1.yml
+1 -1
View File
@@ -63,7 +63,7 @@ postgresql:
password: rep-pass
network: 127.0.0.1/32
superuser:
user: postgres
username: postgres
password: zalando
admin:
username: admin
postgres2.yml
+1 -1
View File
@@ -63,7 +63,7 @@ postgresql:
password: rep-pass
network: 127.0.0.1/32
superuser:
user: postgres
username: postgres
password: zalando
admin:
username: admin
tests/test_api.py
+16 -9
View File
@@ -19,10 +19,12 @@ class MockPostgresql(Mock):
server_version = '999999'
scope = 'dummy'
def connection(self):
@staticmethod
def connection():
return psycopg2_connect()
def is_running(self):
@staticmethod
def is_running():
return True
@@ -31,23 +33,28 @@ class MockHa(Mock):
dcs = Mock()
state_handler = MockPostgresql()
def schedule_restart(self):
@staticmethod
def schedule_restart():
return 'restart'
def schedule_reinitialize(self):
@staticmethod
def schedule_reinitialize():
return 'reinitialize'
def restart(self):
@staticmethod
def restart():
return (True, '')
def restart_scheduled(self):
@staticmethod
def restart_scheduled():
return False
def fetch_nodes_statuses(self, members):
@staticmethod
def fetch_nodes_statuses(members):
return [[None, True, None, None, {}]]
class MockPatroni:
class MockPatroni(Mock):
postgresql = MockPostgresql()
ha = MockHa()
@@ -56,7 +63,7 @@ class MockPatroni:
version = '0.00'
class MockRequest:
class MockRequest(object):
def __init__(self, path):
self.path = path
tests/test_aws.py
+18 -18
View File
@@ -1,12 +1,15 @@
import unittest
import requests
import boto.ec2
import requests
import sys
import unittest
from mock import Mock, patch
from collections import namedtuple
from patroni.scripts.aws import AWSConnection
from patroni.scripts.aws import AWSConnection, main as _main
from requests.exceptions import RequestException
class MockEc2Connection:
class MockEc2Connection(object):
def __init__(self, error=False):
self.error = error
@@ -23,7 +26,7 @@ class MockEc2Connection:
return True
class MockResponse:
class MockResponse(object):
def __init__(self, content):
self.content = content
@@ -35,15 +38,6 @@ class MockResponse:
class TestAWSConnection(unittest.TestCase):
def __init__(self, method_name='runTest'):
super(TestAWSConnection, self).__init__(method_name)
def set_error(self):
self.error = True
def set_json_error(self):
self.json_error = True
def boto_ec2_connect_to_region(self, region):
return MockEc2Connection(self.error)
@@ -74,21 +68,27 @@ class TestAWSConnection(unittest.TestCase):
self.assertTrue(self.conn.on_role_change('master'))
def test_non_aws(self):
self.set_error()
self.error = True
conn = AWSConnection('test')
self.assertFalse(conn.aws_available())
self.assertFalse(conn._tag_ebs('master'))
self.assertFalse(conn._tag_ec2('master'))
def test_aws_bizare_response(self):
self.set_json_error()
self.json_error = True
conn = AWSConnection('test')
self.assertFalse(conn.aws_available())
def test_aws_tag_ebs_error(self):
self.set_error()
self.error = True
self.assertFalse(self.conn._tag_ebs("master"))
def test_aws_tag_ec2_error(self):
self.set_error()
self.error = True
self.assertFalse(self.conn._tag_ec2("master"))
@patch('sys.exit', Mock())
def test_main(self):
self.assertIsNone(_main())
sys.argv = ['aws.py', 'on_start', 'replica', 'foo']
self.assertIsNone(_main())
tests/test_ctl.py
+85 -92
View File
@@ -8,7 +8,8 @@ import psycopg2
import requests
import patroni.exceptions
import etcd
from mock import patch, Mock
from mock import patch, Mock, MagicMock
from click.testing import CliRunner
from patroni.ctl import ctl, members, store_config, load_config, output_members, post_patroni, get_dcs, \
@@ -23,9 +24,10 @@ from test_postgresql import MockConnect, psycopg2_connect
CONFIG_FILE_PATH = './test-ctl.yaml'
def test_rw_config():
runner = CliRunner()
config = {'a':'b'}
config = {'a': 'b'}
with runner.isolated_filesystem():
store_config(config, CONFIG_FILE_PATH + '/dummy')
os.remove(CONFIG_FILE_PATH + '/dummy')
@@ -45,23 +47,25 @@ def test_rw_config():
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('socket.getaddrinfo', socket_getaddrinfo)
@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()
def setUp(self):
self.runner = CliRunner()
with patch.object(Client, 'machines') as 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):
@@ -80,9 +84,9 @@ class TestCtl(unittest.TestCase):
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')
output_members(cluster, name='abc', fmt='pretty')
output_members(cluster, name='abc', fmt='json')
output_members(cluster, name='abc', fmt='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))
@@ -92,49 +96,47 @@ class TestCtl(unittest.TestCase):
@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
result = self.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
result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
other
N''')
assert 'Aborting failover' in str(result.output)
result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
leader
y''')
assert 'target and source are the same' in str(result.output)
result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
Reality
y''')
assert 'Reality does not exist' in str(result.output)
result = runner.invoke(ctl, ['failover', 'dummy', '--force'])
result = self.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')
result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='dummy')
assert 'is not the leader of cluster' in str(result.output)
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
result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
other
y''')
assert 'No candidates found to failover to' in str(result.output)
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
result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
other
y''')
assert 'This cluster has no master' in str(result.output)
with patch('patroni.ctl.post_patroni', Mock(side_effect=Exception())):
result = runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
other
y''')
assert 'falling back to DCS' in result.output
@@ -143,27 +145,27 @@ y''')
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
result = self.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')
# result = self.runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='nonsense')
# assert 'is not the leader of cluster' in str(result.output)
# result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8', '--master', 'nonsense'])
# assert 'is not the leader of cluster' in str(result.output)
# result = runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nn')
# assert 'Aborting failover' in str(result.output)
# 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
#
# result = self.runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8', '--master', 'nonsense'])
# assert 'is not the leader of cluster' in str(result.output)
#
# result = self.runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nn')
# assert 'Aborting failover' in str(result.output)
#
# with patch('patroni.ctl.wait_for_leader', Mock(return_value = get_cluster_initialized_with_leader())):
# result = self.runner.invoke(ctl, ['failover', 'alpha', '--dcs', '8.8.8.8'], input='leader\nother\nY')
# assert 'master did not change after' in result.output
#
# result = self.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')
@@ -171,10 +173,8 @@ y''')
@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, [
result = self.runner.invoke(ctl, [
'query',
'alpha',
'--member',
@@ -184,18 +184,23 @@ y''')
])
assert 'mutually exclusive' in str(result.output)
with runner.isolated_filesystem():
dummy_file = open('dummy', 'w')
dummy_file.write('SELECT 1')
dummy_file.close()
with self.runner.isolated_filesystem():
with open('dummy', 'w') as dummy_file:
dummy_file.write('SELECT 1')
result = runner.invoke(ctl, [
result = self.runner.invoke(ctl, [
'query',
'alpha'
])
assert 'You need to specify' in str(result.output)
result = runner.invoke(ctl, [
result = self.runner.invoke(ctl, [
'query',
'alpha'
])
assert 'You need to specify' in str(result.output)
result = self.runner.invoke(ctl, [
'query',
'alpha',
'--file',
@@ -205,14 +210,15 @@ y''')
])
assert 'mutually exclusive' in str(result.output)
result = runner.invoke(ctl, ['query', 'alpha', '--file', 'dummy'])
result = self.runner.invoke(ctl, ['query', 'alpha', '--file', 'dummy'])
os.remove('dummy')
result = runner.invoke(ctl, ['query', 'alpha', '--command', 'SELECT 1'])
result = self.runner.invoke(ctl, ['query', 'alpha', '--command', 'SELECT 1'])
assert 'mock column' in result.output
result = runner.invoke(ctl, ['query', 'alpha', '--command', 'SELECT 1', '--dbname', 'dummy', '--password', '--username', 'dummy'], input='password\n')
result = self.runner.invoke(ctl, ['query', 'alpha', '--command', 'SELECT 1', '--dbname', 'dummy',
'--password', '--username', 'dummy'], input='password\n')
assert 'mock column' in result.output
@patch('patroni.ctl.get_cursor', Mock(return_value=MockConnect().cursor()))
@@ -238,13 +244,11 @@ y''')
@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'])
result = self.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, [
result = self.runner.invoke(ctl, [
'dsn',
'alpha',
'--role',
@@ -254,10 +258,10 @@ y''')
])
assert 'mutually exclusive' in str(result.output)
result = runner.invoke(ctl, ['dsn', 'alpha', '--member', 'dummy'])
result = self.runner.invoke(ctl, ['dsn', 'alpha', '--member', 'dummy'])
assert 'Can not find' in str(result.output)
# result = runner.invoke(ctl, ['dsn', 'alpha', '--dcs', '8.8.8.8', '--role', 'replica'])
# result = self.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()))
@@ -265,13 +269,11 @@ y''')
@patch('requests.get', requests_get)
@patch('requests.post', requests_get)
def test_restart_reinit(self):
runner = CliRunner()
result = self.runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='y')
result = self.runner.invoke(ctl, ['reinit', 'alpha', '--dcs', '8.8.8.8'], input='y')
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, [
result = self.runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='N')
result = self.runner.invoke(ctl, [
'restart',
'alpha',
'--dcs',
@@ -282,36 +284,34 @@ y''')
assert 'not a member' in str(result.output)
with patch('requests.post', Mock(return_value=MockResponse())):
result = runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='y')
result = self.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')
result = self.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.output)
result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='''alpha
result = self.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.output)
result = runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='beta\nleader')
result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='beta\nleader')
assert 'Cluster names specified do not match' in str(result.output)
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
result = self.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
result = self.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.output)
@@ -326,15 +326,14 @@ leader''')
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', {})
with patch('requests.post', MagicMock(side_effect=requests.exceptions.ConnectionError('foo'))):
member = get_cluster_initialized_with_leader().leader.member
self.assertRaises(requests.exceptions.ConnectionError, post_patroni, member, 'dummy', {})
def test_ctl(self):
runner = CliRunner()
self.runner.invoke(ctl, ['list'])
runner.invoke(ctl, ['list'])
result = runner.invoke(ctl, ['--help'])
result = self.runner.invoke(ctl, ['--help'])
assert 'Usage:' in result.output
def test_get_any_member(self):
@@ -364,15 +363,11 @@ leader''')
@patch('requests.get', requests_get)
@patch('requests.post', requests_get)
def test_members(self):
runner = CliRunner()
result = runner.invoke(members, ['alpha'])
result = self.runner.invoke(members, ['alpha'])
assert result.exit_code == 0
def test_configure(self):
runner = CliRunner()
result = runner.invoke(configure, [
result = self.runner.invoke(configure, [
'--dcs',
'abc',
'-c',
@@ -382,5 +377,3 @@ leader''')
])
assert result.exit_code == 0
tests/test_etcd.py
+10 -7
View File
@@ -11,7 +11,7 @@ from patroni.dcs import Cluster, DCSError, Leader
from patroni.etcd import Client, Etcd, EtcdError
class MockResponse:
class MockResponse(object):
def __init__(self):
self.status_code = 200
@@ -34,6 +34,7 @@ class MockResponse:
def status(self):
return self.status_code
@staticmethod
def getheader(*args):
return ''
@@ -43,7 +44,8 @@ class MockPostgresql(Mock):
server_version = '999999'
scope = 'dummy'
def last_operation(self):
@staticmethod
def last_operation():
return '0'
@@ -84,9 +86,9 @@ def etcd_watch(key, index=None, timeout=None, recursive=None):
def etcd_write(key, value, **kwargs):
if key == '/service/exists/leader':
raise etcd.EtcdAlreadyExist
if key == '/service/test/leader' or key == '/patroni/test/leader':
if kwargs.get('prevValue', None) == 'foo' or not kwargs.get('prevExist', True):
return True
if key in ['/service/test/leader', '/patroni/test/leader'] and \
(kwargs.get('prevValue') == 'foo' or not kwargs.get('prevExist', True)):
return True
raise etcd.EtcdException
@@ -127,12 +129,12 @@ class SleepException(Exception):
pass
class MockSRV:
class MockSRV(object):
port = 2380
target = '127.0.0.1'
def dns_query(name, type):
def dns_query(name, _):
if name == '_etcd-server._tcp.blabla':
return []
elif name == '_etcd-server._tcp.exception':
@@ -174,6 +176,7 @@ class TestClient(unittest.TestCase):
self.client._machines_cache = []
self.assertRaises(etcd.EtcdConnectionFailed, self.client.api_execute, '/', 'GET')
self.assertTrue(self.client._update_machines_cache)
self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', 'GET')
def test_get_srv_record(self):
self.assertEquals(self.client.get_srv_record('blabla'), [])
tests/test_ha.py
+38 -27
View File
@@ -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', 'xlog_location':4})
'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,7 @@ 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)
@@ -51,38 +52,48 @@ class MockPostgresql(Mock):
server_version = '999999'
scope = 'dummy'
def is_healthy(self):
@staticmethod
def is_healthy():
return True
def start(self):
@staticmethod
def start():
return True
def is_healthiest_node(self, members):
@staticmethod
def is_healthiest_node(members):
return True
def is_leader(self):
@staticmethod
def is_leader():
return True
def xlog_position(self):
@staticmethod
def xlog_position():
return 0
def last_operation(self):
@staticmethod
def last_operation():
return 0
def data_directory_empty(self):
@staticmethod
def data_directory_empty():
return False
def bootstrap(self, *args, **kwargs):
@staticmethod
def bootstrap(*args, **kwargs):
return True
def check_replication_lag(self, last_leader_operation):
@staticmethod
def check_replication_lag(last_leader_operation):
return True
def check_recovery_conf(self, leader):
@staticmethod
def check_recovery_conf(leader):
return False
class MockPatroni:
class MockPatroni(object):
def __init__(self, p, d):
self.postgresql = p
@@ -95,26 +106,26 @@ class MockPatroni:
def run_async(func, args=()):
func(*args) if args else func()
return func(*args) if args else func()
class TestHa(unittest.TestCase):
@patch('socket.getaddrinfo', socket_getaddrinfo)
@patch.object(Client, 'machines')
def setUp(self, mock_machines):
mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
self.p = MockPostgresql()
self.p.can_create_replica_without_leader = MagicMock(return_value=False)
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()
def setUp(self):
with patch.object(Client, 'machines') as mock_machines:
mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
self.p = MockPostgresql()
self.p.can_create_replica_without_leader = MagicMock(return_value=False)
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()
def test_update_lock(self):
self.p.last_operation = Mock(side_effect=PostgresException(''))
tests/test_patroni.py
+17 -17
View File
@@ -7,7 +7,7 @@ from mock import Mock, patch
from patroni.api import RestApiServer
from patroni.async_executor import AsyncExecutor
from patroni.etcd import Etcd
from patroni import Patroni, main
from patroni import Patroni, main as _main
from patroni.zookeeper import ZooKeeper
from six.moves import BaseHTTPServer
from test_etcd import Client, SleepException, etcd_read, etcd_write
@@ -28,19 +28,19 @@ def time_sleep(*args):
@patch.object(AsyncExecutor, 'run', Mock())
class TestPatroni(unittest.TestCase):
@patch.object(Client, 'machines')
def setUp(self, mock_machines):
mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
self.touched = False
self.init_cancelled = False
RestApiServer._BaseServer__is_shut_down = Mock()
RestApiServer._BaseServer__shutdown_request = True
RestApiServer.socket = 0
with open('postgres0.yml', 'r') as f:
config = yaml.load(f)
self.p = Patroni(config)
self.p.ha.dcs.client.write = etcd_write
self.p.ha.dcs.client.read = etcd_read
def setUp(self):
with patch.object(Client, 'machines') as mock_machines:
mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
self.touched = False
self.init_cancelled = False
RestApiServer._BaseServer__is_shut_down = Mock()
RestApiServer._BaseServer__shutdown_request = True
RestApiServer.socket = 0
with open('postgres0.yml', 'r') as f:
config = yaml.load(f)
self.p = Patroni(config)
self.p.ha.dcs.client.write = etcd_write
self.p.ha.dcs.client.read = etcd_read
@patch('patroni.zookeeper.KazooClient', MockKazooClient())
def test_get_dcs(self):
@@ -51,14 +51,14 @@ class TestPatroni(unittest.TestCase):
@patch.object(Etcd, 'delete_leader', Mock())
@patch.object(Client, 'machines')
def test_patroni_main(self, mock_machines):
main()
_main()
sys.argv = ['patroni.py', 'postgres0.yml']
mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
with patch.object(Patroni, 'run', Mock(side_effect=SleepException())):
self.assertRaises(SleepException, main)
self.assertRaises(SleepException, _main)
with patch.object(Patroni, 'run', Mock(side_effect=KeyboardInterrupt())):
main()
_main()
@patch('time.sleep', Mock(side_effect=SleepException()))
def test_run(self):
tests/test_postgresql.py
+37 -44
View File
@@ -19,7 +19,7 @@ def is_file_raise_on_backup(*args, **kwargs):
raise Exception("foo")
class MockCursor:
class MockCursor(object):
def __init__(self, connection):
self.connection = connection
@@ -59,7 +59,8 @@ class MockCursor:
def fetchall(self):
return self.results
def close(self):
@staticmethod
def close():
pass
def __iter__(self):
@@ -154,9 +155,12 @@ def psycopg2_connect(*args, **kwargs):
return MockConnect()
def fake_listdir(path):
return ["a", "b", "c"] if path.endswith('pg_xlog/archive_status') else []
@patch('subprocess.call', Mock(return_value=0))
@patch('psycopg2.connect', psycopg2_connect)
@patch('shutil.copy', Mock())
class TestPostgresql(unittest.TestCase):
@patch('subprocess.call', Mock(return_value=0))
@@ -165,7 +169,7 @@ class TestPostgresql(unittest.TestCase):
self.p = Postgresql({'name': 'test0', 'scope': 'batman', 'data_dir': 'data/test0',
'listen': '127.0.0.1, *:5432', 'connect_address': '127.0.0.2:5432',
'pg_hba': ['hostssl all all 0.0.0.0/0 md5', 'host all all 0.0.0.0/0 md5'],
'superuser': {'password': 'test'},
'superuser': {'username': 'test', 'password': 'test'},
'admin': {'username': 'admin', 'password': 'admin'},
'pg_rewind': {'username': 'admin', 'password': 'admin'},
'replication': {'username': 'replicator',
@@ -205,6 +209,11 @@ class TestPostgresql(unittest.TestCase):
self.assertTrue(self.p.initialize())
self.assertTrue(os.path.exists(os.path.join(self.p.data_dir, 'pg_hba.conf')))
@patch('os.path.exists', Mock(return_value=True))
@patch('os.unlink', Mock())
def test_delete_trigger_file(self):
self.p.delete_trigger_file()
def test_start(self):
self.assertTrue(self.p.start())
self.p.is_running = false
@@ -231,8 +240,10 @@ class TestPostgresql(unittest.TestCase):
@patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
def test_sync_replica(self):
self.assertTrue(self.p.sync_replica(self.leader))
self.p.create_replica = Mock(return_value=1)
self.assertFalse(self.p.sync_replica(self.leader))
@patch('subprocess.call', side_effect=Exception("Test"))
@patch('subprocess.call', side_effect=OSError)
@patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
def test_pg_rewind(self, mock_call):
self.assertTrue(self.p.rewind(self.leader))
@@ -243,6 +254,7 @@ class TestPostgresql(unittest.TestCase):
@patch('patroni.postgresql.Postgresql.remove_data_directory', MagicMock(return_value=True))
@patch('patroni.postgresql.Postgresql.single_user_mode', MagicMock(return_value=1))
@patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
@patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string))
def test_follow(self, mock_pg_rewind):
self.p.follow(None)
self.p.follow(self.leader)
@@ -263,6 +275,7 @@ class TestPostgresql(unittest.TestCase):
with mock.patch('patroni.postgresql.Postgresql.check_recovery_conf', MagicMock(return_value=True)):
self.assertTrue(self.p.follow(None))
@patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string))
def test_can_rewind(self):
tmp = self.p.pg_rewind
self.p.pg_rewind = None
@@ -272,7 +285,7 @@ class TestPostgresql(unittest.TestCase):
self.assertFalse(self.p.can_rewind)
with mock.patch('subprocess.call', side_effect=OSError("foo")):
self.assertFalse(self.p.can_rewind)
tmp = self.p.controldata()
tmp = self.p.controldata
self.p.controldata = lambda: {'wal_log_hints setting': 'on'}
self.assertTrue(self.p.can_rewind)
self.p.controldata = tmp
@@ -295,12 +308,6 @@ class TestPostgresql(unittest.TestCase):
with patch('subprocess.call', Mock(side_effect=Exception("foo"))):
self.assertEquals(self.p.create_replica(self.leader, ''), 1)
def test_create_connection_users(self):
cfg = self.p.config
cfg['superuser']['username'] = 'test'
p = Postgresql(cfg)
p.create_connection_users()
def test_sync_replication_slots(self):
self.p.start()
cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leadermem], None)
@@ -381,26 +388,20 @@ class TestPostgresql(unittest.TestCase):
open(self.p.data_dir, 'w').close()
self.p.remove_data_directory()
os.symlink('unexisting', self.p.data_dir)
with patch('os.unlink', Mock(side_effect=Exception)):
with patch('os.unlink', Mock(side_effect=OSError)):
self.p.remove_data_directory()
self.p.remove_data_directory()
@patch('subprocess.check_output', MagicMock(return_value=0, side_effect=pg_controldata_string))
@patch('subprocess.check_output', side_effect=subprocess.CalledProcessError)
@patch('subprocess.check_output', side_effect=Exception('Failed'))
def test_controldata(self, check_output_call_error, check_output_generic_exception):
data = self.p.controldata()
self.assertEquals(len(data), 50)
self.assertEquals(data['Database cluster state'], 'shut down in recovery')
self.assertEquals(data['wal_log_hints setting'], 'on')
self.assertEquals(int(data['Database block size']), 8192)
def test_controldata(self):
with patch('subprocess.check_output', Mock(return_value=0, side_effect=pg_controldata_string)):
data = self.p.controldata()
self.assertEquals(len(data), 50)
self.assertEquals(data['Database cluster state'], 'shut down in recovery')
self.assertEquals(data['wal_log_hints setting'], 'on')
self.assertEquals(int(data['Database block size']), 8192)
subprocess.check_output = check_output_call_error
data = self.p.controldata()
self.assertEquals(data, dict())
subprocess.check_output = check_output_generic_exception
self.assertRaises(Exception, self.p.controldata())
with patch('subprocess.check_output', Mock(side_effect=subprocess.CalledProcessError(1, ''))):
self.assertEquals(self.p.controldata(), {})
def test_read_postmaster_opts(self):
m = mock_open(read_data=postmaster_opts_string())
@@ -436,13 +437,7 @@ class TestPostgresql(unittest.TestCase):
subprocess_popen_mock.return_value = None
self.assertEquals(self.p.single_user_mode(), 1)
def fake_listdir(path):
if path.endswith(os.path.join('pg_xlog', 'archive_status')):
return ["a", "b", "c"]
return []
@patch('os.listdir', MagicMock(side_effect=fake_listdir))
@patch('os.path.isdir', MagicMock(return_value=True))
@patch('os.unlink', return_value=True)
@patch('os.remove', return_value=True)
@patch('os.path.islink', return_value=False)
@@ -464,8 +459,8 @@ class TestPostgresql(unittest.TestCase):
mock_unlink.reset_mock()
mock_remove.reset_mock()
mock_file.side_effect = Exception("foo")
mock_link.side_effect = Exception("foo")
mock_file.side_effect = OSError
mock_link.side_effect = OSError
self.p.cleanup_archive_status()
mock_unlink.assert_not_called()
mock_remove.assert_not_called()
@@ -474,16 +469,14 @@ class TestPostgresql(unittest.TestCase):
def test_sysid(self):
self.assertEqual(self.p.sysid, "6200971513092291716")
@patch('os.path.isfile', MagicMock(return_value=True))
@patch('shutil.copy', side_effect=Exception)
def test_save_configuration_files(self, mock_copy):
shutil.copy = mock_copy
@patch('os.path.isfile', Mock(return_value=True))
@patch('shutil.copy', Mock(side_effect=IOError))
def test_save_configuration_files(self):
self.p.save_configuration_files()
@patch('os.path.isfile', MagicMock(side_effect=is_file_raise_on_backup))
@patch('shutil.copy', side_effect=Exception)
def test_restore_configuration_files(self, mock_copy):
shutil.copy = mock_copy
@patch('os.path.isfile', Mock(side_effect=[False, True]))
@patch('shutil.copy', Mock(side_effect=IOError))
def test_restore_configuration_files(self):
self.p.restore_configuration_files()
def test_can_create_replica_without_leader(self):
tests/test_utils.py
+8 -10
View File
@@ -29,7 +29,8 @@ class TestUtils(unittest.TestCase):
@patch('time.sleep', Mock())
class TestRetrySleeper(unittest.TestCase):
def _fail(self, times=1):
@staticmethod
def _fail(times=1):
scope = dict(times=0)
def inner():
@@ -40,36 +41,33 @@ class TestRetrySleeper(unittest.TestCase):
raise PatroniException('Failed!')
return inner
def _makeOne(self, *args, **kwargs):
return Retry(*args, **kwargs)
def test_reset(self):
retry = self._makeOne(delay=0, max_tries=2)
retry = Retry(delay=0, max_tries=2)
retry(self._fail())
self.assertEquals(retry._attempts, 1)
retry.reset()
self.assertEquals(retry._attempts, 0)
def test_too_many_tries(self):
retry = self._makeOne(delay=0)
retry = Retry(delay=0)
self.assertRaises(RetryFailedError, retry, self._fail(times=999))
self.assertEquals(retry._attempts, 1)
def test_maximum_delay(self):
retry = self._makeOne(delay=10, max_tries=100)
retry = Retry(delay=10, max_tries=100)
retry(self._fail(times=10))
self.assertTrue(retry._cur_delay < 4000, retry._cur_delay)
# gevent's sleep function is picky about the type
self.assertEquals(type(retry._cur_delay), float)
def test_deadline(self):
retry = self._makeOne(deadline=0.0001)
retry = Retry(deadline=0.0001)
self.assertRaises(RetryFailedError, retry, self._fail(times=100))
def test_copy(self):
def _sleep(t):
None
pass
retry = self._makeOne(sleep_func=_sleep)
retry = Retry(sleep_func=_sleep)
rcopy = retry.copy()
self.assertTrue(rcopy.sleep_func is _sleep)
tests/test_wale_restore.py
+10 -6
View File
@@ -1,9 +1,9 @@
import unittest
from mock import MagicMock, patch, PropertyMock
import os
import psycopg2
import subprocess
from patroni.scripts.wale_restore import WALERestore, main
import unittest
from mock import MagicMock, patch, PropertyMock
from patroni.scripts.wale_restore import WALERestore, main as _main
def fake_cursor_fetchone(*args, **kwargs):
@@ -28,16 +28,19 @@ def fake_backup_data(self, *args, **kwargs):
base_00000001000000000000007F_00000040 2015-05-18T10:13:25.000Z 167772160 00000001000000000000007F 00000040 00000001000000000000007F 00000240
"""
def fake_backup_data_2(self, *args, **kwargs):
""" return the fake result of WAL-E backup-list"""
return """name last_modified expanded_size_bytes wal_segment_backup_start wal_segment_offset_backup_start wal_segment_backup_stop wal_segment_offset_backup_stop """
def fake_backup_data_3(self, *args, **kwargs):
""" return the fake result of WAL-E backup-list"""
return """name last_modified expanded_size_bytes wal_segment_backup_start wal_segment_offset_backup_start wal_segment_backup_stop
base_00000001000000000000007F_00000040 2015-05-18T10:13:25.000Z 167772160 00000001000000000000007F 00000040 00000001000000000000007F 00000240
"""
def fake_backup_data_4(self, *args, **kwargs):
""" return the fake result of WAL-E backup-list"""
return """name last_modified expanded_size_foo wal_segment_backup_start wal_segment_offset_backup_start wal_segment_backup_stop wal_segment_offset_backup_stop
@@ -92,6 +95,7 @@ class TestWALERestore(unittest.TestCase):
with patch.object(self.wale_restore, 'create_replica_with_s3', MagicMock(return_value=0)):
self.assertEqual(self.wale_restore.run(), 0)
@patch('sys.exit', MagicMock())
@patch.object(WALERestore, 'run', MagicMock(return_value=0))
def test_main(self):
with patch('sys.exit', MagicMock(return_value=0)):
self.assertEqual(main(), None)
self.assertEqual(_main(), None)
tests/test_zookeeper.py
+8 -6
View File
@@ -20,7 +20,8 @@ class MockKazooClient(Mock):
def client_id(self):
return (-1, '')
def retry(self, func, *args, **kwargs):
@staticmethod
def retry(func, *args, **kwargs):
func(*args, **kwargs)
def get(self, path, watch=None):
@@ -43,7 +44,8 @@ class MockKazooClient(Mock):
return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
return (b'', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
def get_children(self, path, watch=None, include_data=False):
@staticmethod
def get_children(path, watch=None, include_data=False):
if not isinstance(path, six.string_types):
raise TypeError("Invalid type for 'path' (string expected)")
if path.startswith('/no_node'):
@@ -62,16 +64,16 @@ class MockKazooClient(Mock):
elif value == b'retry' or (value == b'exists' and self.exists):
raise NodeExistsError
def set(self, path, value, version=-1):
@staticmethod
def set(path, value, version=-1):
if not isinstance(path, six.string_types):
raise TypeError("Invalid type for 'path' (string expected)")
if not isinstance(value, (six.binary_type,)):
raise TypeError("Invalid type for 'value' (must be a byte string)")
if path == '/service/bla/optime/leader':
raise Exception
if path == '/service/test/members/bar':
if value == b'retry':
return
if path == '/service/test/members/bar' and value == b'retry':
return
if path == '/service/test/failover':
if value == b'Exception':
raise Exception