diff --git a/.travis.yml b/.travis.yml index 23c69358..73d49104 100644 --- a/.travis.yml +++ b/.travis.yml @@ -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 diff --git a/.zappr.yml b/.zappr.yml new file mode 100644 index 00000000..4d628636 --- /dev/null +++ b/.zappr.yml @@ -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 diff --git a/TODO.md b/TODO.md index 17d1cb0c..91b77499 100644 --- a/TODO.md +++ b/TODO.md @@ -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. diff --git a/features/basic_replication.feature b/features/basic_replication.feature new file mode 100644 index 00000000..b3307e40 --- /dev/null +++ b/features/basic_replication.feature @@ -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 diff --git a/features/cascading_replication.feature b/features/cascading_replication.feature new file mode 100644 index 00000000..8a3d2f80 --- /dev/null +++ b/features/cascading_replication.feature @@ -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 diff --git a/features/environment.py b/features/environment.py new file mode 100644 index 00000000..73f77011 --- /dev/null +++ b/features/environment.py @@ -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() diff --git a/features/patroni_api.feature b/features/patroni_api.feature new file mode 100644 index 00000000..3c408dfc --- /dev/null +++ b/features/patroni_api.feature @@ -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 + diff --git a/features/steps/basic_replication.py b/features/steps/basic_replication.py new file mode 100644 index 00000000..364de61f --- /dev/null +++ b/features/steps/basic_replication.py @@ -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)) diff --git a/features/steps/cascading_replication.py b/features/steps/cascading_replication.py new file mode 100644 index 00000000..59399c97 --- /dev/null +++ b/features/steps/cascading_replication.py @@ -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) diff --git a/features/steps/patroni_api.py b/features/steps/patroni_api.py new file mode 100644 index 00000000..0f802dee --- /dev/null +++ b/features/steps/patroni_api.py @@ -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)))) diff --git a/patroni/__init__.py b/patroni/__init__.py index 36c8feac..87f0b754 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -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() diff --git a/patroni/api.py b/patroni/api.py index 07767263..cb7b2d1f 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -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] }) } diff --git a/patroni/ctl.py b/patroni/ctl.py index fb4c241b..b14db933 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -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([ diff --git a/patroni/dcs.py b/patroni/dcs.py index ad426fbc..689470fc 100644 --- a/patroni/dcs.py +++ b/patroni/dcs.py @@ -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 diff --git a/patroni/etcd.py b/patroni/etcd.py index 79842ca4..3f798cf8 100644 --- a/patroni/etcd.py +++ b/patroni/etcd.py @@ -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') diff --git a/patroni/ha.py b/patroni/ha.py index 9e9f88aa..d4d7fb93 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -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 diff --git a/patroni/postgresql.py b/patroni/postgresql.py index f5f3d37c..b3e2ab00 100644 --- a/patroni/postgresql.py +++ b/patroni/postgresql.py @@ -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): diff --git a/patroni/scripts/wale_restore.py b/patroni/scripts/wale_restore.py index c80cdfae..f3691907 100755 --- a/patroni/scripts/wale_restore.py +++ b/patroni/scripts/wale_restore.py @@ -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, diff --git a/patroni/utils.py b/patroni/utils.py index a5db5638..cb101758 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -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\d{4})\-(?P\d{2})\-(?P\d{2}) # date -T -(?P\d{2}):(?P\d{2}):(?P\d{2})\.(?P\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()) diff --git a/patroni/version.py b/patroni/version.py index b7831c97..45b09904 100644 --- a/patroni/version.py +++ b/patroni/version.py @@ -1 +1 @@ -__version__ = '0.76' +__version__ = '0.80' diff --git a/patroni/zookeeper.py b/patroni/zookeeper.py index 79e8a72a..13faa4d3 100644 --- a/patroni/zookeeper.py +++ b/patroni/zookeeper.py @@ -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 diff --git a/requirements-py3.txt b/requirements-py3.txt deleted file mode 100644 index 13c3010c..00000000 --- a/requirements-py3.txt +++ /dev/null @@ -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 diff --git a/requirements-py2.txt b/requirements.txt similarity index 67% rename from requirements-py2.txt rename to requirements.txt index 1e194bc0..66b059f7 100644 --- a/requirements-py2.txt +++ b/requirements.txt @@ -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 diff --git a/setup.py b/setup.py index 437354e2..a8936d7c 100644 --- a/setup.py +++ b/setup.py @@ -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}, ) diff --git a/tests/test_api.py b/tests/test_api.py index 2faee523..1a3d00ce 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -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)) diff --git a/tests/test_ctl.py b/tests/test_ctl.py index fb63aebd..95dd5a07 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -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 diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 20b2ad7c..742c57c5 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -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) diff --git a/tests/test_ha.py b/tests/test_ha.py index f13bff10..d81d5bbd 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -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:rep-pass@127.0.0.1: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:rep-pass@127.0.0.1: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:rep-pass@127.0.0.1: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:rep-pass@127.0.0.1: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 diff --git a/tests/test_patroni.py b/tests/test_patroni.py index aaefdf1b..fc806026 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -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) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 4c701c60..932e860d 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -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')) diff --git a/tests/test_utils.py b/tests/test_utils.py index 6f66f4c3..740fef03 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -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()) diff --git a/tests/test_wale_restore.py b/tests/test_wale_restore.py index 40761051..9e15dc9b 100644 --- a/tests/test_wale_restore.py +++ b/tests/test_wale_restore.py @@ -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"))): diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index f5cbdf13..5e9ebc82 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -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