Implement acceptance test for dynamic configuration functionality

and fix some bugs revealed by acceptance tests

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

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:

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)

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

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

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:

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)

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)