Merge remote-tracking branch 'refs/remotes/zalando/master'

This commit is contained in:
Jan Keirse
2016-04-05 16:44:21 +02:00
33 changed files with 1418 additions and 710 deletions
+22 -5
View File
@@ -1,15 +1,32 @@
sudo: required
language: python
addons:
postgresql: "9.5"
env:
global:
- BOTO_CONFIG='' ETCDVERSION=2.2.5
matrix:
- TEST_SUITE="python setup.py test"
- TEST_SUITE="behave"
python:
- "2.7"
- "3.4"
- "3.5"
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 codacy-coverage
- sudo /etc/init.d/postgresql stop
- sudo apt-get -y remove --purge postgresql-9.1 postgresql-9.2 postgresql-9.3 postgresql-9.4
- sudo apt-get -y autoremove
- sudo apt-key adv --keyserver keys.gnupg.net --recv-keys 7FCC7D46ACCC4CF8
- sudo sh -c 'echo "deb http://apt.postgresql.org/pub/repos/apt/ precise-pgdg main 9.5" >> /etc/apt/sources.list.d/postgresql.list'
- sudo apt-get update
- sudo apt-get -y install postgresql-9.5
- sudo /etc/init.d/postgresql stop
- pip install -r requirements.txt
- curl -L https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-amd64.tar.gz | tar xz -C . --strip=1 --wildcards --no-anchored etcd
- pip install behave codacy-coverage coverage coveralls
script:
- python setup.py test
- PATH=.:$PATH $TEST_SUITE
- python setup.py flake8
after_success:
- coveralls
- python-codacy-coverage -r coverage.xml
- if [[ -f coverage.xml ]]; then python-codacy-coverage -r coverage.xml; fi
+12
View File
@@ -0,0 +1,12 @@
approvals:
# PR needs at least 4 approvals
minimum: 1
# approval = comment that matches this regex
pattern: "^:?\\+1:?$"
from:
# commenter must be either one of:
# a public zalando org member
orgs:
- zalando
# a collaborator of the repo
collaborators: true
+14 -2
View File
@@ -1,4 +1,16 @@
Failover
========
- When determining who should become master, include the minor version of PostgreSQL in the decision
- Create a way to disable governance of a cluster, something like the existence of a "nogover" or "admin" file in PGDATA will stop governor from changing the cluster state
- When determining who should become master, include the minor version of PostgreSQL in the decision.
- Create a way to disable governance of a cluster, something like the existence of a "nogover" or "admin" file in PGDATA will stop patroni from changing the cluster state.
Configuration
==============
- Provide a way to change postgresql.conf and pg_hba.conf of a running cluster on the Patroni level, without changing individual nodes.
- Provide hooks to store and retrieve cluster-wide passwords without exposing them in a plain-text form to unauthorized users.
- Implement patronictl command to create initial configuration of the cluster with leader and member keys fixed to the user-supplied values in order to simplify migrations.
- Implement support for consul in addtion to etcd and zookeeper
- Complete zookeeper support in patronictl
Documentation
==============
- Document how to run cascading replication and possibly initialize the cluster without an access to the master node.
+17
View File
@@ -0,0 +1,17 @@
Feature: basic replication
We should check that the basic bootstrapping, replication and failover works.
Scenario: check replication of a single table
Given I start postgres0
And postgres0 is a leader after 10 seconds
And I start postgres1
When I add the table foo to postgres0
Then table foo is present on postgres1 after 15 seconds
Scenario: check the basic failover
When I kill postgres0
Then postgres1 role is the primary after 30 seconds
When I start postgres0
Then postgres0 role is the secondary after 15 seconds
When I add the table bar to postgres1
Then table bar is present on postgres0 after 10 seconds
+13
View File
@@ -0,0 +1,13 @@
Feature: cascading replication
We should check that patroni can do base backup and streaming from the replica
Scenario: check a base backup from the replica
Given I start postgres0
And postgres0 is a leader after 10 seconds
And I start postgres1
And replication works from postgres0 to postgres1 after 15 seconds
And I create label with "postgres0" in postgres0 data directory
And I create label with "postgres1" in postgres1 data directory
And I configure and start postgres2 with a tag clonefrom postgres1
Then replication works from postgres0 to postgres2 after 30 seconds
And there is a label with "postgres1" in postgres2 data directory
+305
View File
@@ -0,0 +1,305 @@
import os
import psycopg2
import requests
import shutil
import subprocess
import tempfile
import time
import yaml
class PatroniController(object):
PATRONI_CONFIG = '{}.yml'
""" starts and stops individual patronis"""
def __init__(self):
self._output_dir = None
self._patroni_path = None
self._connections = {}
self._config = {}
self._connstring = {}
self._cursors = {}
self._log = {}
self._processes = {}
@property
def patroni_path(self):
if self._patroni_path is None:
cwd = os.path.realpath(__file__)
while True:
path, entry = os.path.split(cwd)
cwd = path
if entry == 'features' or cwd == '/':
break
self._patroni_path = cwd
return self._patroni_path
def data_dir(self, pg_name):
return os.path.join(self.patroni_path, 'data', pg_name)
def write_label(self, pg_name, content):
with open(os.path.join(self.data_dir(pg_name), 'label'), 'w') as f:
f.write(content)
def read_label(self, pg_name):
content = None
try:
with open(os.path.join(self.data_dir(pg_name), 'label'), 'r') as f:
content = f.read()
except IOError:
return None
return content.strip()
def start(self, pg_name, max_wait_limit=20, tags=None):
if not self._is_running(pg_name):
if pg_name in self._processes:
del self._processes[pg_name]
cwd = self.patroni_path
self._log[pg_name] = open(os.path.join(self._output_dir, 'patroni_{0}.log'.format(pg_name)), 'a')
self._config[pg_name] = self._make_patroni_test_config(pg_name, tags=tags)
p = subprocess.Popen(['coverage', 'run', '--branch', '--source=patroni', '-p', 'patroni.py', self._config[pg_name]],
stdout=self._log[pg_name], stderr=subprocess.STDOUT, cwd=cwd)
if not (p and p.pid and p.poll() is None):
assert False, "PostgreSQL {0} is not running after being started".format(pg_name)
self._processes[pg_name] = p
# wait while patroni is available for queries, but not more than 10 seconds.
for _ in range(max_wait_limit):
if self.query(pg_name, "SELECT 1", fail_ok=True) is not None:
break
time.sleep(1)
else:
assert False,\
"Patroni instance is not available for queries after {0} seconds".format(max_wait_limit)
def stop(self, pg_name, kill=False, timeout=15):
start_time = time.time()
while self._is_running(pg_name):
if not kill:
self._processes[pg_name].terminate()
else:
self._processes[pg_name].kill()
time.sleep(1)
if not kill and time.time() - start_time > timeout:
kill = True
if self._log.get('pg_name') and not self._log['pg_name'].closed:
self._log[pg_name].close()
if pg_name in self._processes:
del self._processes[pg_name]
def query(self, pg_name, query, fail_ok=False):
try:
cursor = self._cursor(pg_name)
cursor.execute(query)
return cursor
except psycopg2.Error:
if fail_ok:
return None
else:
raise
def check_role_has_changed_to(self, pg_name, new_role, timeout=10):
bound_time = time.time() + timeout
recovery_status = False if new_role == 'primary' else True
role_has_changed = False
while not role_has_changed:
cur = self.query(pg_name, "SELECT pg_is_in_recovery()", fail_ok=True)
if cur:
row = cur.fetchone()
if row and len(row) > 0 and row[0] == recovery_status:
role_has_changed = True
if time.time() > bound_time:
break
time.sleep(1)
return role_has_changed
def stop_all(self):
for patroni in self._processes.copy():
self.stop(patroni)
def create_and_set_output_directory(self, feature_name):
feature_dir = os.path.join(self.patroni_path, "features", "output",
feature_name.replace(' ', '_'))
if os.path.exists(feature_dir):
shutil.rmtree(feature_dir)
os.makedirs(feature_dir)
self._output_dir = feature_dir
def _is_running(self, pg_name):
return pg_name in self._processes and self._processes[pg_name].pid and (self._processes[pg_name].poll() is None)
def _make_patroni_test_config(self, pg_name, tags=None):
patroni_config_name = PatroniController.PATRONI_CONFIG.format(pg_name)
patroni_config_path = os.path.join(self._output_dir, patroni_config_name)
with open(patroni_config_name) as f:
config = yaml.load(f)
postgresql = config['postgresql']
postgresql['name'] = pg_name
postgresql['data_dir'] = 'data/{0}'.format(pg_name)
postgresql_params = postgresql['parameters']
postgresql_params['logging_collector'] = 'on'
postgresql_params['log_destination'] = 'csvlog'
postgresql_params['log_directory'] = self._output_dir
postgresql_params['log_filename'] = '{0}.log'.format(pg_name)
postgresql_params['log_statement'] = 'all'
postgresql_params['log_min_messages'] = 'debug1'
postgresql_params['unix_socket_directories'] = '.'
if tags:
config['tags'] = tags
with open(patroni_config_path, 'w') as f:
yaml.dump(config, f, default_flow_style=False)
return patroni_config_path
def _make_connstring(self, pg_name):
if pg_name in self._connstring:
return self._connstring[pg_name]
try:
patroni_path = self.patroni_path
with open(os.path.join(patroni_path, PatroniController.PATRONI_CONFIG.format(pg_name)), 'r') as f:
config = yaml.load(f)
except IOError:
return None
connstring = config['postgresql']['connect_address']
if ':' in connstring:
address, port = connstring.split(':')
else:
address = connstring
port = '5432'
user = "postgres"
dbname = "postgres"
self._connstring[pg_name] = "host={0} port={1} dbname={2} user={3}".format(address, port, dbname, user)
return self._connstring[pg_name]
def _connection(self, pg_name):
if pg_name not in self._connections or self._connections[pg_name].closed:
conn = psycopg2.connect(self._make_connstring(pg_name))
conn.autocommit = True
self._connections[pg_name] = conn
return self._connections[pg_name]
def _cursor(self, pg_name):
if pg_name not in self._cursors or self._cursors[pg_name].closed:
cursor = self._connection(pg_name).cursor()
self._cursors[pg_name] = cursor
return self._cursors[pg_name]
class EtcdController(object):
""" handles all etcd related tasks, used for the tests setup and cleanup """
ETCD_VERSION_URL = 'http://127.0.0.1:2379/version'
ETCD_CLEANUP_URL = 'http://127.0.0.1:2379/v2/keys/service/batman?recursive=true'
def __init__(self, log_directory):
self.handle = None
self.work_directory = None
self.log_directory = log_directory
self.log_file = None
self.pid = None
self.start_timeout = 5
def start(self):
""" start etcd if it's not already running """
if self._is_running():
return True
self.work_directory = tempfile.mkdtemp()
# etcd is running throughout the tests, no need to append to the log
output_dir = os.path.join(self.log_directory, "features", "output")
if not os.path.exists(output_dir):
os.makedirs(output_dir)
self.log_file = open(os.path.join(output_dir, 'etcd.log'), 'w')
self.handle =\
subprocess.Popen(["etcd", "--debug", "--data-dir", self.work_directory],
stdout=self.log_file, stderr=subprocess.STDOUT)
start_time = time.time()
while (not self._is_running()):
if time.time() - start_time > self.start_timeout:
assert False, "Failed to start etcd"
time.sleep(1)
return True
@staticmethod
def query(key):
""" query etcd for a value of a given key """
r = requests.get("http://127.0.0.1:2379/v2/keys/service/batman/{0}".format(key))
if r.ok:
content = r.json()
if content:
return content.get('node', {}).get('value')
return None
def stop_and_remove_work_directory(self, timeout=15):
""" terminate etcd and wipe out the temp work directory, but only if we actually started it"""
kill = False
start_time = time.time()
while self._is_running() and self.handle:
if not kill:
self.handle.terminate()
else:
self.handle.kill()
time.sleep(1)
if not kill and time.time() - start_time > timeout:
kill = True
self.handle = None
if self.log_file and not self.log_file.closed:
self.log_file.close()
if self.work_directory:
shutil.rmtree(self.work_directory)
self.work_directory = None
@staticmethod
def cleanup_service_tree():
""" clean all contents stored in the tree used for the tests """
r = None
try:
r = requests.delete(EtcdController.ETCD_CLEANUP_URL)
if r and not r.ok:
assert False,\
"request to cleanup the etcd contents was not successfull: status code {0}".format(r.status_code)
except requests.exceptions.RequestException as e:
assert False, "exception when cleaning up etcd contents: {0}".format(e)
@staticmethod
def _is_running():
# if etcd is running, but we didn't start it
try:
r = requests.get(EtcdController.ETCD_VERSION_URL)
running = (r and r.ok and b'etcdserver' in r.content)
except requests.ConnectionError:
running = False
return running
# actions to execute on start/stop of the tests and before running invidual features
def before_all(context):
context.pctl = PatroniController()
context.etcd_ctl = EtcdController(context.pctl.patroni_path)
context.etcd_ctl.start()
try:
context.etcd_ctl.cleanup_service_tree()
except AssertionError: # after.all handlers won't be executed in before.all
context.etcd_ctl.stop_and_remove_work_directory()
raise
def after_all(context):
context.etcd_ctl.stop_and_remove_work_directory()
subprocess.call(['coverage', 'combine'])
subprocess.call(['coverage', 'report'])
def before_feature(context, feature):
""" create per-feature output directory to collect Patroni and PostgreSQL logs """
context.pctl.create_and_set_output_directory(feature.name)
def after_feature(context, feature):
""" stop all Patronis, remove their data directory and cleanup the keys in etcd """
context.pctl.stop_all()
shutil.rmtree(os.path.join(context.pctl.patroni_path, 'data'))
context.etcd_ctl.cleanup_service_tree()
+51
View File
@@ -0,0 +1,51 @@
Feature: patroni api
We should check that patroni correctly responds to valid and not-valid API requests.
Scenario: check API requests on a stand-alone server
Given I start postgres0
And postgres0 is a leader after 10 seconds
When I issue a GET request to http://127.0.0.1:8008/
Then I receive a response code 200
And I receive a response state running
And I receive a response role master
When I issue a GET request to http://127.0.0.1:8008/replica
Then I receive a response code 503
When I issue an empty POST request to http://127.0.0.1:8008/reinitialize
Then I receive a response code 503
And I receive a response text "I am the leader, can not reinitialize"
When I issue a POST request to http://127.0.0.1:8008/failover with leader=postgres0
Then I receive a response code 500
And I receive a response text "failover is not possible: cluster does not have members except leader"
When I issue an empty POST request to http://127.0.0.1:8008/failover
Then I receive a response code 400
And I receive a response text "No values given for required parameters leader and candidate"
Scenario: check API requests for the primary-replica pair
Given I start postgres1
And replication works from postgres0 to postgres1 after 15 seconds
When I issue a GET request to http://127.0.0.1:8009/replica
Then I receive a response code 200
And I receive a response state running
And I receive a response role replica
When I issue an empty POST request to http://127.0.0.1:8009/reinitialize
Then I receive a response code 200
When I issue an empty POST request to http://127.0.0.1:8008/restart
Then I receive a response code 200
And postgres0 is a leader after 5 seconds
When I sleep for 10 seconds
Then postgres1 role is the secondary after 15 seconds
Scenario: check the failover via the API
Given I issue a POST request to http://127.0.0.1:8008/failover with leader=postgres0,candidate=postgres1
Then I receive a response code 200
And postgres1 is a leader after 5 seconds
And postgres1 role is the primary after 5 seconds
And postgres0 role is the secondary after 5 seconds
And replication works from postgres1 to postgres0 after 15 seconds
Scenario: check the scheduled failover
Given I issue a scheduled failover at http://127.0.0.1:8009 from postgres1 to postgres0 in 10 seconds
Then I receive a response code 200
And postgres0 is a leader after 20 seconds
And replication works from postgres0 to postgres1 after 25 seconds
+55
View File
@@ -0,0 +1,55 @@
import psycopg2 as pg
from behave import step, then
from time import sleep, time
@step('I start {name:w}')
def start_patroni(context, name):
return context.pctl.start(name)
@step('I shut down {name:w}')
def stop_patroni(context, name):
return context.pctl.stop(name)
@step('I kill {name:w}')
def kill_patroni(context, name):
return context.pctl.stop(name, kill=True)
@step('I add the table {table_name:w} to {pg_name:w}')
def add_table(context, table_name, pg_name):
# parse the configuration file and get the port
try:
context.pctl.query(pg_name, "CREATE TABLE {0}()".format(table_name))
except pg.Error as e:
assert False, "Error creating table {0} on {1}: {2}".format(table_name, pg_name, e)
@then('Table {table_name:w} is present on {pg_name:w} after {max_replication_delay:d} seconds')
def table_is_present_on(context, table_name, pg_name, max_replication_delay):
for _ in range(int(max_replication_delay)):
if context.pctl.query(pg_name, "SELECT 1 FROM {0}".format(table_name), fail_ok=True) is not None:
break
sleep(1)
else:
assert False,\
"Table {0} is not present on {1} after {2} seconds".format(table_name, pg_name, max_replication_delay)
@then('{pg_name:w} role is the {pg_role:w} after {max_promotion_timeout:d} seconds')
def check_role(context, pg_name, pg_role, max_promotion_timeout):
if not context.pctl.check_role_has_changed_to(pg_name, pg_role, timeout=int(max_promotion_timeout)):
assert False,\
"{0} role didn't change to {1} after {2} seconds".format(pg_name, pg_role, max_promotion_timeout)
@step('replication works from {master:w} to {replica:w} after {time_limit:d} seconds')
@then('replication works from {master:w} to {replica:w} after {time_limit:d} seconds')
def replication_works(context, master, replica, time_limit):
context.execute_steps(u"""
When I add the table test_{0} to {1}
Then table test_{0} is present on {2} after {3} seconds
""".format(int(time()), master, replica, time_limit))
+17
View File
@@ -0,0 +1,17 @@
from behave import step, then
@step('I configure and start {name:w} with a tag {tag_name:w} {tag_value:w}')
def start_patroni_with_a_name_value_tag(context, name, tag_name, tag_value):
return context.pctl.start(name, tags={tag_name: tag_value})
@then('There is a label with "{content:w}" in {name:w} data directory')
def check_label(context, content, name):
label = context.pctl.read_label(name)
assert label == content, "{0} is not equal to {1}".format(label, content)
@step('I create label with "{content:w}" in {name:w} data directory')
def write_label(context, content, name):
context.pctl.write_label(name, content)
+101
View File
@@ -0,0 +1,101 @@
import parse
import pytz
import requests
import time
from behave import register_type, step, then
from datetime import datetime, timedelta
@parse.with_pattern(r'https?://(?:\w|\.|:|/)+')
def parse_url(text):
return text
@parse.with_pattern(r'(?:\w+=(?:\w|\.|:|-|\+|\s)+,?)+')
def parse_data(text):
return text
register_type(url=parse_url, data=parse_data)
# there is no way we can find out if the node has already
# started as a leader without checking the DCS. We cannot
# just rely on the database availability, since there is
# a short gap between the time PostgreSQL becomes available
# and Patroni assuming the leader role.
@step('{name:w} is a leader after {time_limit:d} seconds')
@then('{name:w} is a leader after {time_limit:d} seconds')
def is_a_leader(context, name, time_limit):
max_time = time.time() + int(time_limit)
while (context.etcd_ctl.query("leader") != name):
time.sleep(1)
if time.time() > max_time:
assert False, "{0} is not a leader in etcd after {1} seconds".format(name, time_limit)
@step('I sleep for {value:d} seconds')
def sleep_for_n_seconds(context, value):
time.sleep(int(value))
@step('I issue a GET request to {url:url}')
def do_get(context, url):
try:
r = requests.get(url)
except requests.exceptions.RequestException:
context.status_code = None
context.response = None
else:
context.status_code = r.status_code
try:
context.response = r.json()
except ValueError:
context.response = r.content.decode('utf-8')
@step('I issue an empty POST request to {url:url}')
def do_post_empty(context, url):
do_post(context, url, None)
@step('I issue a POST request to {url:url} with {data:data}')
def do_post(context, url, data):
post_data = {}
if data:
post_components = data.split(',')
for pc in post_components:
if '=' in pc:
k, v = pc.split('=', 2)
post_data[k.strip()] = v.strip()
try:
r = requests.post(url, json=post_data)
except requests.exceptions.RequestException:
context.status_code = None
context.response = None
else:
context.status_code = r.status_code
try:
context.response = r.json()
except ValueError:
context.response = r.content.decode('utf-8')
@then('I receive a response {component:w} {data}')
def check_response(context, component, data):
if component == 'code':
assert context.status_code == int(data),\
"status code {0} != {1}, response: {2}".format(context.status_code, int(data), context.response)
elif component == 'text':
assert context.response == data.strip('"'), "response {0} does not contain {1}".format(context.response, data)
else:
assert component in context.response, "{0} is not part of the response".format(component)
assert context.response[component] == data, "{0} does not contain {1}".format(component, data)
@step('I issue a scheduled failover at {at_url:url} from {from_host:w} to {to_host:w} in {in_seconds:d} seconds')
def scheduled_failover(context, at_url, from_host, to_host, in_seconds):
context.execute_steps(u"""
Given I issue a POST request to {0}/failover with leader={1},candidate={2},scheduled_at={3}
""".format(at_url, from_host, to_host, datetime.now(pytz.utc) + timedelta(seconds=int(in_seconds))))
+22 -6
View File
@@ -16,6 +16,7 @@ logger = logging.getLogger(__name__)
class Patroni(object):
PATRONI_CONFIG_VARIABLE = 'PATRONI_CONFIGURATION'
def __init__(self, config):
self.nap_time = config['loop_wait']
@@ -35,6 +36,10 @@ class Patroni(object):
def replicatefrom(self):
return self.tags.get('replicatefrom')
@property
def clonefrom(self):
return self.tags.get('clonefrom')
@staticmethod
def get_dcs(name, config):
if 'etcd' in config:
@@ -67,12 +72,23 @@ def main():
logging.getLogger('requests').setLevel(logging.WARNING)
setup_signal_handlers()
if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]):
print('Usage: {0} config.yml'.format(sys.argv[0]))
return
# Patroni reads the configuration from the command-line argument if it exists, and from the environment otherwise.
use_env = False
use_file = (len(sys.argv) >= 2 and os.path.isfile(sys.argv[1]))
if not use_file:
config_env = os.environ.get(Patroni.PATRONI_CONFIG_VARIABLE)
use_env = config_env is not None
if not use_env:
print('Usage: {0} config.yml'.format(sys.argv[0]))
print('\tPatroni may also read the configuration from the {} environment variable'.
format(Patroni.PATRONI_CONFIG_VARIABLE))
return
with open(sys.argv[1], 'r') as f:
config = yaml.load(f)
if use_file:
with open(sys.argv[1], 'r') as f:
config = yaml.load(f)
elif use_env:
config = yaml.load(config_env)
patroni = Patroni(config)
try:
@@ -81,5 +97,5 @@ def main():
pass
finally:
patroni.api.shutdown()
patroni.postgresql.stop()
patroni.postgresql.stop(checkpoint=False)
patroni.dcs.delete_leader()
+62 -22
View File
@@ -5,6 +5,9 @@ import logging
import psycopg2
import socket
import time
import dateutil
import datetime
import pytz
from patroni.exceptions import PostgresConnectionException
from patroni.utils import Retry, RetryFailedError
@@ -101,13 +104,13 @@ class RestApiHandler(BaseHTTPRequestHandler):
@check_auth
def do_POST_restart(self):
status_code = 503
status_code = 500
data = b'restart failed'
try:
status, msg = self.server.patroni.ha.restart()
status_code = 200 if status else 503
data = msg.encode('utf-8')
except:
except Exception:
logger.exception('Exception during restart')
self.send_response(status_code)
@@ -139,31 +142,35 @@ class RestApiHandler(BaseHTTPRequestHandler):
self.end_headers()
self.wfile.write(data)
def poll_failover_result(self, leader, member):
for a in range(0, 15):
def poll_failover_result(self, leader, candidate):
for _ in range(0, 15):
time.sleep(1)
try:
cluster = self.server.patroni.dcs.get_cluster()
if cluster.leader and cluster.leader.name != leader:
return 200, ('Successfully failed over to ' + cluster.leader.name).encode('utf-8')
if not candidate or candidate == cluster.leader.name:
return 200, ('Successfully failed over to ' + cluster.leader.name).encode('utf-8')
else:
return 200, 'Failed over to "{0}" instead of "{1}"'.format(cluster.leader.name,
candidate).encode('utf-8')
if not cluster.failover:
return 503, b'Failover failed'
except:
pass
except Exception as e:
logger.debug('Exception occured during polling failover result: %s', e)
return 503, b'Failover status unknown'
def is_failover_possible(self, cluster, leader, member):
def is_failover_possible(self, cluster, leader, candidate):
if leader and not cluster.leader or cluster.leader.name != leader:
return b'leader name does not match'
if member:
members = [m for m in cluster.members if m.name == member]
if candidate:
members = [m for m in cluster.members if m.name == candidate]
if not members:
return b'member does not exists'
return b'candidate does not exists'
else:
members = [m for m in cluster.members if m.name != cluster.leader.name and m.api_url]
if not members:
return b'failover is not possible: cluster does not have members except leader'
for member, reachable, in_recovery, xlog_location, tags in self.server.patroni.ha.fetch_nodes_statuses(members):
for _, reachable, _, _, tags in self.server.patroni.ha.fetch_nodes_statuses(members):
if reachable and not tags.get('nofailover', False):
return None
return b'failover is not possible: no good candidates have been found'
@@ -171,18 +178,49 @@ class RestApiHandler(BaseHTTPRequestHandler):
@check_auth
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'))
try:
request = json.loads(self.rfile.read(content_length).decode('utf-8'))
except ValueError:
request = {}
leader = request.get('leader')
member = request.get('member')
candidate = request.get('candidate') or request.get('member')
scheduled_at = request.get('scheduled_at')
cluster = self.server.patroni.ha.dcs.get_cluster()
status_code = 503
data = self.is_failover_possible(cluster, leader, member)
if not data:
if not self.server.patroni.dcs.manual_failover(leader, member):
data = b'failed to write failover key into DCS'
status_code = 500
logger.info("received failover request with leader=%s candidate=%s scheduled_at=%s",
leader, candidate, scheduled_at)
data = b''
if leader or candidate:
if scheduled_at:
try:
scheduled_at = dateutil.parser.parse(scheduled_at)
if scheduled_at.tzinfo is None:
data = b'Timezone information is mandatory for scheduled_at'
status_code = 400
elif scheduled_at < datetime.datetime.now(pytz.utc):
data = b'Cannot schedule failover in the past'
status_code = 422
elif self.server.patroni.dcs.manual_failover(leader, candidate, scheduled_at=scheduled_at):
data = b'Failover scheduled'
status_code = 200
except (ValueError, TypeError):
logger.exception('Invalid scheduled failover time: %s', request['scheduled_at'])
data = b'Unable to parse scheduled timestamp. It should be in an unambiguous format, e.g. ISO 8601'
status_code = 422
else:
self.server.patroni.dcs.event.set()
status_code, data = self.poll_failover_result(cluster.leader and cluster.leader.name, member)
data = self.is_failover_possible(cluster, leader, candidate)
if not data:
if not self.server.patroni.dcs.manual_failover(leader, candidate):
data = b'failed to write failover key into DCS'
status_code = 503
else:
self.server.patroni.dcs.event.set()
status_code, data = self.poll_failover_result(cluster.leader and cluster.leader.name, candidate)
else:
status_code = 400
data = b'No values given for required parameters leader and candidate'
self.send_response(status_code)
self.send_header('Content-Type', 'text/html')
@@ -229,6 +267,7 @@ class RestApiHandler(BaseHTTPRequestHandler):
END,
pg_xlog_location_diff(pg_last_xlog_receive_location(), '0/0')::bigint,
pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')::bigint,
to_char(pg_last_xact_replay_timestamp(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'),
pg_is_in_recovery() AND pg_is_xlog_replay_paused()""", retry=retry)[0]
return {
'state': self.server.patroni.postgresql.state,
@@ -238,7 +277,8 @@ class RestApiHandler(BaseHTTPRequestHandler):
'xlog': ({
'received_location': row[3],
'replayed_location': row[4],
'paused': row[5]} if row[1] else {
'replayed_timestamp': row[5],
'paused': row[6]} if row[1] else {
'location': row[2]
})
}
+44 -30
View File
@@ -14,8 +14,11 @@ import datetime
from prettytable import PrettyTable
from six.moves.urllib_parse import urlparse
import logging
import dateutil
import tzlocal
from .etcd import Etcd
from .zookeeper import ZooKeeper
from .exceptions import PatroniCtlException
from .postgresql import parseurl
@@ -44,12 +47,12 @@ def parse_dcs(dcs):
parsed = urlparse('//' + dcs)
if scheme == '':
default_schemes = {'2181': 'zookeeper', '8500': 'consul'}
default_schemes = {'2181': 'zookeeper', '8181': 'exhibitor', '8500': 'consul'}
scheme = default_schemes.get(str(parsed.port), 'etcd')
port = parsed.port
if port is None:
default_ports = {'consul': 8500, 'zookeeper': 2181}
default_ports = {'consul': 8500, 'zookeeper': 2181, 'exhibitor': 8181}
port = default_ports.get(str(scheme), 4001)
return {'scheme': str(scheme), 'hostname': str(parsed.hostname), 'port': int(port)}
@@ -103,6 +106,12 @@ def get_dcs(config, scope):
if scheme == 'etcd':
return Etcd(name=scope, config={'scope': scope, 'host': '{0}:{1}'.format(hostname, port)})
if scheme == 'zookeeper':
return ZooKeeper(name=scope, config={'scope': scope, 'hosts': [hostname], 'port': port})
if scheme == 'exhibitor':
return ZooKeeper(name=scope, config={'scope': scope, 'exhibitor': {'hosts': [hostname], 'port': port}})
raise PatroniCtlException('Can not find suitable configuration of distributed configuration store')
@@ -238,7 +247,7 @@ def dsn(cluster_name, config_file, dcs, role, member):
if member is None and role is None:
role = 'master'
config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs)
_, 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')
@@ -307,8 +316,7 @@ def query(
cursor = None
for _ in watching(w, watch, clear=False):
output, cursor = query_member(cluster=cluster, cursor=cursor, member=member, role=role, command=command,
connect_parameters=connect_parameters)
output, cursor = query_member(cluster, cursor, member, role, command, connect_parameters)
print_output(None, output, fmt=fmt, delimiter=delimiter)
if cursor is None:
@@ -354,9 +362,6 @@ def query_member(cluster, cursor, member, role, command, connect_parameters=None
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 {0}'.format(type(dcs)))
output_members(cluster, fmt=fmt)
confirm = click.prompt('Please confirm the cluster name to remove', type=str)
@@ -375,7 +380,7 @@ def remove(config_file, cluster_name, fmt, dcs):
if confirm != cluster.leader.name:
raise PatroniCtlException('You did not specify the current master of the cluster')
dcs.client.delete(dcs.client_path(''), recursive=True)
dcs.delete_cluster()
def wait_for_leader(dcs, timeout=30):
@@ -468,10 +473,12 @@ def reinit(cluster_name, member_names, config_file, dcs, force):
@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('--scheduled', help='Timestamp of a scheduled failover in unambiguous format (e.g. ISO 8601)',
default=None)
@click.option('--force', is_flag=True)
@option_config_file
@option_dcs
def failover(config_file, cluster_name, master, candidate, force, dcs):
def failover(config_file, cluster_name, master, candidate, force, dcs, scheduled):
"""
We want to trigger a failover for the specified cluster name.
@@ -509,6 +516,25 @@ def failover(config_file, cluster_name, master, candidate, force, dcs):
if candidate and candidate not in candidate_names:
raise PatroniCtlException('Member {0} does not exist in cluster {1}'.format(candidate, cluster_name))
if scheduled is None and not force:
scheduled = click.prompt('When should the failover take place (e.g. 2015-10-01T14:30) ', type=str,
default='now')
if (scheduled or 'now') == 'now':
scheduled_at = None
else:
try:
scheduled_at = dateutil.parser.parse(scheduled)
if scheduled_at.tzinfo is None:
scheduled_at = tzlocal.get_localzone().localize(scheduled_at)
except (ValueError, TypeError):
message = 'Unable to parse scheduled timestamp ({}). It should be in an unambiguous format (e.g. ISO 8601)'
raise PatroniCtlException(message.format(scheduled))
scheduled_at = scheduled_at.isoformat()
failover_value = {'leader': master, 'candidate': candidate, 'scheduled_at': scheduled_at}
logging.debug(failover_value)
# 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)
@@ -520,17 +546,14 @@ def failover(config_file, cluster_name, master, candidate, force, dcs):
if not a:
raise PatroniCtlException('Aborting failover')
failover_value = '{0}:{1}'.format(master, candidate or '')
t_started = time.time()
r = None
try:
r = post_patroni(cluster.leader.member, 'failover', {'leader': master, 'member': candidate or ''})
r = post_patroni(cluster.leader.member, 'failover', failover_value)
if r.status_code == 200:
logging.debug(r)
logging.debug(r.text)
cluster = dcs.get_cluster()
click.echo(timestamp() + ' Failing over to new leader: {0}'.format(cluster.leader.member.name))
logging.debug(cluster)
click.echo('{0} {1}'.format(timestamp(), r.text))
else:
click.echo('Failover failed, details: {0}, {1}'.format(r.status_code, r.text))
return
@@ -538,17 +561,9 @@ def failover(config_file, cluster_name, master, candidate, force, dcs):
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 {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)
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() + ' Initializing failover from master {0}'.format(master))
dcs.manual_failover(master, candidate, scheduled_at=failover_value)
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)
@@ -572,10 +587,9 @@ def output_members(cluster, name=None, fmt='pretty'):
host = build_connect_parameters(m.conn_url)['host']
xlog_location = m.data.get('xlog_location')
if xlog_location is None or (xlog_location_cluster < xlog_location):
lag = ''
else:
xlog_location = m.data.get('xlog_location') or 0
lag = ''
if (xlog_location_cluster >= xlog_location):
lag = round((xlog_location_cluster - xlog_location)/1024/1024)
rows.append([
+56 -14
View File
@@ -1,8 +1,9 @@
import abc
import dateutil
import json
import six
from collections import namedtuple
from patroni.exceptions import DCSError
from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl
from threading import Event, Lock
@@ -89,12 +90,44 @@ class Leader(namedtuple('Leader', 'index,session,member')):
return self.member.conn_url
class Failover(namedtuple('Failover', 'index,leader,member')):
class Failover(namedtuple('Failover', 'index,leader,candidate,scheduled_at')):
"""
>>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader"}'))
True
>>> 'Failover' in str(Failover.from_node(1, '{"leader": "cluster_leader", "member": "cluster_candidate"}'))
True
>>> Failover.from_node(1, 'null') is None
True
>>> n = '{"leader": "cluster_leader", "member": "cluster_candidate", "scheduled_at": "2016-01-14T10:09:57.1394Z"}'
>>> 'tzinfo=' in str(Failover.from_node(1, n))
True
>>> Failover.from_node(1, None) is None
True
>>> Failover.from_node(1, '{}') is None
True
>>> 'abc' in Failover.from_node(1, 'abc:def')
True
"""
@staticmethod
def from_node(index, value):
t = [a.strip() for a in value.split(':')] + ['']
return Failover(index, t[0], t[1]) if t[0] or t[1] else None
if not value:
return None
try:
data = json.loads(value)
if not data:
return None
except ValueError:
t = [a.strip() for a in value.split(':')]
leader = t[0]
candidate = t[1] if len(t) > 1 else None
return Failover(index, leader, candidate, None) if leader or candidate else None
if data.get('scheduled_at'):
data['scheduled_at'] = dateutil.parser.parse(data['scheduled_at'])
return Failover(index, data.get('leader'), data.get('member'), data.get('scheduled_at'))
class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members,failover')):
@@ -114,11 +147,13 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem
def has_member(self, member_name):
return any(m for m in self.members if m.name == member_name)
def get_member(self, member_name):
return ([m for m in self.members if m.name == member_name] or [None])[0]
@six.add_metaclass(abc.ABCMeta)
class AbstractDCS(object):
__metaclass__ = abc.ABCMeta
_INITIALIZE = 'initialize'
_LEADER = 'leader'
_FAILOVER = 'failover'
@@ -223,15 +258,18 @@ class AbstractDCS(object):
def set_failover_value(self, value, index=None):
"""Create or update `/failover` key"""
def manual_failover(self, leader, member, index=None):
return self.set_failover_value(leader + (':' + member if member else ''), index)
def manual_failover(self, leader, candidate, scheduled_at=None, index=None):
failover_value = {}
if leader:
failover_value['leader'] = leader
def current_leader(self):
try:
cluster = self.get_cluster()
return None if cluster.is_unlocked() else cluster.leader
except DCSError:
return None
if candidate:
failover_value['member'] = candidate
if scheduled_at:
failover_value['scheduled_at'] = scheduled_at.isoformat()
return self.set_failover_value(json.dumps(failover_value), index)
@abc.abstractmethod
def touch_member(self, connection_string, ttl=None):
@@ -270,6 +308,10 @@ class AbstractDCS(object):
def cancel_initialization(self):
""" Removes the initialize key for a cluster """
@abc.abstractmethod
def delete_cluster(self):
"""Delete cluster from DCS"""
def watch(self, timeout):
"""If the current node is a master it should just sleep.
Any other node should watch for changes of leader key with a given timeout
+66 -26
View File
@@ -14,6 +14,7 @@ from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member
from patroni.exceptions import DCSError
from patroni.utils import Retry, RetryFailedError, sleep
from requests.exceptions import RequestException
from six.moves.http_client import HTTPException
logger = logging.getLogger(__name__)
@@ -49,12 +50,57 @@ class Client(etcd.Client):
self._update_machines_cache = True
return [self._base_uri]
def api_execute(self, path, method, **kwargs):
def _do_http_request(self, request_executor, method, url, fields=None, **kwargs):
try:
response = request_executor(method, url, fields=fields, **kwargs)
response.data.decode('utf-8')
self._check_cluster_id(response)
except (urllib3.exceptions.HTTPError, HTTPException, socket.error) as e:
if (isinstance(fields, dict) and fields.get("wait") == "true" and
isinstance(e, urllib3.exceptions.ReadTimeoutError)):
logger.debug("Watch timed out.")
raise etcd.EtcdWatchTimedOut("Watch timed out: {0}".format(e), cause=e)
logger.error("Request to server %s failed: %r", self._base_uri, e)
logger.info("Reconnection allowed, looking for another server.")
self._base_uri = self._next_server(cause=e)
response = False
return response
def api_execute(self, path, method, params=None, timeout=None):
if not path.startswith('/'):
raise ValueError('Path does not start with /')
if timeout is None:
timeout = self.read_timeout
if timeout == 0:
timeout = None
kwargs = {'timeout': timeout, 'fields': params, 'redirect': self.allow_redirect,
'headers': self._get_headers(), 'preload_content': False}
if method in [self._MGET, self._MDELETE]:
request_executor = self.http.request
elif method in [self._MPUT, self._MPOST]:
request_executor = self.http.request_encode_body
kwargs['encode_multipart'] = False
else:
raise etcd.EtcdException('HTTP method {0} not supported'.format(method))
# Update machines_cache if previous attempt of update has failed
if self._update_machines_cache:
self._load_machines_cache()
response = False
try:
return super(Client, self).api_execute(path, method, **kwargs)
while not response:
response = self._do_http_request(request_executor, method, self._base_uri + path, **kwargs)
if response is False and not self._use_proxies:
self._machines_cache = self.machines
self._machines_cache.remove(self._base_uri)
return self._handle_server_response(response)
except etcd.EtcdConnectionFailed:
self._update_machines_cache = True
raise
@@ -67,16 +113,6 @@ class Client(etcd.Client):
logger.exception('Can not resolve SRV for %s', host)
return []
# try to workarond bug in python-etcd: https://github.com/jplana/python-etcd/issues/81
def _result_from_response(self, response):
try:
response.data.decode('utf-8')
except urllib3.exceptions.TimeoutError:
raise
except Exception as 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):
"""Fetch list of etcd-cluster member by resolving _etcd-server._tcp. SRV record.
This record should contain list of host and peer ports which could be used to run
@@ -163,7 +199,7 @@ class Etcd(AbstractDCS):
etcd.EtcdLeaderElectionInProgress,
etcd.EtcdWatcherCleared,
etcd.EtcdEventIndexCleared))
self.client = self.get_etcd_client(config)
self._client = self.get_etcd_client(config)
def retry(self, *args, **kwargs):
return self._retry.copy()(*args, **kwargs)
@@ -185,7 +221,7 @@ class Etcd(AbstractDCS):
def _load_cluster(self):
try:
result = self.retry(self.client.read, self.client_path(''), recursive=True)
result = self.retry(self._client.read, self.client_path(''), recursive=True)
nodes = {os.path.relpath(node.key, result.key): node for node in result.leaves}
# get initialize flag
@@ -220,15 +256,15 @@ class Etcd(AbstractDCS):
@catch_etcd_errors
def touch_member(self, connection_string, ttl=None):
return self.retry(self.client.set, self.member_path, connection_string, ttl or self.ttl)
return self.retry(self._client.set, self.member_path, connection_string, ttl or self.ttl)
@catch_etcd_errors
def take_leader(self):
return self.retry(self.client.set, self.leader_path, self._name, self.ttl)
return self.retry(self._client.set, self.leader_path, self._name, self.ttl)
def attempt_to_acquire_leader(self):
try:
return bool(self.retry(self.client.write, self.leader_path, self._name, ttl=self.ttl, prevExist=False))
return bool(self.retry(self._client.write, self.leader_path, self._name, ttl=self.ttl, prevExist=False))
except etcd.EtcdAlreadyExist:
logger.info('Could not take out TTL lock')
except (RetryFailedError, etcd.EtcdException):
@@ -237,27 +273,31 @@ class Etcd(AbstractDCS):
@catch_etcd_errors
def set_failover_value(self, value, index=None):
return self.client.write(self.failover_path, value, prevIndex=index or 0)
return self._client.write(self.failover_path, value, prevIndex=index or 0)
@catch_etcd_errors
def write_leader_optime(self, last_operation):
return self.client.set(self.leader_optime_path, last_operation)
return self._client.set(self.leader_optime_path, last_operation)
@catch_etcd_errors
def update_leader(self):
return self.retry(self.client.test_and_set, self.leader_path, self._name, self._name, self.ttl)
return self.retry(self._client.test_and_set, self.leader_path, self._name, self._name, self.ttl)
@catch_etcd_errors
def initialize(self, create_new=True, sysid=""):
return self.retry(self.client.write, self.initialize_path, sysid, prevExist=(not create_new))
return self.retry(self._client.write, self.initialize_path, sysid, prevExist=(not create_new))
@catch_etcd_errors
def delete_leader(self):
return self.client.delete(self.leader_path, prevValue=self._name)
return self._client.delete(self.leader_path, prevValue=self._name)
@catch_etcd_errors
def cancel_initialization(self):
return self.retry(self.client.delete, self.initialize_path)
return self.retry(self._client.delete, self.initialize_path)
@catch_etcd_errors
def delete_cluster(self):
return self.retry(self._client.delete, self.client_path(''), recursive=True)
def watch(self, timeout):
cluster = self.cluster
@@ -268,12 +308,12 @@ class Etcd(AbstractDCS):
while index and timeout >= 1: # when timeout is too small urllib3 doesn't have enough time to connect
try:
self.client.watch(self.leader_path, index=index + 1, timeout=timeout + 0.5)
self._client.watch(self.leader_path, index=index + 1, timeout=timeout + 0.5)
# Synchronous work of all cluster members with etcd is less expensive
# than reestablishing http connection every time from every replica.
return True
except urllib3.exceptions.TimeoutError:
self.client.http.clear()
except etcd.EtcdWatchTimedOut:
self._client.http.clear()
return False
except etcd.EtcdException:
logging.exception('watch')
+71 -36
View File
@@ -3,10 +3,13 @@ import logging
import psycopg2
import requests
import sys
import datetime
import pytz
from multiprocessing.pool import ThreadPool
from patroni.async_executor import AsyncExecutor
from patroni.exceptions import DCSError, PostgresConnectionException
from multiprocessing.pool import ThreadPool
from patroni.utils import sleep
logger = logging.getLogger(__name__)
@@ -62,19 +65,25 @@ class Ha(object):
pass
self.dcs.touch_member(json.dumps(data, separators=(',', ':')))
def clone(self, leader):
if self.state_handler.bootstrap(cluster_initialized=True, current_leader=leader):
logger.info('bootstrapped from leader' if leader else 'bootstrapped without leader')
def clone(self, clone_member, clone_member_name="leader"):
if self.state_handler.bootstrap(cluster_initialized=True, clone_member=clone_member):
logger.info('bootstrapped from {0}'.format(clone_member_name)
if clone_member else 'bootstrapped without leader')
else:
self.state_handler.stop('immediate')
self.state_handler.remove_data_directory()
logger.error('failed to bootstrap from leader' if leader else 'failed to bootstrap (without leader)')
logger.error('failed to bootstrap from {0}'.format(clone_member_name)
if clone_member else 'failed to bootstrap (without leader)')
def bootstrap(self):
if not self.cluster.is_unlocked(): # cluster already has leader
self._async_executor.schedule('bootstrap from leader')
self._async_executor.run_async(self.clone, args=(self.cluster.leader, ))
return 'trying to bootstrap from leader'
clonefrom = self.patroni.clonefrom
clone_member = self.cluster.get_member(clonefrom)\
if self.cluster.has_member(clonefrom) else self.cluster.leader
clone_member_name = 'leader' if clone_member == self.cluster.leader else 'replica \'{0}\''.format(clonefrom)
self._async_executor.schedule('bootstrap from {0}'.format(clone_member_name))
self._async_executor.run_async(self.clone, args=(clone_member, clone_member_name))
return 'trying to bootstrap from {0}'.format(clone_member_name)
elif not self.cluster.initialize and not self.patroni.nofailover: # no initialize key
if self.dcs.initialize(create_new=True): # race for initialization
try:
@@ -88,11 +97,12 @@ class Ha(object):
self.state_handler.move_data_directory()
raise
self.dcs.take_leader()
self.load_cluster_from_dcs()
return 'initialized a new cluster'
else:
return 'failed to acquire initialize lock'
else:
if self.state_handler.can_create_replica_without_leader():
if self.state_handler.can_create_replica_without_replication_connection():
self._async_executor.run_async(self.clone, args=(None, ))
return "trying to bootstrap without leader"
return 'waiting for leader to bootstrap'
@@ -100,32 +110,30 @@ class Ha(object):
def recover(self):
# try to see if we are the former master that crashed. If so - we likely need to run pg_rewind
# in order to join the former standby being promoted.
pg_controldata = self.state_handler.controldata()
if (self.state_handler.role == 'master') and pg_controldata and\
pg_controldata.get('Database cluster state', '') == 'in production': # crashed master
self.state_handler.require_rewind()
if self.state_handler.role == 'master':
pg_controldata = self.state_handler.controldata()
if pg_controldata and pg_controldata.get('Database cluster state', '') == 'in production': # crashed master
self.state_handler.require_rewind()
self.recovering = True
return self.follow("started as readonly because i had the session lock",
"started as a secondary",
refresh=True, recovery=True)
return self.follow("starting as readonly because i had the session lock", "starting as a secondary", True, True)
def follow(self, demote_reason, follow_reason, refresh=True, recovery=False):
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
ret = demote_reason if not recovery and self.state_handler.is_leader() else 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:
node_to_follow = [m for m in self.cluster.members if m.name == self.patroni.replicatefrom]
node_to_follow = node_to_follow[0] if node_to_follow else self.cluster.leader
else:
node_to_follow = self.cluster.leader
node_to_follow = None if node_to_follow and node_to_follow.name == self.state_handler.name else node_to_follow
if node_to_follow and node_to_follow.name == self.state_handler.name:
ret = demote_reason
node_to_follow = None
if not self.state_handler.check_recovery_conf(node_to_follow) or recovery:
self._async_executor.schedule('changing primary_conninfo and restarting')
self._async_executor.run_async(self.state_handler.follow, (node_to_follow, recovery))
@@ -199,7 +207,7 @@ class Ha(object):
ret = False
members = [m for m in members if m.name != self.state_handler.name and not m.nofailover and m.api_url]
if members:
for member, reachable, in_recovery, xlog_location, tags in self.fetch_nodes_statuses(members):
for member, reachable, _, _, tags in self.fetch_nodes_statuses(members):
if reachable and not tags.get('nofailover', False):
ret = True # TODO: check xlog_location
elif not reachable:
@@ -212,14 +220,14 @@ class Ha(object):
def manual_failover_process_no_leader(self):
failover = self.cluster.failover
if failover.member: # manual failover to specific member
if failover.member == self.state_handler.name: # manual failover to me
if failover.candidate: # manual failover to specific member
if failover.candidate == self.state_handler.name: # manual failover to me
return True
# find specific node and check that it is healthy
members = [m for m in self.cluster.members if m.name == failover.member]
members = [m for m in self.cluster.members if m.name == failover.candidate]
if members:
member, reachable, in_recovery, xlog_location, tags = self.fetch_node_status(members[0])
member, reachable, _, _, tags = self.fetch_node_status(members[0])
if reachable and not tags.get('nofailover', False): # node is healthy
logger.info('manual failover: to %s, i am %s', member.name, self.state_handler.name)
return False
@@ -230,13 +238,13 @@ class Ha(object):
logger.warning('manual failover: member %s is not allowed to promote', member.name)
# at this point we should consider all members as a candidates for failover
# i.e. we assume that failover.member is None
# i.e. we assume that failover.candidate is None
# try to pick some other members to failover and check that they are healthy
if failover.leader:
if self.state_handler.name == failover.leader: # I was the leader
# exclude me and desired member which is unhealthy (failover.member can be None)
members = [m for m in self.cluster.members if m.name not in (failover.member, failover.leader)]
# exclude me and desired member which is unhealthy (failover.candidate can be None)
members = [m for m in self.cluster.members if m.name not in (failover.candidate, failover.leader)]
if self.is_failover_possible(members): # check that there are healthy members
return False
else: # I was the leader and it looks like currently I am the only healthy member
@@ -269,13 +277,40 @@ class Ha(object):
self.dcs.delete_leader()
self.touch_member()
self.dcs.reset_cluster()
self.state_handler.follow_the_leader(None)
sleep(2) # Give a time to somebody to promote
self.recover()
else:
self.state_handler.follow(None)
def process_manual_failover_from_leader(self):
failover = self.cluster.failover
if failover.scheduled_at:
# If the failover is in the far future, we shouldn't do anything and just return.
# If the failover is in the past, we consider the value to be stale and we remove
# the value.
# If the value is close to now, we initiate the failover
now = datetime.datetime.now(pytz.utc)
try:
delta = (failover.scheduled_at - now).total_seconds()
if delta > self.patroni.nap_time:
logging.info('Awaiting failover at %s (in %.0f seconds)', failover.scheduled_at.isoformat(), delta)
return
elif delta < - int(self.patroni.nap_time * 1.5):
logger.warning('Found a stale failover value, cleaning up: %s', failover.scheduled_at)
self.dcs.manual_failover('', '', index=self.cluster.failover.index)
return
# The value is very close to now
sleep(max(delta, 0))
logger.info('Manual scheduled failover at {}'.format(failover.scheduled_at.isoformat()))
except TypeError:
logger.warning('Incorrect value in of scheduled_at: %s', failover.scheduled_at)
if not failover.leader or failover.leader == self.state_handler.name:
if not failover.member or failover.member != self.state_handler.name:
members = [m for m in self.cluster.members if not failover.member or m.name == failover.member]
if not failover.candidate or failover.candidate != self.state_handler.name:
members = [m for m in self.cluster.members if not failover.candidate or m.name == failover.candidate]
if self.is_failover_possible(members): # check that there are healthy members
self._async_executor.schedule('manual failover: demote')
self._async_executor.run_async(self.demote)
@@ -289,7 +324,7 @@ class Ha(object):
self.cluster.failover.leader, self.state_handler.name)
logger.info('Trying to clean up failover key')
self.dcs.manual_failover('', '', self.cluster.failover.index)
self.dcs.manual_failover('', '', index=self.cluster.failover.index)
def process_unhealthy_cluster(self):
if self.is_healthiest_node():
@@ -297,7 +332,7 @@ class Ha(object):
if self.cluster.failover:
logger.info('Cleaning up failover key after acquiring leader lock...')
self.dcs.manual_failover('', '')
self.dcs.get_cluster()
self.load_cluster_from_dcs()
return self.enforce_master_role('acquired session lock as a leader',
'promoted self to leader by acquiring session lock')
else:
@@ -429,8 +464,8 @@ class Ha(object):
else:
# check if we are allowed to join
if self.sysid_valid(self.cluster.initialize) and self.cluster.initialize != self.state_handler.sysid:
logger.fatal("system ID mismatch, node {0} belongs to a different cluster".
format(self.state_handler.name))
logger.fatal("system ID mismatch, node %s belongs to a different cluster: %s != %s",
self.state_handler.name, self.cluster.initialize, self.state_handler.sysid)
sys.exit(1)
# try to start dead postgres
+71 -69
View File
@@ -207,7 +207,7 @@ class Postgresql(object):
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.write(fd, self.superuser['password'].encode('utf-8'))
os.close(fd)
options.append('--pwfile={0}'.format(pwfile))
@@ -233,9 +233,10 @@ class Postgresql(object):
env['PGPASSFILE'] = self.pgpass
return env
def sync_replica(self, leader):
env = self.write_pgpass(parseurl(leader.conn_url)) if leader else os.environ.copy()
if self.create_replica(leader, env) == 0:
def sync_replica(self, clone_member):
# add the credentials to connect to the replica origin to pgpass.
env = self.write_pgpass(parseurl(clone_member.conn_url)) if clone_member else os.environ.copy()
if self.create_replica(clone_member, env) == 0:
self.delete_trigger_file()
return True
return False
@@ -248,33 +249,35 @@ class Postgresql(object):
"""
return ' '.join('{0}={1}'.format(param, val) for param, val in sorted(conn.items()))
def replica_method_can_work_without_leader(self, method):
def replica_method_can_work_without_replication_connection(self, method):
return method != 'basebackup' and self.config and self.config.get(method, {}).get('no_master')
def can_create_replica_without_leader(self):
def can_create_replica_without_replication_connection(self):
""" go through the replication methods to see if there are ones
that does not require a running leader to create the replica.
that does not require a working replication connection.
"""
replica_methods = self.config.get('create_replica_method', [])
return any(self.replica_method_can_work_without_leader(replica_method) for replica_method in replica_methods)
return any(self.replica_method_can_work_without_replication_connection(replica_method)
for replica_method in replica_methods)
def create_replica(self, leader, env):
def create_replica(self, clone_member, env):
# create the replica according to the replica_method
# defined by the user. this is a list, so we need to
# loop through all methods the user supplies
connstring = leader.conn_url if leader else ""
connstring = clone_member.conn_url if clone_member else ""
# get list of replica methods from config.
# If there is no configuration key, or no value is specified, use basebackup
replica_methods = self.config.get('create_replica_method') or ['basebackup']
# if we don't have any leader, leave only replica methods that work without it
replica_methods = [r for r in replica_methods if self.replica_method_can_work_without_leader(r)] if not leader \
else replica_methods
# if we don't have any source, leave only replica methods that work without it
replica_methods = \
[r for r in replica_methods if self.replica_method_can_work_without_replication_connection(r)]\
if not clone_member else replica_methods
# go through them in priority order
ret = 1
for replica_method in replica_methods:
# if the method is basebackup, then use the built-in
if replica_method == "basebackup":
ret = self.basebackup(leader, env)
ret = self.basebackup(clone_member, env)
if ret == 0:
logger.info("replica has been created using basebackup")
# if basebackup succeeds, exit with success
@@ -385,7 +388,7 @@ class Postgresql(object):
except psycopg2.Error:
logging.exception('Exception during CHECKPOINT')
def stop(self, mode='fast', block_callbacks=False):
def stop(self, mode='fast', block_callbacks=False, checkpoint=True):
# make sure we close all connections established against
# the former node, otherwise, we might get a stalled one
# after kill -9, which would report incorrect data to
@@ -397,9 +400,10 @@ class Postgresql(object):
self.set_state('stopped')
return True
if block_callbacks:
if checkpoint:
self.checkpoint()
else:
if not block_callbacks:
self.set_state('stopping')
ret = subprocess.call(self._pg_ctl + ['stop', '-m', mode]) == 0
@@ -506,7 +510,7 @@ recovery_target_timeline = 'latest'
try:
data = subprocess.check_output(['pg_controldata', self.data_dir])
if data:
data = data.decode().splitlines()
data = data.decode('utf-8').splitlines()
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")
@@ -526,8 +530,7 @@ recovery_target_timeline = 'latest'
result[name] = val
except IOError:
logger.exception('Error when reading postmaster.opts')
finally:
return result
return result
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 """
@@ -560,47 +563,45 @@ recovery_target_timeline = 'latest'
logger.exception("Unable to list %s", status_dir)
def follow(self, leader, recovery=False):
if not self.check_recovery_conf(leader) or recovery:
change_role = (self.role == 'master')
self._need_rewind = (self._need_rewind or change_role) and self.can_rewind
if self._need_rewind:
logger.info("set the rewind flag after demote")
self.write_recovery_conf(leader)
if not leader or not self._need_rewind: # do not rewind until the leader becomes available
ret = self.restart()
else: # we have a leader and need to rewind
if self.is_running():
self.stop()
# at present, pg_rewind only runs when the cluster is shut down cleanly
# and not shutdown in recovery. We have to remove the recovery.conf if present
# and start/shutdown in a single user mode to emulate this.
# XXX: if recovery.conf is linked, it will be written anew as a normal file.
if os.path.islink(self.recovery_conf):
os.unlink(self.recovery_conf)
else:
os.remove(self.recovery_conf)
# Archived segments might be useful to pg_rewind,
# clean the flags that tell we should remove them.
self.cleanup_archive_status()
# Start in a single user mode and stop to produce a clean shutdown
opts = self.read_postmaster_opts()
opts['archive_mode'] = 'on'
opts['archive_command'] = 'false'
self.single_user_mode(options=opts)
if self.rewind(leader):
ret = self.start()
else:
logger.error("unable to rewind the former master")
self.remove_data_directory()
ret = True
self._need_rewind = False
if change_role and ret:
self.call_nowait(ACTION_ON_ROLE_CHANGE)
return ret
else:
if self.check_recovery_conf(leader) and not recovery:
return True
change_role = self.role == 'master'
self._need_rewind = (self._need_rewind or change_role) and self.can_rewind
if self._need_rewind:
logger.info("set the rewind flag after demote")
self.write_recovery_conf(leader)
if leader and self._need_rewind: # we have a leader and need to rewind
if self.is_running():
self.stop()
# at present, pg_rewind only runs when the cluster is shut down cleanly
# and not shutdown in recovery. We have to remove the recovery.conf if present
# and start/shutdown in a single user mode to emulate this.
# XXX: if recovery.conf is linked, it will be written anew as a normal file.
if os.path.islink(self.recovery_conf):
os.unlink(self.recovery_conf)
else:
os.remove(self.recovery_conf)
# Archived segments might be useful to pg_rewind,
# clean the flags that tell we should remove them.
self.cleanup_archive_status()
# Start in a single user mode and stop to produce a clean shutdown
opts = self.read_postmaster_opts()
opts.update({'archive_mode': 'on', 'archive_command': 'false'})
self.single_user_mode(options=opts)
if self.rewind(leader):
ret = self.start()
else:
logger.error("unable to rewind the former master")
self.remove_data_directory()
ret = True
self._need_rewind = False
else: # do not rewind until the leader becomes available
ret = self.restart()
if change_role and ret:
self.call_nowait(ACTION_ON_ROLE_CHANGE)
return ret
def save_configuration_files(self):
"""
copy postgresql.conf to postgresql.conf.backup to be able to retrive configuration files
@@ -701,18 +702,19 @@ $$""".format(name, options), name, password, password)
def last_operation(self):
return str(self.xlog_position())
def bootstrap(self, cluster_initialized=False, current_leader=None):
def bootstrap(self, cluster_initialized=False, clone_member=None):
"""
Populate PostgreSQL data directory by doing one of the following:
- create with initdb if there is no master.
- initialize the replica from an existing master
- initialize the replica from an existing member (master or replica)
- initialize the replica using the replica creation method that
works without the master (i.e. restore from on-disk base backup)
works without the replication connection (i.e. restore from on-disk
base backup)
The choice between the last 2 is triggered by the initialize flag.
We should never try to initdb an already initialized cluster, nor
try to bootstrap the cluster that lacks the initialize key from from
the master-less replica creation method (in the latter case, there is
try to bootstrap the cluster that lacks the initialize key using the
master-less replica creation method (in the latter case, there is
no clear inidicator of the moment we should abandon our attempts and
swich to initdb).
@@ -722,7 +724,7 @@ $$""".format(name, options), name, password, password)
that should be retried in the future.
"""
ret = False
if not (cluster_initialized or current_leader):
if not (cluster_initialized or clone_member):
ret = self.initialize() and self.start()
if ret:
self.create_replication_user()
@@ -730,9 +732,9 @@ $$""".format(name, options), name, password, password)
else:
raise PostgresException("Could not bootstrap master PostgreSQL")
else:
if self.sync_replica(current_leader):
if self.sync_replica(clone_member):
self.restore_configuration_files()
self.write_recovery_conf(current_leader, True)
self.write_recovery_conf(clone_member, True)
ret = self.start()
return ret
@@ -760,12 +762,12 @@ $$""".format(name, options), name, password, password)
logger.exception('Could not remove data directory %s', self.data_dir)
self.move_data_directory()
def basebackup(self, leader, env):
def basebackup(self, clone_member, env):
# creates a replica data dir using pg_basebackup.
# this is the default, built-in create_replica_method
# tries twice, then returns failure (as 1)
# uses "stream" as the xlog-method to avoid sync issues
master_connection = leader.conn_url
master_connection = clone_member.conn_url
maxfailures = 2
ret = 1
for bbfailures in range(0, maxfailures):
+1 -1
View File
@@ -154,7 +154,7 @@ def main():
args = parser.parse_args()
# retry cloning in a loop
for retry in range(0, args.retries + 1):
for _ in range(0, args.retries + 1):
restore = WALERestore(scope=args.scope, datadir=args.datadir, connstring=args.connstring,
env_dir=args.envdir, threshold_mb=args.threshold_megabytes,
threshold_pct=args.threshold_backup_size_percentage, use_iam=args.use_iam,
+9 -24
View File
@@ -1,10 +1,11 @@
import datetime
import os
import random
import re
import signal
import sys
import time
import pytz
import dateutil.parser
from patroni.exceptions import PatroniException
@@ -12,39 +13,23 @@ __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
T
(?P<hour>\d{2}):(?P<minute>\d{2}):(?P<second>\d{2})\.(?P<microsecond>\d{6}) # time
\d*Z$''', re.X)
def parse_datetime(time_str):
"""
>>> parse_datetime('2015-06-10T12:56:30.552539016Z')
datetime.datetime(2015, 6, 10, 12, 56, 30, 552539)
>>> parse_datetime('2015-06-10 12:56:30.552539016Z')
"""
m = _DATE_TIME_RE.match(time_str)
if not m:
return None
p = dict((n, int(m.group(n))) for n in 'year month day hour minute second microsecond'.split(' '))
return datetime.datetime(**p)
def calculate_ttl(expiration):
"""
>>> calculate_ttl(None)
>>> calculate_ttl('2015-06-10 12:56:30.552539016Z')
>>> calculate_ttl('2015-06-10 12:56:30.552539016Z') < 0
True
>>> calculate_ttl('2015-06-10T12:56:30.552539016Z') < 0
True
>>> calculate_ttl('fail-06-10T12:56:30.552539016Z')
"""
if not expiration:
return None
expiration = parse_datetime(expiration)
if not expiration:
try:
expiration = dateutil.parser.parse(expiration)
except (ValueError, TypeError):
return None
now = datetime.datetime.utcnow()
now = datetime.datetime.now(pytz.utc)
return int((expiration - now).total_seconds())
+1 -1
View File
@@ -1 +1 @@
__version__ = '0.76'
__version__ = '0.80'
+49 -43
View File
@@ -83,32 +83,32 @@ class ZooKeeper(AbstractDCS):
self.exhibitor = ExhibitorEnsembleProvider(exhibitor['hosts'], exhibitor['port'], poll_interval=interval)
hosts = self.exhibitor.zookeeper_hosts
self.client = KazooClient(hosts=hosts,
timeout=(config.get('session_timeout') or 30),
command_retry={
'deadline': (config.get('reconnect_timeout') or 10),
'max_delay': 1,
'max_tries': -1},
connection_retry={'max_delay': 1, 'max_tries': -1})
self.client.add_listener(self.session_listener)
self._client = KazooClient(hosts=hosts,
timeout=(config.get('session_timeout') or 30),
command_retry={
'deadline': (config.get('reconnect_timeout') or 10),
'max_delay': 1,
'max_tries': -1},
connection_retry={'max_delay': 1, 'max_tries': -1})
self._client.add_listener(self.session_listener)
self._my_member_data = None
self.fetch_cluster = True
self.last_leader_operation = 0
self._fetch_cluster = True
self._last_leader_operation = 0
self.client.start(None)
self._client.start(None)
def session_listener(self, state):
if state in [KazooState.SUSPENDED, KazooState.LOST]:
self.cluster_watcher(None)
def cluster_watcher(self, event):
self.fetch_cluster = True
self._fetch_cluster = True
self.event.set()
def get_node(self, key, watch=None):
try:
ret = self.client.get(key, watch)
ret = self._client.get(key, watch)
return (ret[0].decode('utf-8'), ret[1])
except NoNodeError:
return None
@@ -119,7 +119,7 @@ class ZooKeeper(AbstractDCS):
def get_children(self, key, watch=None):
try:
return self.client.get_children(key, watch)
return self._client.get_children(key, watch)
except NoNodeError:
return []
@@ -132,11 +132,11 @@ class ZooKeeper(AbstractDCS):
return members
def _inner_load_cluster(self):
self.fetch_cluster = False
self._fetch_cluster = False
self.event.clear()
nodes = set(self.get_children(self.client_path(''), self.cluster_watcher))
if not nodes:
self.fetch_cluster = True
self._fetch_cluster = True
# get initialize flag
initialize = (self.get_node(self.initialize_path) or [None])[0] if self._INITIALIZE in nodes else None
@@ -147,17 +147,17 @@ class ZooKeeper(AbstractDCS):
# get leader
leader = self.get_node(self.leader_path) if self._LEADER in nodes else None
if leader:
client_id = self.client.client_id
client_id = self._client.client_id
if leader[0] == self._name and client_id is not None and client_id[0] != leader[1].ephemeralOwner:
logger.info('I am leader but not owner of the session. Removing leader node')
self.client.delete(self.leader_path)
self._client.delete(self.leader_path)
leader = None
if leader:
member = Member(-1, leader[0], None, {})
member = ([m for m in members if m.name == leader[0]] or [member])[0]
leader = Leader(leader[1].version, leader[1].ephemeralOwner, member)
self.fetch_cluster = member.index == -1
self._fetch_cluster = member.index == -1
# failover key
failover = self.get_node(self.failover_path, watch=self.cluster_watcher) if self._FAILOVER in nodes else None
@@ -165,17 +165,17 @@ class ZooKeeper(AbstractDCS):
failover = Failover.from_node(failover[1].version, failover[0])
# get last leader operation
optime = self.get_node(self.leader_optime_path) if self._OPTIME in nodes and self.fetch_cluster else None
self.last_leader_operation = 0 if optime is None else int(optime[0])
self._cluster = Cluster(initialize, leader, self.last_leader_operation, members, failover)
optime = self.get_node(self.leader_optime_path) if self._OPTIME in nodes and self._fetch_cluster else None
self._last_leader_operation = 0 if optime is None else int(optime[0])
self._cluster = Cluster(initialize, leader, self._last_leader_operation, members, failover)
def _load_cluster(self):
if self.exhibitor and self.exhibitor.poll():
self.client.set_hosts(self.exhibitor.zookeeper_hosts)
self._client.set_hosts(self.exhibitor.zookeeper_hosts)
if self.fetch_cluster:
if self._fetch_cluster or self._cluster is None:
try:
self.client.retry(self._inner_load_cluster)
self._client.retry(self._inner_load_cluster)
except:
logger.exception('get_cluster')
self.session_listener(KazooState.LOST)
@@ -183,20 +183,20 @@ class ZooKeeper(AbstractDCS):
def _create(self, path, value, **kwargs):
try:
self.client.retry(self.client.create, path, value.encode('utf-8'), **kwargs)
self._client.retry(self._client.create, path, value.encode('utf-8'), **kwargs)
return True
except:
return False
def attempt_to_acquire_leader(self):
ret = self._create(self.leader_path, self._name, makepath=True, ephemeral=True)
if ret:
if not ret:
logger.info('Could not take out TTL lock')
return ret
def set_failover_value(self, value, index=None):
try:
self.client.retry(self.client.set, self.failover_path, value.encode('utf-8'), version=index or -1)
self._client.retry(self._client.set, self.failover_path, value.encode('utf-8'), version=index or -1)
return True
except NoNodeError:
return value == '' or (not index and self._create(self.failover_path, value))
@@ -206,7 +206,7 @@ class ZooKeeper(AbstractDCS):
def initialize(self, create_new=True, sysid=""):
return self._create(self.initialize_path, sysid, makepath=True) if create_new \
else self.client.retry(self.client.set, self.initialize_path, sysid.encode("utf-8"))
else self._client.retry(self._client.set, self.initialize_path, sysid.encode("utf-8"))
def touch_member(self, data, ttl=None):
cluster = self.cluster
@@ -214,9 +214,9 @@ class ZooKeeper(AbstractDCS):
path = self.member_path
data = data.encode('utf-8')
create = not me
if me and self.client.client_id is not None and me.session != self.client.client_id[0]:
if me and self._client.client_id is not None and me.session != self._client.client_id[0]:
try:
self.client.retry(self.client.delete, path)
self._client.retry(self._client.delete, path)
except NoNodeError:
pass
except:
@@ -228,14 +228,14 @@ class ZooKeeper(AbstractDCS):
try:
if create:
self.client.retry(self.client.create, path, data, makepath=True, ephemeral=True)
self._client.retry(self._client.create, path, data, makepath=True, ephemeral=True)
else:
self.client.retry(self.client.set, path, data)
self._client.retry(self._client.set, path, data)
self._my_member_data = data
return True
except NodeExistsError:
try:
self.client.retry(self.client.set, path, data)
self._client.retry(self._client.set, path, data)
self._my_member_data = data
return True
except:
@@ -249,14 +249,14 @@ class ZooKeeper(AbstractDCS):
def write_leader_optime(self, last_operation):
last_operation = last_operation.encode('utf-8')
if last_operation != self.last_leader_operation:
self.last_leader_operation = last_operation
if last_operation != self._last_leader_operation:
self._last_leader_operation = last_operation
path = self.leader_optime_path
try:
self.client.retry(self.client.set, path, last_operation)
self._client.retry(self._client.set, path, last_operation)
except NoNodeError:
try:
self.client.retry(self.client.create, path, last_operation, makepath=True)
self._client.retry(self._client.create, path, last_operation, makepath=True)
except:
logger.exception('Failed to create %s', path)
except:
@@ -266,22 +266,28 @@ class ZooKeeper(AbstractDCS):
return True
def delete_leader(self):
self.client.restart()
self._client.restart()
self._my_member_data = None
return True
def _cancel_initialization(self):
node = self.get_node(self.initialize_path)
if node:
self.client.delete(self.initialize_path, version=node[1].version)
self._client.delete(self.initialize_path, version=node[1].version)
def cancel_initialization(self):
try:
self.client.retry(self._cancel_initialization)
self._client.retry(self._cancel_initialization)
except:
logger.exception("Unable to delete initialize key")
def delete_cluster(self):
try:
return self._client.retry(self._client.delete, self.client_path(''), recursive=True)
except NoNodeError:
return True
def watch(self, timeout):
if super(ZooKeeper, self).watch(timeout):
self.fetch_cluster = True
return self.fetch_cluster
self._fetch_cluster = True
return self._fetch_cluster
-11
View File
@@ -1,11 +0,0 @@
boto
mock
dnspython3
psycopg2>=2.6.1
PyYAML
requests
six
kazoo>=2.2.1
python-etcd==0.4.2
click>=4.1
prettytable>=0.7
+3 -3
View File
@@ -1,11 +1,11 @@
boto
dnspython
mock
psycopg2>=2.6.1
PyYAML
requests
six >= 1.7
kazoo>=2.2.1
python-etcd==0.4.2
python-etcd==0.4.3
click>=4.1
prettytable>=0.7
tzlocal
python-dateutil
+5 -6
View File
@@ -51,8 +51,8 @@ CLASSIFIERS = [
'Operating System :: POSIX :: Linux',
'Programming Language :: Python',
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3.3',
'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3.5',
'Programming Language :: Python :: Implementation :: CPython',
]
@@ -76,7 +76,7 @@ class PyTest(TestCommand):
def finalize_options(self):
TestCommand.finalize_options(self)
if self.cov_xml or self.cov_html:
self.cov = ['--cov', MAIN_PACKAGE, '--cov', MAIN_PACKAGE, '--cov-report', 'term-missing']
self.cov = ['--cov', MAIN_PACKAGE, '--cov-report', 'term-missing']
if self.cov_xml:
self.cov.extend(['--cov-report', 'xml'])
if self.cov_html:
@@ -116,8 +116,7 @@ def setup_package():
# Some helper variables
version = os.getenv('GO_PIPELINE_LABEL', VERSION)
requirements = 'requirements-py2.txt' if sys.version_info[0] == 2 else 'requirements-py3.txt'
install_reqs = get_install_requirements(requirements)
install_reqs = get_install_requirements('requirements.txt')
command_options = {'test': {'test_suite': ('setup.py', 'tests')}}
if JUNIT_XML:
@@ -142,9 +141,9 @@ def setup_package():
packages=setuptools.find_packages(exclude=['tests', 'tests.*']),
package_data={MAIN_PACKAGE: ["*.json"]},
install_requires=install_reqs,
setup_requires=['six', 'flake8'],
setup_requires=['flake8'],
cmdclass=cmdclass,
tests_require=['pytest-cov', 'pytest'],
tests_require=['mock', 'pytest-cov', 'pytest'],
command_options=command_options,
entry_points={'console_scripts': CONSOLE_SCRIPTS},
)
+52 -25
View File
@@ -11,7 +11,7 @@ import socket
from test_postgresql import psycopg2_connect, MockCursor
class MockPostgresql(Mock):
class MockPostgresql(object):
name = 'test'
state = 'running'
@@ -23,20 +23,12 @@ class MockPostgresql(Mock):
def connection():
return psycopg2_connect()
@staticmethod
def is_running():
return True
class MockHa(Mock):
class MockHa(object):
dcs = Mock()
state_handler = MockPostgresql()
@staticmethod
def schedule_restart():
return 'restart'
@staticmethod
def schedule_reinitialize():
return 'reinitialize'
@@ -54,7 +46,7 @@ class MockHa(Mock):
return [[None, True, None, None, {}]]
class MockPatroni(Mock):
class MockPatroni(object):
postgresql = MockPostgresql()
ha = MockHa()
@@ -101,10 +93,10 @@ class TestRestApiHandler(unittest.TestCase):
MockRestApiServer(RestApiHandler, b'GET /master')
with patch.object(MockHa, 'restart_scheduled', Mock(return_value=True)):
MockRestApiServer(RestApiHandler, b'GET /master')
MockRestApiServer(RestApiHandler, b'GET /master')
self.assertIsNotNone(MockRestApiServer(RestApiHandler, b'GET /master'))
def test_do_OPTIONS(self):
MockRestApiServer(RestApiHandler, b'OPTIONS / HTTP/1.0')
self.assertIsNotNone(MockRestApiServer(RestApiHandler, b'OPTIONS / HTTP/1.0'))
with patch.object(BaseHTTPRequestHandler, 'handle_one_request') as mock_handle_request:
mock_handle_request.side_effect = socket.error("foo")
@@ -113,20 +105,20 @@ class TestRestApiHandler(unittest.TestCase):
# make sure socket.error gets propagated via wfile object in finalize()
with patch.object(MockRequest, 'makefile') as makefile:
makefile.return_value.closed = False
makefile.return_value.readline.side_effect = lambda x: b"foo"
makefile.return_value.flush = Mock(side_effect=socket.error("foo"))
makefile.return_value.readline = Mock(return_value=b'foo')
makefile.return_value.flush = Mock(side_effect=socket.error('foo'))
MockRestApiServer(RestApiHandler, b'OPTIONS / HTTP/1.0')
def test_do_GET_patroni(self):
MockRestApiServer(RestApiHandler, b'GET /patroni')
self.assertIsNotNone(MockRestApiServer(RestApiHandler, b'GET /patroni'))
def test_basicauth(self):
MockRestApiServer(RestApiHandler, b'POST /restart HTTP/1.0')
self.assertIsNotNone(MockRestApiServer(RestApiHandler, b'POST /restart HTTP/1.0'))
MockRestApiServer(RestApiHandler, b'POST /restart HTTP/1.0\nAuthorization:')
def test_do_POST_restart(self):
request = b'POST /restart HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0'
MockRestApiServer(RestApiHandler, request)
self.assertIsNotNone(MockRestApiServer(RestApiHandler, request))
with patch.object(MockHa, 'restart', Mock(side_effect=Exception)):
MockRestApiServer(RestApiHandler, request)
@@ -140,39 +132,74 @@ class TestRestApiHandler(unittest.TestCase):
with patch.object(MockHa, 'schedule_reinitialize', Mock(return_value=None)):
MockRestApiServer(RestApiHandler, request)
cluster.leader.name = 'test'
MockRestApiServer(RestApiHandler, request)
self.assertIsNotNone(MockRestApiServer(RestApiHandler, request))
@patch('time.sleep', Mock())
def test_RestApiServer_query(self):
with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)):
MockRestApiServer(RestApiHandler, b'GET /patroni')
self.assertIsNotNone(MockRestApiServer(RestApiHandler, b'GET /patroni'))
with patch.object(MockPostgresql, 'connection', Mock(side_effect=psycopg2.OperationalError)):
MockRestApiServer(RestApiHandler, b'GET /patroni')
self.assertIsNotNone(MockRestApiServer(RestApiHandler, b'GET /patroni'))
@patch('time.sleep', Mock())
@patch.object(MockHa, 'dcs')
def test_do_POST_failover(self, dcs):
cluster = dcs.get_cluster.return_value
request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\n' +\
b'Content-Length: 0\n\n'
MockRestApiServer(RestApiHandler, request)
cluster.leader.name = 'postgresql1'
MockRestApiServer(RestApiHandler, request)
request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\n' +\
b'Content-Length: 25\n\n{"leader": "postgresql1"}'
MockRestApiServer(RestApiHandler, request)
cluster.leader.name = 'postgresql2'
request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\n' +\
b'Content-Length: 53\n\n{"leader": "postgresql1", "candidate": "postgresql2"}'
MockRestApiServer(RestApiHandler, request)
cluster.leader.name = 'postgresql1'
MockRestApiServer(RestApiHandler, request)
cluster.members = [Member(0, 'postgresql0', 30, {'api_url': 'http'})]
cluster.members = [Member(0, 'postgresql0', 30, {'api_url': 'http'}),
Member(0, 'postgresql2', 30, {'api_url': 'http'})]
MockRestApiServer(RestApiHandler, request)
with patch.object(MockPatroni, 'dcs') as d:
cluster = d.get_cluster.return_value
cluster.leader.name = 'postgresql0'
MockRestApiServer(RestApiHandler, request)
cluster.leader.name = 'postgresql2'
MockRestApiServer(RestApiHandler, request)
cluster.leader.name = 'postgresql1'
cluster.failover = None
MockRestApiServer(RestApiHandler, request)
d.get_cluster = Mock(side_effect=Exception())
d.get_cluster = Mock(side_effect=Exception)
MockRestApiServer(RestApiHandler, request)
d.manual_failover.return_value = False
MockRestApiServer(RestApiHandler, request)
with patch.object(MockHa, 'fetch_nodes_statuses', Mock(return_value=[])):
MockRestApiServer(RestApiHandler, request)
request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\n' +\
b'Content-Length: 50\n\n{"leader": "postgresql1", "member": "postgresql2"}'
# Valid future date
request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\nContent-Length: 103\n\n{"leader": ' +\
b'"postgresql1", "member": "postgresql2", "scheduled_at": "6016-02-15T18:13:30.568224+01:00"}'
MockRestApiServer(RestApiHandler, request)
# Exception: No timezone specified
request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\nContent-Length: 97\n\n{"leader": ' +\
b'"postgresql1", "member": "postgresql2", "scheduled_at": "6016-02-15T18:13:30.568224"}'
MockRestApiServer(RestApiHandler, request)
# Exception: Scheduled in the past
request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\nContent-Length: 103\n\n{"leader": ' +\
b'"postgresql1", "member": "postgresql2", "scheduled_at": "1016-02-15T18:13:30.568224+01:00"}'
MockRestApiServer(RestApiHandler, request)
# Invalid date
request = b'POST /failover HTTP/1.0\nAuthorization: Basic dGVzdDp0ZXN0\nContent-Length: 103\n\n{"leader": ' +\
b'"postgresql1", "member": "postgresql2", "scheduled_at": "2010-02-29T18:13:30.568224+01:00"}'
self.assertIsNotNone(MockRestApiServer(RestApiHandler, request))
+151 -196
View File
@@ -1,25 +1,20 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import etcd
import os
import pytest
import requests.exceptions
import unittest
import psycopg2
import requests
import patroni.exceptions
import etcd
from mock import patch, Mock, MagicMock
from click.testing import CliRunner
from mock import patch, Mock
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 patroni.exceptions import PatroniCtlException
from psycopg2 import OperationalError
from test_etcd import etcd_read, requests_get, socket_getaddrinfo, MockResponse
from test_zookeeper import MockKazooClient
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, socket_getaddrinfo, MockResponse
get_cluster_initialized_with_only_leader
from test_postgresql import MockConnect, psycopg2_connect
CONFIG_FILE_PATH = './test-ctl.yaml'
@@ -56,190 +51,177 @@ class TestCtl(unittest.TestCase):
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):
c = get_cursor(get_cluster_initialized_without_leader(), role='master')
assert c is None
self.assertIsNone(get_cursor(get_cluster_initialized_without_leader(), role='master'))
c = get_cursor(get_cluster_initialized_with_leader(), role='master')
assert c is not None
self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), role='master'))
c = get_cursor(get_cluster_initialized_with_leader(), role='replica')
# # MockCursor returns pg_is_in_recovery as false
assert c is None
# MockCursor returns pg_is_in_recovery as false
self.assertIsNone(get_cursor(get_cluster_initialized_with_leader(), role='replica'))
c = get_cursor(get_cluster_initialized_with_leader(), role='any')
assert c is not None
self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), role='any'))
def test_output_members(self):
cluster = get_cluster_initialized_with_leader()
output_members(cluster, name='abc', fmt='pretty')
output_members(cluster, name='abc', fmt='json')
output_members(cluster, name='abc', fmt='tsv')
self.assertIsNone(output_members(cluster, name='abc', fmt='pretty'))
self.assertIsNone(output_members(cluster, name='abc', fmt='json'))
self.assertIsNone(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))
@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):
with patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())):
result = self.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 = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
y''')
assert 'leader' in result.output
result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
other
2100-01-01T12:23:00
y''')
assert result.exit_code == 0
result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
other
2030-01-01T12:23:00
y''')
assert result.exit_code == 0
# Aborting failover,as we anser NO to the confirmation
result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
other
N''')
assert 'Aborting failover' in str(result.output)
assert result.exit_code == 1
result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
# Target and source are equal
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 = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
y''')
assert result.exit_code == 1
# Reality is not part of this cluster
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)
assert result.exit_code == 1
result = self.runner.invoke(ctl, ['failover', 'dummy', '--force'])
assert 'Failing over to new leader' in result.output
result = self.runner.invoke(ctl, ['failover', 'dummy', '--force'])
assert 'Member' in result.output
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)
result = self.runner.invoke(ctl, ['failover', 'dummy', '--force', '--scheduled', '2015-01-01T12:00:00+01:00'])
assert result.exit_code == 0
# Invalid timestamp
result = self.runner.invoke(ctl, ['failover', 'dummy', '--force', '--scheduled', 'invalid'])
assert result.exit_code != 0
# Invalid timestamp
result = self.runner.invoke(ctl, ['failover', 'dummy', '--force', '--scheduled', '2115-02-30T12:00:00+01:00'])
assert result.exit_code != 0
# Specifying wrong leader
result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='dummy')
assert result.exit_code == 1
with patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_only_leader())):
# No members available
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)
assert result.exit_code == 1
with patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_without_leader())):
# No master available
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())):
y''')
assert result.exit_code == 1
with patch('patroni.ctl.post_patroni', Mock(side_effect=Exception)):
# Non-responding patroni
result = self.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)):
with patch('patroni.ctl.post_patroni') as mocked:
mocked.return_value.status_code = 500
result = self.runner.invoke(ctl, ['failover', 'dummy', '--dcs', '8.8.8.8'], input='''leader
other
y''')
assert 'Failover failed, details' in result.output
assert 'Failover failed' in result.output
# with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=get_cluster_initialized_with_leader())):
# 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 = 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')
@patch('patroni.zookeeper.KazooClient', MockKazooClient)
@patch('requests.get', requests_get)
def test_get_dcs(self):
self.assertIsNotNone(get_dcs({'dcs': {'scheme': 'zookeeper', 'hostname': 'foo', 'port': 2181}}, 'dummy'))
self.assertIsNotNone(get_dcs({'dcs': {'scheme': 'exhibitor', 'hostname': 'exhibitor', 'port': 8181}}, 'dummy'))
self.assertRaises(PatroniCtlException, get_dcs, {'scheme': 'dummy'}, 'dummy')
@patch('psycopg2.connect', psycopg2_connect)
@patch('patroni.ctl.query_member', Mock(return_value=([['mock column']], None)))
@patch.object(etcd.Client, 'read', etcd_read)
def test_query(self):
with patch('patroni.ctl.get_dcs', Mock(return_value=self.e)):
result = self.runner.invoke(ctl, [
'query',
'alpha',
'--member',
'abc',
'--role',
'master',
])
assert 'mutually exclusive' in str(result.output)
# Mutually exclusive
result = self.runner.invoke(ctl, ['query', 'alpha', '--member', 'abc', '--role', 'master'])
assert result.exit_code == 1
with self.runner.isolated_filesystem():
with open('dummy', 'w') as dummy_file:
dummy_file.write('SELECT 1')
result = self.runner.invoke(ctl, [
'query',
'alpha'
])
assert 'You need to specify' in str(result.output)
result = self.runner.invoke(ctl, [
'query',
'alpha'
])
assert 'You need to specify' in str(result.output)
result = self.runner.invoke(ctl, [
'query',
'alpha',
'--file',
'dummy',
'--command',
'dummy',
])
assert 'mutually exclusive' in str(result.output)
# Mutually exclusive
result = self.runner.invoke(ctl, ['query', 'alpha', '--file', 'dummy', '--command', 'dummy'])
assert result.exit_code == 1
result = self.runner.invoke(ctl, ['query', 'alpha', '--file', 'dummy'])
assert result.exit_code == 0
os.remove('dummy')
result = self.runner.invoke(ctl, ['query', 'alpha', '--command', 'SELECT 1'])
assert 'mock column' in result.output
result = self.runner.invoke(ctl, ['query', 'alpha', '--command', 'SELECT 1', '--dbname', 'dummy',
'--password', '--username', 'dummy'], input='password\n')
# --command or --file is mandatory
result = self.runner.invoke(ctl, ['query', 'alpha'])
assert result.exit_code == 1
result = self.runner.invoke(ctl, ['query', 'alpha', '--command', 'SELECT 1', '--username', 'root',
'--password', '--dbname', 'postgres'], input='ab\nab')
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)
with patch('patroni.ctl.get_cursor', Mock(return_value=MockConnect().cursor())):
rows = query_member(None, None, None, 'master', 'SELECT pg_is_in_recovery()')
self.assertTrue('False' in str(rows))
rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()')
assert rows == (None, None)
rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()')
self.assertEquals(rows, (None, None))
with patch('test_postgresql.MockCursor.execute', Mock(side_effect=OperationalError('bla'))):
rows = query_member(None, None, None, 'replica', 'SELECT pg_is_in_recovery()')
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)
self.assertTrue('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)
self.assertTrue('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'))):
with patch('patroni.ctl.get_cursor', Mock(side_effect=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()))
@@ -248,87 +230,73 @@ y''')
result = self.runner.invoke(ctl, ['dsn', 'alpha', '--dcs', '8.8.8.8'])
assert 'host=127.0.0.1 port=5435' in result.output
result = self.runner.invoke(ctl, [
'dsn',
'alpha',
'--role',
'master',
'--member',
'dummy',
])
assert 'mutually exclusive' in str(result.output)
# Mutually exclusive options
result = self.runner.invoke(ctl, ['dsn', 'alpha', '--role', 'master', '--member', 'dummy'])
assert result.exit_code == 1
# Non-existing member
result = self.runner.invoke(ctl, ['dsn', 'alpha', '--member', 'dummy'])
assert 'Can not find' in str(result.output)
# 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
assert result.exit_code == 1
@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):
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')
assert result.exit_code == 0
result = self.runner.invoke(ctl, ['reinit', 'alpha', '--dcs', '8.8.8.8'], input='y')
assert result.exit_code == 1
# Aborted restart
result = self.runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='N')
result = self.runner.invoke(ctl, [
'restart',
'alpha',
'--dcs',
'8.8.8.8',
'dummy',
'--any',
], input='y')
assert 'not a member' in str(result.output)
assert result.exit_code == 1
# Not a member
result = self.runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8', 'dummy', '--any'], input='y')
assert result.exit_code == 1
with patch('requests.post', Mock(return_value=MockResponse())):
result = self.runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='y')
assert result.exit_code == 0
@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.object(etcd.Client, 'delete', Mock(side_effect=etcd.EtcdException))
def test_remove(self):
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)
with patch('patroni.ctl.get_dcs', Mock(return_value=self.e)):
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
# Not typing an exact confirmation
assert result.exit_code == 1
result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='''alpha
# master specified does not match master of cluster
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)
assert result.exit_code == 1
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)
# cluster specified on cmdline does not match verification prompt
result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='beta\nleader')
assert result.exit_code == 1
with patch('patroni.etcd.Etcd.get_cluster', get_cluster_initialized_with_leader):
result = self.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 = 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)
assert result.exit_code == 0
@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)
self.assertRaises(PatroniCtlException, wait_for_leader, self.e, 0)
cluster = wait_for_leader(dcs=dcs, timeout=2)
cluster = wait_for_leader(self.e, timeout=2)
assert cluster.leader.member.name == 'leader'
@patch('requests.post', Mock(side_effect=requests.exceptions.ConnectionError('foo')))
def test_post_patroni(self):
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', {})
member = get_cluster_initialized_with_leader().leader.member
self.assertRaises(requests.exceptions.ConnectionError, post_patroni, member, 'dummy', {})
def test_ctl(self):
self.runner.invoke(ctl, ['list'])
@@ -337,43 +305,30 @@ leader''')
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
self.assertIsNone(get_any_member(get_cluster_initialized_without_leader(), role='master'))
m = get_any_member(get_cluster_initialized_with_leader(), role='master')
assert m.name == 'leader'
self.assertEquals(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
self.assertEquals(list(get_all_members(get_cluster_initialized_without_leader(), role='master')), [])
r = list(get_all_members(get_cluster_initialized_with_leader(), role='master'))
assert len(r) == 1
assert r[0].name == 'leader'
self.assertEquals(len(r), 1)
self.assertEquals(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'
self.assertEquals(len(r), 1)
self.assertEquals(r[0].name, 'other')
r = list(get_all_members(get_cluster_initialized_without_leader(), role='replica'))
assert len(r) == 2
self.assertEquals(len(list(get_all_members(get_cluster_initialized_without_leader(), role='replica'))), 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):
result = self.runner.invoke(members, ['alpha'])
assert result.exit_code == 0
def test_configure(self):
result = self.runner.invoke(configure, [
'--dcs',
'abc',
'-c',
'dummy',
'-n',
'bla',
])
result = self.runner.invoke(configure, ['--dcs', 'abc', '-c', 'dummy', '-n', 'bla'])
assert result.exit_code == 0
+21 -44
View File
@@ -7,8 +7,9 @@ import unittest
from dns.exception import DNSException
from mock import Mock, patch
from patroni.dcs import Cluster, DCSError, Leader
from patroni.dcs import Cluster
from patroni.etcd import Client, Etcd, EtcdError
from patroni.exceptions import DCSError
class MockResponse(object):
@@ -24,11 +25,7 @@ class MockResponse(object):
@property
def data(self):
if self.content == 'TimeoutError':
raise urllib3.exceptions.TimeoutError
if self.content == 'Exception':
raise Exception
return self.content
return self.content.encode('utf-8')
@property
def status(self):
@@ -39,16 +36,6 @@ class MockResponse(object):
return ''
class MockPostgresql(Mock):
server_version = '999999'
scope = 'dummy'
@staticmethod
def last_operation():
return '0'
def requests_get(url, **kwargs):
members = '[{"id":14855829450254237642,"peerURLs":["http://localhost:2380","http://localhost:7001"],' +\
'"name":"default","clientURLs":["http://localhost:2379","http://localhost:4001"]}]'
@@ -58,10 +45,7 @@ def requests_get(url, **kwargs):
elif ':8011/patroni' in url:
response.content = '{"role": "replica", "xlog": {"replayed_location": 0}, "tags": {}}'
elif url.endswith('/members'):
if url.startswith('http://error'):
response.content = '[{}]'
else:
response.content = members
response.content = '[{}]' if url.startswith('http://error') else members
elif url.startswith('http://exhibitor'):
response.content = '{"servers":["127.0.0.1","127.0.0.2","127.0.0.3"],"port":2181}'
else:
@@ -70,20 +54,18 @@ def requests_get(url, **kwargs):
return response
def etcd_watch(key, index=None, timeout=None, recursive=None):
def etcd_watch(self, key, index=None, timeout=None, recursive=None):
if timeout == 2.0:
raise urllib3.exceptions.TimeoutError
raise etcd.EtcdWatchTimedOut
elif timeout == 5.0:
return etcd.EtcdResult('delete', {})
elif timeout == 10.0:
raise etcd.EtcdException
elif index == 20729:
return etcd.EtcdResult('set', {'value': 'postgresql1', 'modifiedIndex': index + 1})
elif index == 20731:
return etcd.EtcdResult('set', {'value': 'postgresql2', 'modifiedIndex': index + 1})
def etcd_write(key, value, **kwargs):
def etcd_write(self, key, value, **kwargs):
if key == '/service/exists/leader':
raise etcd.EtcdAlreadyExist
if key in ['/service/test/leader', '/patroni/test/leader'] and \
@@ -92,7 +74,7 @@ def etcd_write(key, value, **kwargs):
raise etcd.EtcdException
def etcd_read(key, **kwargs):
def etcd_read(self, key, **kwargs):
if key == '/service/noleader/':
raise DCSError('noleader')
elif key == '/service/nocluster/':
@@ -149,6 +131,8 @@ def socket_getaddrinfo(*args):
def http_request(method, url, **kwargs):
if url == 'http://localhost:2379/timeout':
raise urllib3.exceptions.ReadTimeoutError(None, None, None)
if url == 'http://localhost:2379/':
return MockResponse()
raise socket.error
@@ -166,31 +150,27 @@ class TestClient(unittest.TestCase):
mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])
self.client = Client({'discovery_srv': 'test'})
self.client.http.request = http_request
self.client.http.request_encode_body = http_request
def test_api_execute(self):
self.client._base_uri = 'http://localhost:4001'
self.client._machines_cache = ['http://localhost:2379']
self.client.api_execute('/', 'GET')
self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'})
self.client._update_machines_cache = False
self.client.api_execute('/', 'POST', timeout=0)
self.client._update_machines_cache = False
self.client._base_uri = 'http://localhost:4001'
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')
self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', '')
self.assertRaises(ValueError, self.client.api_execute, '', '')
def test_get_srv_record(self):
self.assertEquals(self.client.get_srv_record('blabla'), [])
self.assertEquals(self.client.get_srv_record('exception'), [])
def test__result_from_response(self):
response = MockResponse()
response.content = 'TimeoutError'
self.assertRaises(urllib3.exceptions.TimeoutError, self.client._result_from_response, response)
response.content = 'Exception'
self.assertRaises(etcd.EtcdException, self.client._result_from_response, response)
response.content = b'{}'
self.assertRaises(etcd.EtcdException, self.client._result_from_response, response)
def test__get_machines_cache_from_srv(self):
self.client.get_srv_record = Mock(return_value=[('localhost', 2380)])
self.client._get_machines_cache_from_srv('blabla')
@@ -206,15 +186,15 @@ class TestClient(unittest.TestCase):
@patch('requests.get', requests_get)
@patch.object(etcd.Client, 'write', etcd_write)
@patch.object(etcd.Client, 'read', etcd_read)
@patch.object(etcd.Client, 'delete', Mock(side_effect=etcd.EtcdException))
class TestEtcd(unittest.TestCase):
def setUp(self):
with patch.object(Client, 'machines') as mock_machines:
mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])
self.etcd = Etcd('foo', {'namespace': '/patroni/', 'ttl': 30, 'host': 'localhost:2379', 'scope': 'test'})
self.etcd.client.write = etcd_write
self.etcd.client.read = etcd_read
self.etcd.client.delete = Mock(side_effect=etcd.EtcdException())
def test_base_path(self):
self.assertEquals(self.etcd._base_path, '/patroni/test')
@@ -232,11 +212,8 @@ class TestEtcd(unittest.TestCase):
cluster = self.etcd.get_cluster()
self.assertIsInstance(cluster, Cluster)
self.assertIsNone(cluster.leader)
def test_current_leader(self):
self.assertIsInstance(self.etcd.current_leader(), Leader)
self.etcd._base_path = '/service/noleader'
self.assertIsNone(self.etcd.current_leader())
self.assertRaises(EtcdError, self.etcd.get_cluster)
def test_touch_member(self):
self.assertFalse(self.etcd.touch_member('', ''))
@@ -265,8 +242,8 @@ class TestEtcd(unittest.TestCase):
def test_delete_leader(self):
self.assertFalse(self.etcd.delete_leader())
@patch.object(etcd.Client, 'watch', etcd_watch)
def test_watch(self):
self.etcd.client.watch = etcd_watch
self.etcd.watch(0)
self.etcd.get_cluster()
self.etcd.watch(1.5)
+80 -77
View File
@@ -1,11 +1,14 @@
import etcd
import unittest
import datetime
import pytz
from mock import Mock, MagicMock, patch
from patroni.dcs import Cluster, Failover, Leader, Member
from patroni.etcd import Client, Etcd
from patroni.exceptions import DCSError, PostgresException
from patroni.ha import Ha
from patroni.postgresql import Postgresql
from test_etcd import socket_getaddrinfo, etcd_read, etcd_write, requests_get
@@ -26,12 +29,12 @@ 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})
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'})
return get_cluster(True, l, [m, o], failover)
m1 = Member(0, 'leader', 28, {'conn_url': 'postgres://replicator:[email protected]:5435/postgres',
'api_url': 'http://127.0.0.1:8008/patroni', 'xlog_location': 4})
l = Leader(0, 0, m1) if leader else None
m2 = Member(0, 'other', 28, {'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
'api_url': 'http://127.0.0.1:8011/patroni'})
return get_cluster(True, l, [m1, m2], failover)
def get_cluster_initialized_with_leader(failover=None):
@@ -43,56 +46,6 @@ def get_cluster_initialized_with_only_leader(failover=None):
return get_cluster(True, l, [l], failover)
class MockPostgresql(Mock):
name = 'postgresql0'
role = 'replica'
state = 'running'
connection_string = 'postgres://foo@bar/postgres'
server_version = '999999'
scope = 'dummy'
@staticmethod
def is_healthy():
return True
@staticmethod
def start():
return True
@staticmethod
def is_healthiest_node(members):
return True
@staticmethod
def is_leader():
return True
@staticmethod
def xlog_position():
return 0
@staticmethod
def last_operation():
return 0
@staticmethod
def data_directory_empty():
return False
@staticmethod
def bootstrap(*args, **kwargs):
return True
@staticmethod
def check_replication_lag(last_leader_operation):
return True
@staticmethod
def check_recovery_conf(leader):
return False
class MockPatroni(object):
def __init__(self, p, d):
@@ -101,26 +54,46 @@ class MockPatroni(object):
self.api = Mock()
self.tags = {}
self.nofailover = None
self.nap_time = 10
self.replicatefrom = None
self.api.connection_string = 'http://127.0.0.1:8008'
self.clonefrom = None
def run_async(func, args=()):
return func(*args) if args else func()
@patch.object(Postgresql, 'is_running', Mock(return_value=True))
@patch.object(Postgresql, 'is_leader', Mock(return_value=True))
@patch.object(Postgresql, 'xlog_position', Mock(return_value=0))
@patch.object(Postgresql, 'call_nowait', Mock(return_value=True))
@patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False))
@patch.object(Postgresql, 'controldata', Mock(return_value={'Database system identifier': '1234567890'}))
@patch.object(Postgresql, 'sync_replication_slots', Mock())
@patch.object(Postgresql, 'write_pg_hba', Mock())
@patch.object(Postgresql, 'write_pgpass', Mock())
@patch.object(Postgresql, 'write_recovery_conf', Mock())
@patch.object(Postgresql, 'query', Mock())
@patch.object(Postgresql, 'checkpoint', Mock())
@patch.object(etcd.Client, 'write', etcd_write)
@patch.object(etcd.Client, 'read', etcd_read)
@patch.object(etcd.Client, 'delete', Mock(side_effect=etcd.EtcdException))
@patch('subprocess.call', Mock(return_value=0))
class TestHa(unittest.TestCase):
@patch('socket.getaddrinfo', socket_getaddrinfo)
@patch.object(etcd.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.p = MockPostgresql()
self.p.can_create_replica_without_leader = MagicMock(return_value=False)
self.p = Postgresql({'name': 'postgresql0', 'scope': 'dummy', 'listen': '127.0.0.1:5432',
'data_dir': 'data/postgresql0', 'superuser': {}, 'admin': {},
'replication': {'username': '', 'password': '', 'network': ''}})
self.p.set_state('running')
self.p.check_replication_lag = true
self.p.can_create_replica_without_replication_connection = 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()
@@ -137,25 +110,25 @@ class TestHa(unittest.TestCase):
def test_start_as_replica(self):
self.p.is_healthy = false
self.assertEquals(self.ha.run_cycle(), 'started as a secondary')
self.assertEquals(self.ha.run_cycle(), 'starting as a secondary')
def test_recover_replica_failed(self):
self.p.controldata = lambda: {'Database cluster state': 'in production'}
self.p.is_healthy = false
self.p.is_running = false
self.p.follow = false
self.assertEquals(self.ha.run_cycle(), 'started as a secondary')
self.assertEquals(self.ha.run_cycle(), 'starting as a secondary')
self.assertEquals(self.ha.run_cycle(), 'failed to start postgres')
def test_recover_master_failed(self):
self.p.follow = false
self.p.is_healthy = false
self.p.is_running = false
self.ha.has_lock = true
self.p.role = 'master'
self.p.name = 'leader'
self.p.set_role('master')
self.p.controldata = lambda: {'Database cluster state': 'in production'}
self.assertEquals(self.ha.run_cycle(), 'started as readonly because i had the session lock')
self.assertEquals(self.ha.run_cycle(), 'removed leader key after trying and failing to start postgres')
self.ha.cluster = get_cluster_initialized_with_leader()
self.assertEquals(self.ha.run_cycle(), 'starting as readonly because i had the session lock')
@patch('sys.exit', return_value=1)
@patch('patroni.ha.Ha.sysid_valid', MagicMock(return_value=True))
@@ -234,13 +207,18 @@ class TestHa(unittest.TestCase):
self.p.bootstrap = false
self.assertEquals(self.ha.bootstrap(), 'trying to bootstrap from leader')
def test_bootstrap_from_another_member(self):
self.ha.cluster = get_cluster_initialized_with_leader()
self.ha.patroni.clonefrom = 'other'
self.assertEquals(self.ha.bootstrap(), 'trying to bootstrap from replica \'other\'')
def test_bootstrap_waiting_for_leader(self):
self.ha.cluster = get_cluster_initialized_without_leader()
self.assertEquals(self.ha.bootstrap(), 'waiting for leader to bootstrap')
def test_bootstrap_without_leader(self):
self.ha.cluster = get_cluster_initialized_without_leader()
self.p.can_create_replica_without_leader = MagicMock(return_value=True)
self.p.can_create_replica_without_replication_connection = MagicMock(return_value=True)
self.assertEquals(self.ha.bootstrap(), "trying to bootstrap without leader")
def test_bootstrap_initialize_lock_failed(self):
@@ -296,40 +274,65 @@ class TestHa(unittest.TestCase):
self.assertEquals(self.ha.run_cycle(), 'failed to update leader lock during restart')
@patch('requests.get', requests_get)
@patch('time.sleep', Mock())
def test_manual_failover_from_leader(self):
self.ha.has_lock = true
self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', ''))
self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', '', None))
self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock')
self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', MockPostgresql.name))
self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', self.p.name, None))
self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock')
self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', 'blabla'))
self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, '', 'blabla', None))
self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock')
f = Failover(0, MockPostgresql.name, '')
f = Failover(0, self.p.name, '', None)
self.ha.cluster = get_cluster_initialized_with_leader(f)
self.assertEquals(self.ha.run_cycle(), 'manual failover: demoting myself')
self.ha.fetch_node_status = lambda e: (e, True, True, 0, {'nofailover': 'True'})
self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock')
# manual failover from the previous leader to us won't happen if we hold the nofailover flag
self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', MockPostgresql.name))
self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, None))
self.assertEquals(self.ha.run_cycle(), 'no action. i am the leader with the lock')
# Failover scheduled time must include timezone
scheduled = datetime.datetime.now()
self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
self.ha.run_cycle()
scheduled = datetime.datetime.utcnow().replace(tzinfo=pytz.UTC)
self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
self.assertEquals('no action. i am the leader with the lock', self.ha.run_cycle())
scheduled = scheduled + datetime.timedelta(seconds=30)
self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
self.assertEquals('no action. i am the leader with the lock', self.ha.run_cycle())
scheduled = scheduled + datetime.timedelta(seconds=-600)
self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
self.assertEquals('no action. i am the leader with the lock', self.ha.run_cycle())
scheduled = None
self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, scheduled))
self.assertEquals('no action. i am the leader with the lock', self.ha.run_cycle())
@patch('requests.get', requests_get)
def test_manual_failover_process_no_leader(self):
self.p.is_leader = false
self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', MockPostgresql.name))
self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', self.p.name, None))
self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader'))
self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader', None))
self.p.set_role('replica')
self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
self.ha.fetch_node_status = lambda e: (e, True, True, 0, {}) # accessible, in_recovery
self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, MockPostgresql.name, ''))
self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, self.p.name, '', None))
self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
self.ha.fetch_node_status = lambda e: (e, False, True, 0, {}) # inaccessible, in_recovery
self.p.set_role('replica')
self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
# set failover flag to True for all members of the cluster
# this should elect the current member, as we are not going to call the API for it.
self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other'))
self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None))
self.ha.fetch_node_status = lambda e: (e, True, True, 0, {'nofailover': 'True'}) # accessible, in_recovery
self.p.set_role('replica')
self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
# same as previous, but set the current member to nofailover. In no case it should be elected as a leader
self.ha.patroni.nofailover = True
+13 -7
View File
@@ -1,3 +1,5 @@
import etcd
import os
import sys
import time
import unittest
@@ -15,10 +17,6 @@ from test_postgresql import Postgresql, psycopg2_connect
from test_zookeeper import MockKazooClient
def time_sleep(*args):
raise SleepException()
@patch('time.sleep', Mock())
@patch('subprocess.call', Mock(return_value=0))
@patch('psycopg2.connect', psycopg2_connect)
@@ -26,6 +24,8 @@ def time_sleep(*args):
@patch.object(Postgresql, 'write_recovery_conf', Mock())
@patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock())
@patch.object(AsyncExecutor, 'run', Mock())
@patch.object(etcd.Client, 'write', etcd_write)
@patch.object(etcd.Client, 'read', etcd_read)
class TestPatroni(unittest.TestCase):
def setUp(self):
@@ -39,8 +39,6 @@ class TestPatroni(unittest.TestCase):
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):
@@ -59,10 +57,18 @@ class TestPatroni(unittest.TestCase):
self.assertRaises(SleepException, _main)
with patch.object(Patroni, 'run', Mock(side_effect=KeyboardInterrupt())):
_main()
sys.argv = ['patroni.py']
# read the content of the yaml configuration file into the environment variable
# in order to test how does patroni handle the configuration passed from the environment.
with open('postgres0.yml', 'r') as f:
os.environ[Patroni.PATRONI_CONFIG_VARIABLE] = f.read()
with patch.object(Patroni, 'run', Mock(side_effect=SleepException())):
self.assertRaises(SleepException, _main)
del os.environ[Patroni.PATRONI_CONFIG_VARIABLE]
@patch('time.sleep', Mock(side_effect=SleepException()))
def test_run(self):
self.p.ha.dcs.watch = time_sleep
self.p.ha.dcs.watch = Mock(side_effect=SleepException())
self.assertRaises(SleepException, self.p.run)
self.p.ha.state_handler.is_leader = Mock(return_value=False)
+22 -29
View File
@@ -2,21 +2,16 @@ import mock # for the mock.call method, importing it without a namespace breaks
import os
import psycopg2
import shutil
import subprocess
import unittest
from six.moves import builtins
from mock import Mock, MagicMock, PropertyMock, patch, mock_open
from patroni.dcs import Cluster, Leader, Member
from patroni.exceptions import PostgresException, PostgresConnectionException
from patroni.postgresql import Postgresql
from patroni.utils import RetryFailedError
from six.moves import builtins
from test_ha import false
import subprocess
def is_file_raise_on_backup(*args, **kwargs):
if args[0].endswith('.backup'):
raise Exception("foo")
class MockCursor(object):
@@ -38,7 +33,7 @@ class MockCursor(object):
elif sql == 'SELECT pg_is_in_recovery()':
self.results = [(False, )]
elif sql.startswith('SELECT to_char(pg_postmaster_start_time'):
self.results = [('', True, '', '', '', False)]
self.results = [('', True, '', '', '', '', False)]
else:
self.results = [(
None,
@@ -59,10 +54,6 @@ class MockCursor(object):
def fetchall(self):
return self.results
@staticmethod
def close():
pass
def __iter__(self):
for i in self.results:
yield i
@@ -74,8 +65,9 @@ class MockCursor(object):
pass
class MockConnect(Mock):
class MockConnect(object):
server_version = '99999'
autocommit = False
closed = 0
@@ -88,6 +80,10 @@ class MockConnect(Mock):
def __exit__(self, *args):
pass
@staticmethod
def close():
pass
def pg_controldata_string(*args, **kwargs):
return b"""
@@ -283,7 +279,7 @@ class TestPostgresql(unittest.TestCase):
self.p.pg_rewind = tmp
with mock.patch('subprocess.call', MagicMock(return_value=1)):
self.assertFalse(self.p.can_rewind)
with mock.patch('subprocess.call', side_effect=OSError("foo")):
with mock.patch('subprocess.call', side_effect=OSError):
self.assertFalse(self.p.can_rewind)
tmp = self.p.controldata
self.p.controldata = lambda: {'wal_log_hints setting': 'on'}
@@ -292,7 +288,7 @@ class TestPostgresql(unittest.TestCase):
@patch('time.sleep', Mock())
def test_create_replica(self):
self.p.delete_trigger_file = Mock(side_effect=OSError())
self.p.delete_trigger_file = Mock(side_effect=OSError)
with patch('subprocess.call', Mock(side_effect=[1, 0])):
self.assertEquals(self.p.create_replica(self.leader, ''), 0)
with patch('subprocess.call', Mock(side_effect=[Exception(), 0])):
@@ -349,7 +345,7 @@ class TestPostgresql(unittest.TestCase):
def test_last_operation(self):
self.assertEquals(self.p.last_operation(), '0')
@patch('subprocess.Popen', Mock(side_effect=OSError()))
@patch('subprocess.Popen', Mock(side_effect=OSError))
def test_call_nowait(self):
self.assertFalse(self.p.call_nowait('on_start'))
@@ -369,7 +365,7 @@ class TestPostgresql(unittest.TestCase):
def test_move_data_directory(self):
self.p.is_running = false
self.p.move_data_directory()
with patch('os.rename', Mock(side_effect=OSError())):
with patch('os.rename', Mock(side_effect=OSError)):
self.p.move_data_directory()
@patch('patroni.postgresql.Postgresql.write_pgpass', MagicMock(return_value=dict()))
@@ -411,13 +407,10 @@ class TestPostgresql(unittest.TestCase):
self.assertEquals(int(data['max_replication_slots']), 5)
self.assertEqual(data.get('D'), None)
m.side_effect = IOError("foo")
m.side_effect = IOError
data = self.p.read_postmaster_opts()
self.assertEqual(data, dict())
m.side_effect = Exception("foo")
self.assertRaises(Exception, self.p.read_postmaster_opts())
@patch('subprocess.Popen')
@patch.object(builtins, 'open', MagicMock(return_value=42))
def test_single_user_mode(self, subprocess_popen_mock):
@@ -479,17 +472,17 @@ class TestPostgresql(unittest.TestCase):
def test_restore_configuration_files(self):
self.p.restore_configuration_files()
def test_can_create_replica_without_leader(self):
def test_can_create_replica_without_replication_connection(self):
self.p.config['create_replica_method'] = []
self.assertFalse(self.p.can_create_replica_without_leader())
self.assertFalse(self.p.can_create_replica_without_replication_connection())
self.p.config['create_replica_method'] = ['wale', 'basebackup']
self.p.config['wale'] = {'command': 'foo', 'no_master': 1}
self.assertTrue(self.p.can_create_replica_without_leader())
self.assertTrue(self.p.can_create_replica_without_replication_connection())
def test_replica_method_can_work_without_leader(self):
self.assertFalse(self.p.replica_method_can_work_without_leader('basebackup'))
self.assertFalse(self.p.replica_method_can_work_without_leader('foobar'))
def test_replica_method_can_work_without_replication_connection(self):
self.assertFalse(self.p.replica_method_can_work_without_replication_connection('basebackup'))
self.assertFalse(self.p.replica_method_can_work_without_replication_connection('foobar'))
self.p.config['foo'] = {'command': 'bar', 'no_master': 1}
self.assertTrue(self.p.replica_method_can_work_without_leader('foo'))
self.assertTrue(self.p.replica_method_can_work_without_replication_connection('foo'))
self.p.config['foo'] = {'command': 'bar'}
self.assertFalse(self.p.replica_method_can_work_without_leader('foo'))
self.assertFalse(self.p.replica_method_can_work_without_replication_connection('foo'))
+3 -3
View File
@@ -16,14 +16,14 @@ class TestUtils(unittest.TestCase):
@patch('time.sleep', Mock())
def test_reap_children(self):
reap_children()
self.assertIsNone(reap_children())
with patch('os.waitpid', Mock(return_value=(0, 0))):
sigchld_handler(None, None)
reap_children()
self.assertIsNone(reap_children())
@patch('time.sleep', time_sleep)
def test_sleep(self):
sleep(0.01)
self.assertIsNone(sleep(0.01))
@patch('time.sleep', Mock())
+1 -22
View File
@@ -6,22 +6,6 @@ from mock import MagicMock, patch, PropertyMock
from patroni.scripts.wale_restore import WALERestore, main as _main
def fake_cursor_fetchone(*args, **kwargs):
return ('16777216',)
def fake_call_fail_for_wal_e(*args, **kwargs):
if len(args) > 0 and 'backup-fetch' in args[0]:
return 1
return 0
def fake_call_fail_for_base_backup(*args, **kwargs):
if len(args) > 0 and 'backup-fetch' in args[0]:
return 0
return 1
def fake_backup_data(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
@@ -52,7 +36,6 @@ base_00000001000000000000007F_00000040 2015-05-18T10:13:25.000Z 167772160 000
@patch('os.makedirs', MagicMock(return_value=True))
@patch('os.path.exists', MagicMock(return_value=True))
@patch('os.path.isdir', MagicMock(return_value=True))
@patch('psycopg2.extensions.cursor.fetchone', MagicMock(side_effect=fake_cursor_fetchone))
@patch('psycopg2.extensions.cursor', MagicMock(autospec=True))
@patch('psycopg2.extensions.connection', MagicMock(autospec=True))
@patch('psycopg2.connect', MagicMock(autospec=True))
@@ -60,11 +43,7 @@ base_00000001000000000000007F_00000040 2015-05-18T10:13:25.000Z 167772160 000
class TestWALERestore(unittest.TestCase):
def setUp(self):
self.wale_restore = WALERestore("batman", "/data",
"host=batman port=5432 user=batman", "/etc", 100, 100, 1, 0)
def tearDown(self):
pass
self.wale_restore = WALERestore("batman", "/data", "host=batman port=5432 user=batman", "/etc", 100, 100, 1, 0)
def test_should_use_s3_to_create_replica(self):
with patch('psycopg2.connect', MagicMock(side_effect=psycopg2.Error("foo"))):
+8 -8
View File
@@ -14,7 +14,6 @@ class MockKazooClient(Mock):
leader = False
exists = True
handler = Mock()
@property
def client_id(self):
@@ -34,8 +33,6 @@ class MockKazooClient(Mock):
b'postgres://repuser:rep-pass@localhost:5434/postgres?application_name=http://127.0.0.1:8009/patroni',
ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0 if self.exists else -1, 0, 0, 0)
)
elif path.endswith('/optime/leader'):
return (b'1', ZnodeStat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0))
elif path.endswith('/leader'):
if self.leader:
return (b'foo', ZnodeStat(0, 0, 0, 0, 0, 0, 0, -1, 0, 0, 0))
@@ -86,13 +83,11 @@ class MockKazooClient(Mock):
raise TypeError("Invalid type for 'path' (string expected)")
self.exists = False
if path == '/service/test/leader':
if self.leader:
return
self.leader = True
raise Exception
elif path == '/service/test/members/buzz':
raise Exception
elif path.endswith('/initialize') or path == '/service/test/members/bar':
elif path.endswith('/') or path.endswith('/initialize') or path == '/service/test/members/bar':
raise NoNodeError
@@ -152,16 +147,18 @@ class TestZooKeeper(unittest.TestCase):
self.zk._name = 'bar'
self.zk.touch_member('new')
self.zk._name = 'na'
self.zk.client.exists = 1
self.zk._client.exists = 1
self.zk.touch_member('exists')
self.zk._name = 'bar'
self.zk.touch_member('retry')
self.zk.fetch_cluster = True
self.zk._fetch_cluster = True
self.zk.get_cluster()
self.zk.touch_member('retry')
def test_take_leader(self):
self.zk.take_leader()
with patch.object(MockKazooClient, 'create', Mock(side_effect=Exception)):
self.zk.take_leader()
def test_update_leader(self):
self.assertTrue(self.zk.update_leader())
@@ -172,6 +169,9 @@ class TestZooKeeper(unittest.TestCase):
self.zk._base_path = self.zk._base_path.replace('test', 'bla')
self.zk.write_leader_optime('2')
def test_delete_cluster(self):
self.assertTrue(self.zk.delete_cluster())
def test_watch(self):
self.zk.watch(0)
self.zk.event.isSet = lambda: True