diff --git a/.travis.yml b/.travis.yml index ba4733d3..73d49104 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,14 +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: + - 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 - - pip install coveralls codacy-coverage + - 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/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..f97e059f --- /dev/null +++ b/features/patroni_api.feature @@ -0,0 +1,48 @@ +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 member" + +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 + Given replication works from postgres0 to postgres1 after 10 seconds + 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 + +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 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 postgresq0 in 10 seconds + Then I receive a response code 200 + And postgres0 is a leader after 15 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/ha.py b/patroni/ha.py index d273931a..8c841290 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -463,8 +463,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 d8f649c7..df17200d 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)) @@ -509,7 +509,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")