diff --git a/features/basic_replication.feature b/features/basic_replication.feature index be192715..ba347f45 100644 --- a/features/basic_replication.feature +++ b/features/basic_replication.feature @@ -1,5 +1,5 @@ Feature: basic replication - We should check that the basic bootstrapping, replication and failover works. + We should check that the basic bootstrapping, replication, failover and dyncamic configuration works. Scenario: check replication of a single table Given I start postgres0 @@ -8,10 +8,15 @@ Feature: basic replication When I add the table foo to postgres0 Then table foo is present on postgres1 after 20 seconds + Scenario: check dynamic configuration change via DCS + When I patch global configuration with {"ttl": 20, "loop_wait": 5, "postgresql": {"parameters": {"max_connections": 101}}} + Then Response on GET http://127.0.0.1:8008/patroni contains restart_pending after 11 seconds + And Response on GET http://127.0.0.1:8009/patroni contains restart_pending after 11 seconds + Scenario: check the basic failover - When I kill postgres0 - Then postgres1 role is the primary after 32 seconds + And I kill postgres0 + Then postgres1 role is the primary after 22 seconds When I start postgres0 Then postgres0 role is the secondary after 20 seconds When I add the table bar to postgres1 - Then table bar is present on postgres0 after 20 seconds + Then table bar is present on postgres0 after 10 seconds diff --git a/features/environment.py b/features/environment.py index 16c6a3fe..01e94ce0 100644 --- a/features/environment.py +++ b/features/environment.py @@ -182,7 +182,7 @@ class PatroniController(AbstractController): class AbstractDcsController(AbstractController): - _CLUSTER_NODE = 'service/batman' + _CLUSTER_NODE = '/service/batman' def _is_accessible(self): return self._is_running() @@ -193,10 +193,17 @@ class AbstractDcsController(AbstractController): if self._work_directory: shutil.rmtree(self._work_directory) + def path(self, key=None): + return self._CLUSTER_NODE + (key and '/' + key or '') + @abc.abstractmethod def query(self, key): """ query for a value of a given key """ + @abc.abstractmethod + def set(self, key, value): + """ set a value to a given key """ + @abc.abstractmethod def cleanup_service_tree(self): """ clean all contents stored in the tree used for the tests """ @@ -218,12 +225,18 @@ class ConsulController(AbstractDcsController): except Exception: return False + def path(self, key=None): + return super(ConsulController, self).path(key)[1:] + def query(self, key): - _, value = self._client.kv.get('{0}/{1}'.format(self._CLUSTER_NODE, key)) + _, value = self._client.kv.get(self.path(key)) return value and value['Value'].decode('utf-8') + def set(self, key, value): + self._client.kv.put(self.path(key), value) + def cleanup_service_tree(self): - self._client.kv.delete(self._CLUSTER_NODE, recurse=True) + self._client.kv.delete(self.path(), recurse=True) class EtcdController(AbstractDcsController): @@ -240,13 +253,16 @@ class EtcdController(AbstractDcsController): def query(self, key): try: - return self._client.get('/{0}/{1}'.format(self._CLUSTER_NODE, key)).value + return self._client.get(self.path(key)).value except etcd.EtcdKeyNotFound: return None + def set(self, key, value): + self._client.set(self.path(key), value) + def cleanup_service_tree(self): try: - self._client.delete('/' + self._CLUSTER_NODE, recursive=True) + self._client.delete(self.path(), recursive=True) except (etcd.EtcdKeyNotFound, etcd.EtcdConnectionFailed): return except Exception as e: @@ -273,13 +289,16 @@ class ZooKeeperController(AbstractDcsController): def query(self, key): try: - return self._client.get('/{0}/{1}'.format(self._CLUSTER_NODE, key))[0].decode('utf-8') + return self._client.get(self.path(key))[0].decode('utf-8') except kazoo.exceptions.NoNodeError: return None + def set(self, key, value): + self._client.set(self.path(key), value.encode('utf-8')) + def cleanup_service_tree(self): try: - self._client.delete('/' + self._CLUSTER_NODE, recursive=True) + self._client.delete(self.path(), recursive=True) except (kazoo.exceptions.NoNodeError): return except Exception as e: diff --git a/features/steps/basic_replication.py b/features/steps/basic_replication.py index b59a7639..4f441dac 100644 --- a/features/steps/basic_replication.py +++ b/features/steps/basic_replication.py @@ -1,4 +1,6 @@ +import json import psycopg2 as pg +import requests from behave import step, then from time import sleep, time @@ -52,3 +54,31 @@ def replication_works(context, master, replica, time_limit): 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)) + + +def patch_config_with_data(config, data): + for name, value in data.items(): + if isinstance(value, dict): + patch_config_with_data(config[name], value) + else: + config[name] = value + + +@step('I patch global configuration with {data}') +def patch_config(context, data): + data = json.loads(data) + config = json.loads(context.dcs_ctl.query('config')) + patch_config_with_data(config, data) + context.dcs_ctl.set('config', json.dumps(config)) + + +@then('Response on GET {url} contains {value} after {timeout:d} seconds') +def check_http_response(context, url, value, timeout): + for _ in range(int(timeout)): + r = requests.get(url) + if value in r.content.decode('utf-8'): + break + sleep(1) + else: + assert False,\ + "Value {0} is not present in response after {1} seconds".format(value, timeout) diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index e941d025..499c8ae1 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -95,7 +95,14 @@ class Consul(AbstractDCS): def set_ttl(self, ttl): ttl = int(ttl/2) # My experiments have shown that session expires after 2*ttl time if self._ttl != ttl: + if self._session: + try: + self._client.session.destroy(self._session) + except Exception: + logger.exception("Can not destroy session %s", self._session) self._session = None + self.reset_cluster() + self.event.set() self._ttl = ttl def refresh_session(self): diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 0c64cbfd..beb494b5 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -193,7 +193,7 @@ class Etcd(AbstractDCS): def __init__(self, config): super(Etcd, self).__init__(config) - self.set_ttl(config.get('ttl', 30)) + self._ttl = int(config.get('ttl') or 30) self._retry = Retry(deadline=10, max_delay=1, max_tries=-1, retry_exceptions=(etcd.EtcdConnectionFailed, etcd.EtcdLeaderElectionInProgress, @@ -216,7 +216,11 @@ class Etcd(AbstractDCS): return client def set_ttl(self, ttl): - self.ttl = int(ttl) + ttl = int(ttl) + if self._ttl != ttl: + self.reset_cluster() + self.event.set() + self._ttl = ttl @staticmethod def member(node): @@ -263,15 +267,15 @@ class Etcd(AbstractDCS): @catch_etcd_errors def touch_member(self, data, ttl=None): - return self.retry(self._client.set, self.member_path, data, ttl or self.ttl) + return self.retry(self._client.set, self.member_path, data, 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): @@ -292,7 +296,7 @@ class Etcd(AbstractDCS): @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=""): diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index b374ab51..d4d680ee 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -109,7 +109,7 @@ class ZooKeeper(AbstractDCS): # but there is no other way to change session_timeout without losing session if self._client._session_timeout != ttl: self._client._session_timeout = ttl - self._client._connection._socket.close() + self._client.restart() def get_node(self, key, watch=None): try: diff --git a/tests/test_consul.py b/tests/test_consul.py index b3c5ff16..2c5bb36d 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -129,3 +129,7 @@ class TestConsul(unittest.TestCase): self.c.watch(1) with patch.object(consul.Consul.KV, 'get', Mock(side_effect=ConsulException)): self.c.watch(1) + + @patch.object(consul.Consul.Session, 'destroy', Mock(side_effect=ConsulException)) + def test_set_ttl(self): + self.c.set_ttl(20) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 5631f817..4d248e11 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -257,3 +257,6 @@ class TestEtcd(unittest.TestCase): def test_other_exceptions(self): self.etcd.retry = Mock(side_effect=AttributeError('foo')) self.assertRaises(EtcdError, self.etcd.cancel_initialization) + + def test_set_ttl(self): + self.etcd.set_ttl(20)