From b3ada161cf6d67cbd1656fa0fba92c5b61ebb6a1 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 31 May 2016 10:30:53 +0200 Subject: [PATCH] Implement possibility to configure `retry_timeout` globally Previously it was hardcoded all over the place. --- patroni/__init__.py | 1 + patroni/config.py | 2 +- patroni/dcs/__init__.py | 4 ++++ patroni/dcs/consul.py | 9 ++++++--- patroni/dcs/etcd.py | 5 ++++- patroni/dcs/zookeeper.py | 12 +++++++----- patroni/postgresql.py | 4 +++- postgres0.yml | 2 +- tests/test_consul.py | 5 ++++- tests/test_ctl.py | 2 +- tests/test_etcd.py | 2 +- tests/test_ha.py | 5 +++-- tests/test_postgresql.py | 6 +++--- tests/test_zookeeper.py | 5 ++++- 14 files changed, 43 insertions(+), 21 deletions(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index a8107362..9fc15f50 100644 --- a/patroni/__init__.py +++ b/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: diff --git a/patroni/config.py b/patroni/config.py index 2fb24135..4e03b480 100644 --- a/patroni/config.py +++ b/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 diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 209214c0..d4da8e6b 100644 --- a/patroni/dcs/__init__.py +++ b/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 diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index d20be351..4c5a4bc1 100644 --- a/patroni/dcs/consul.py +++ b/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: diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 48afe914..467e1ddd 100644 --- a/patroni/dcs/etcd.py +++ b/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) diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index d4d680ee..29367419 100644 --- a/patroni/dcs/zookeeper.py +++ b/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), - command_retry={'deadline': (config.get('reconnect_timeout') or 10), - 'max_delay': 1, 'max_tries': -1}, + self._client = KazooClient(hosts=hosts, timeout=config['ttl'], + command_retry={'deadline': config['retry_timeout'], '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 - # but there is no other way to change session_timeout without losing session + # I know, it's weird to access private attributes, 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.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) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index fd8a3692..22916086 100644 --- a/patroni/postgresql.py +++ b/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): diff --git a/postgres0.yml b/postgres0.yml index f6c6ca07..e09cb4f0 100644 --- a/postgres0.yml +++ b/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 diff --git a/tests/test_consul.py b/tests/test_consul.py index 2c5bb36d..baf28719 100644 --- a/tests/test_consul.py +++ b/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) diff --git a/tests/test_ctl.py b/tests/test_ctl.py index d8dbecb7..ff491ba6 100644 --- a/tests/test_ctl.py +++ b/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): diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 4d248e11..d675db43 100644 --- a/tests/test_etcd.py +++ b/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): diff --git a/tests/test_ha.py b/tests/test_ha.py index f8f1f90c..ee90eae0 100644 --- a/tests/test_ha.py +++ b/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() diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 38d89ed1..31ec0c60 100644 --- a/tests/test_postgresql.py +++ b/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({'listen': '*:5433', 'parameters': self._PARAMETERS}) + self.p.reload_config({'retry_timeout': 10, 'listen': '*', '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): diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index b2e6e57e..71c9093f 100644 --- a/tests/test_zookeeper.py +++ b/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'))