Merge pull request #153 from zalando/feature/acceptance_tests_behave

Feature/acceptance tests behave

.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

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

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

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()

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
+

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))

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)

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))))

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

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")