Implement possibility to configure retry_timeout globally

Previously it was hardcoded all over the place.

patroni/__init__.py

@@ -57,6 +57,7 @@ class Patroni(object):
             self.tags = self.get_tags()
             self.nap_time = self.config['loop_wait']
             self.dcs.set_ttl(self.config.get('ttl') or 30)
+            self.dcs.set_retry_timeout(self.config.get('retry_timeout') or self.nap_time)
             self.api.reload_config(self.config['restapi'])
             self.postgresql.reload_config(self.config['postgresql'])
         except Exception:

patroni/config.py

@@ -32,7 +32,7 @@ class Config(object):
 
     __CACHE_FILENAME = 'patroni.dynamic.json'
     __DEFAULT_CONFIG = {
-        'ttl': 30, 'loop_wait': 10, 'retry_timeout': 5,
+        'ttl': 30, 'loop_wait': 10, 'retry_timeout': 10,
         'maximum_lag_on_failover': 1048576,
         'postgresql': {
             'parameters': Postgresql.CMDLINE_OPTIONS

patroni/dcs/__init__.py

@@ -265,6 +265,10 @@ class AbstractDCS(object):
     def set_ttl(self, ttl):
         """Set the new ttl value for leader key"""
 
+    @abc.abstractmethod
+    def set_retry_timeout(self, retry_timeout):
+        """Set the new value for retry_timeout"""
+
     @abc.abstractmethod
     def _load_cluster(self):
         """Internally this method should build  `Cluster` object which

patroni/dcs/consul.py

@@ -21,9 +21,8 @@ class HTTPClient(std.HTTPClient):
 
     def __init__(self, *args, **kwargs):
         super(HTTPClient, self).__init__(*args, **kwargs)
-        self._patch_default_timeout()
 
-    def _patch_default_timeout(self):
+    def patch_default_timeout(self, timeout):
         # Set a default timeout for the `request.session.request` method, that is used
         # internally  by the methods request.session.get, request.session.post and
         # others. We monkey-patch here to avoid reimplementing each individual method from
@@ -78,6 +77,7 @@ class Consul(AbstractDCS):
         self.set_ttl(config.get('ttl') or 30)
         host, port = config.get('host', '127.0.0.1:8500').split(':')
         self._client = ConsulClient(host=host, port=port)
+        self._client.http.patch_default_timeout(config['retry_timeout']/2.0)
         self._scope = config['scope']
         self.create_or_restore_session()
 
@@ -93,7 +93,7 @@ class Consul(AbstractDCS):
                 sleep(5)
 
     def set_ttl(self, ttl):
-        ttl = int(ttl/2)  # My experiments have shown that session expires after 2*ttl time
+        ttl = ttl/2.0  # My experiments have shown that session expires after 2*ttl time
         if self._ttl != ttl:
             if self._session:
                 try:
@@ -107,6 +107,9 @@ class Consul(AbstractDCS):
             self.event.set()
         self._ttl = ttl
 
+    def set_retry_timeout(self, retry_timeout):
+        self._client.http.patch_default_timeout(retry_timeout/2.0)
+
     def refresh_session(self):
         """:returns: `!True` if it had to create new session"""
         if self._session:

patroni/dcs/etcd.py

@@ -194,7 +194,7 @@ class Etcd(AbstractDCS):
     def __init__(self, config):
         super(Etcd, self).__init__(config)
         self._ttl = int(config.get('ttl') or 30)
-        self._retry = Retry(deadline=10, max_delay=1, max_tries=-1,
+        self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1,
                             retry_exceptions=(etcd.EtcdConnectionFailed,
                                               etcd.EtcdLeaderElectionInProgress,
                                               etcd.EtcdWatcherCleared,
@@ -224,6 +224,9 @@ class Etcd(AbstractDCS):
             self.event.set()
         self._ttl = ttl
 
+    def set_retry_timeout(self, retry_timeout):
+        self._retry.deadline = retry_timeout
+
     @staticmethod
     def member(node):
         return Member.from_node(node.modifiedIndex, os.path.basename(node.key), node.ttl, node.value)

patroni/dcs/zookeeper.py

@@ -83,9 +83,8 @@ 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 config.get('ttl') or 30),
+        self._client = KazooClient(hosts=hosts, timeout=config['ttl'],
-                                   command_retry={'deadline': (config.get('reconnect_timeout') or 10),
+                                   command_retry={'deadline': config['retry_timeout'], 'max_delay': 1, 'max_tries': -1},
-                                                  'max_delay': 1, 'max_tries': -1},
                                    connection_retry={'max_delay': 1, 'max_tries': -1})
         self._client.add_listener(self.session_listener)
 
@@ -105,12 +104,15 @@ class ZooKeeper(AbstractDCS):
 
     def set_ttl(self, ttl):
         ttl = int(ttl * 1000)
-        # I know, it's weird to access private attributes and method
+        # I know, it's weird to access private attributes, but there is
-        # but there is no other way to change session_timeout without losing session
+        # no other way to change session_timeout without losing session
         if self._client._session_timeout != ttl:
             self._client._session_timeout = ttl
             self._client.restart()
 
+    def set_retry_timeout(self, retry_timeout):
+        self._client._retry.deadline = retry_timeout
+
     def get_node(self, key, watch=None):
         try:
             ret = self._client.get(key, watch)

patroni/postgresql.py

@@ -95,7 +95,8 @@ class Postgresql(object):
         self._cursor_holder = None
         self._sysid = None
         self._replication_slots = []  # list of already existing replication slots
-        self.retry = Retry(max_tries=-1, deadline=5, max_delay=1, retry_exceptions=PostgresConnectionException)
+        self.retry = Retry(max_tries=-1, deadline=config['retry_timeout']/2.0, max_delay=1,
+                           retry_exceptions=PostgresConnectionException)
 
         self._state_lock = Lock()
         self.set_state('stopped')
@@ -160,6 +161,7 @@ class Postgresql(object):
         if reload_pending:
             self._write_postgresql_conf()
             self.reload()
+        self.retry.deadline = config['retry_timeout']/2.0
 
     @property
     def restart_pending(self):

postgres0.yml

@@ -15,7 +15,7 @@ bootstrap:
   dcs:
     ttl: 30
     loop_wait: 10
-    retry_timeout: 5
+    retry_timeout: 10
     maximum_lag_on_failover: 1048576
     postgresql:
       use_pg_rewind: true

tests/test_consul.py

@@ -51,7 +51,7 @@ class TestConsul(unittest.TestCase):
     @patch.object(consul.Consul.KV, 'get', kv_get)
     @patch.object(consul.Consul.KV, 'delete', Mock())
     def setUp(self):
-        self.c = Consul({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'host': 'localhost:1'})
+        self.c = Consul({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'host': 'localhost:1', 'retry_timeout': 10})
         self.c._base_path = '/service/good'
         self.c._load_cluster()
 
@@ -133,3 +133,6 @@ class TestConsul(unittest.TestCase):
     @patch.object(consul.Consul.Session, 'destroy', Mock(side_effect=ConsulException))
     def test_set_ttl(self):
         self.c.set_ttl(20)
+
+    def test_set_retry_timeout(self):
+        self.c.set_retry_timeout(10)

tests/test_ctl.py

@@ -51,7 +51,7 @@ class TestCtl(unittest.TestCase):
         self.runner = CliRunner()
         with patch.object(etcd.Client, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
-            self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379'}}, 'foo')
+            self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'retry_timeout': 10}}, 'foo')
 
     @patch('psycopg2.connect', psycopg2_connect)
     def test_get_cursor(self):

tests/test_etcd.py

@@ -193,7 +193,7 @@ 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({'namespace': '/patroni/', 'ttl': 30,
+            self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10,
                               'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'})
 
     def test_base_path(self):

tests/test_ha.py

@@ -102,7 +102,7 @@ class TestHa(unittest.TestCase):
         with patch.object(etcd.Client, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
             self.p = Postgresql({'name': 'postgresql0', 'scope': 'dummy', 'listen': '127.0.0.1:5432',
-                                 'data_dir': 'data/postgresql0',
+                                 'data_dir': 'data/postgresql0', 'retry_timeout': 10,
                                  'authentication': {'superuser': {'username': 'foo', 'password': 'bar'},
                                                     'replication': {'username': '', 'password': ''}},
                                  'parameters': {'wal_level': 'hot_standby', 'max_replication_slots': 5, 'foo': 'bar',
@@ -111,7 +111,8 @@ class TestHa(unittest.TestCase):
             self.p.set_role('replica')
             self.p.check_replication_lag = true
             self.p.can_create_replica_without_replication_connection = MagicMock(return_value=False)
-            self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test', 'name': 'foo'}})
+            self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test',
+                                       'name': 'foo', 'retry_timeout': 10}})
             self.ha = Ha(MockPatroni(self.p, self.e))
             self.ha._async_executor.run_async = run_async
             self.ha.old_cluster = self.e.get_cluster()

tests/test_postgresql.py

@@ -160,7 +160,7 @@ class TestPostgresql(unittest.TestCase):
         self.data_dir = 'data/test0'
         if not os.path.exists(self.data_dir):
             os.makedirs(self.data_dir)
-        self.p = Postgresql({'name': 'test0', 'scope': 'batman', 'data_dir': self.data_dir,
+        self.p = Postgresql({'name': 'test0', 'scope': 'batman', 'data_dir': self.data_dir, 'retry_timeout': 10,
                              'listen': '127.0.0.1, *:5432', 'connect_address': '127.0.0.2:5432',
                              'authentication': {'superuser': {'username': 'test', 'password': 'test'},
                                                 'replication': {'username': 'replicator', 'password': 'rep-pass'}},
@@ -475,8 +475,8 @@ class TestPostgresql(unittest.TestCase):
         self.assertFalse(self.p.replica_method_can_work_without_replication_connection('foo'))
 
     def test_reload_config(self):
-        self.p.reload_config({'listen': '*', 'parameters': self._PARAMETERS})
+        self.p.reload_config({'retry_timeout': 10, 'listen': '*', 'parameters': self._PARAMETERS})
-        self.p.reload_config({'listen': '*:5433', 'parameters': self._PARAMETERS})
+        self.p.reload_config({'retry_timeout': 10, 'listen': '*:5433', 'parameters': self._PARAMETERS})
 
     @patch.object(builtins, 'open', mock_open(read_data='9.4'))
     def test_get_major_version(self):

tests/test_zookeeper.py

@@ -104,7 +104,7 @@ class TestZooKeeper(unittest.TestCase):
     @patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient)
     def setUp(self):
         self.zk = ZooKeeper({'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181},
-                             'scope': 'test', 'name': 'foo'})
+                             'scope': 'test', 'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
 
     def test_session_listener(self):
         self.zk.session_listener(KazooState.SUSPENDED)
@@ -112,6 +112,9 @@ class TestZooKeeper(unittest.TestCase):
     def test_set_ttl(self):
         self.zk.set_ttl(20)
 
+    def test_set_retry_timeout(self):
+        self.zk.set_retry_timeout(10)
+
     def test_get_node(self):
         self.assertIsNone(self.zk.get_node('/no_node'))