gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Better support for static etcd cluster (#986)

Browse Source 
if the `etcd.use_proxies` is set to true, Patroni will stick to the list of hosts specified in the `etcd.hosts` and avoid doing topology discovery. Such mode might be useful when you know that you connect to the etcd cluster via the set of proxies or when th etcd cluster has static topology.
This commit is contained in:
Alexander Kukushkin
2019-03-07 11:36:36 +01:00
committed by GitHub
parent f0990532dc
commit c64d51f79c
5 changed files with 60 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/ENVIRONMENT.rst
+1
View File
@@ -50,6 +50,7 @@ Etcd
- **PATRONI\_ETCD\_PROXY**: proxy url for the etcd. If you are connecting to the etcd using proxy, use this parameter instead of **PATRONI\_ETCD\_URL**
- **PATRONI\_ETCD\_URL**: url for the etcd, in format: http(s)://(username:password@)host:port
- **PATRONI\_ETCD\_HOSTS**: list of etcd endpoints in format 'host1:port1','host2:port2',etc...
- **PATRONI\_ETCD\_USE\_PROXIES**: If this parameter is set to true, Patroni will consider **hosts** as a list of proxies and will not perform a topology discovery of etcd cluster but stick to a fixed list of **hosts**.
- **PATRONI\_ETCD\_PROTOCOL**: http or https, if not specified http is used. If the **url** or **proxy** is specified - will take protocol from them.
- **PATRONI\_ETCD\_HOST**: the host:port for the etcd endpoint.
- **PATRONI\_ETCD\_SRV**: Domain to search the SRV record(s) for cluster autodiscovery.
docs/SETTINGS.rst
+3 -2
View File
@@ -50,8 +50,8 @@ Bootstrap configuration
- **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Patroni will try to create slots before opening connections to the cluster.
- **my_slot_name**: the name of replication slot. It is the responsibility of the operator to make sure that there are no clashes in names between replication slots automatically created by Patroni for members and permanent replication slots.
- **type**: slot type. Could be ``physical`` or ``logical``. If the slot is logical, you have to additionally define ``database`` and ``plugin``.
**database**: the database name where logical slots should be created.
**plugin**: the plugin name for the logical slot.
- **database**: the database name where logical slots should be created.
- **plugin**: the plugin name for the logical slot.
- **method**: custom script to use for bootstrapping this cluster.
See :ref:`custom bootstrap methods documentation <custom_bootstrap>` for details.
When ``initdb`` is specified revert to the default ``initdb`` command. ``initdb`` is also triggered when no ``method``
@@ -97,6 +97,7 @@ Most of the parameters are optional, but you have to specify one of the **host**
- **host**: the host:port for the etcd endpoint.
- **hosts**: list of etcd endpoint in format host1:port1,host2:port2,etc... Could be a comma separated string or an actual yaml list.
- **use\_proxies**: If this parameter is set to true, Patroni will consider **hosts** as a list of proxies and will not perform a topology discovery of etcd cluster.
- **url**: url for the etcd
- **proxy**: proxy url for the etcd. If you are connecting to the etcd using proxy, use this parameter instead of **url**
- **srv**: Domain to search the SRV record(s) for cluster autodiscovery.
patroni/config.py
+6 -6
View File
@@ -233,7 +233,7 @@ class Config(object):
ret[section][param] = value
_set_section_values('restapi', ['listen', 'connect_address', 'certfile', 'keyfile'])
_set_section_values('postgresql', ['listen', 'connect_address', 'data_dir', 'pgpass', 'bin_dir'])
_set_section_values('postgresql', ['listen', 'connect_address', 'config_dir', 'data_dir', 'pgpass', 'bin_dir'])
_set_section_values('log', ['level', 'format', 'dateformat', 'dir', 'file_size', 'file_num', 'loggers'])
def _parse_dict(value):
@@ -285,10 +285,10 @@ class Config(object):
if param.startswith(Config.PATRONI_ENV_PREFIX):
# PATRONI_(ETCD|CONSUL|ZOOKEEPER|EXHIBITOR|...)_(HOSTS?|PORT|..)
name, suffix = (param[8:].split('_', 1) + [''])[:2]
if suffix in ('HOST', 'HOSTS', 'PORT', 'PROTOCOL', 'SRV', 'URL', 'PROXY', 'CACERT', 'CERT', 'KEY',
'VERIFY', 'TOKEN', 'CHECKS', 'DC', 'REGISTER_SERVICE', 'SERVICE_CHECK_INTERVAL',
'NAMESPACE', 'CONTEXT', 'USE_ENDPOINTS', 'SCOPE_LABEL', 'ROLE_LABEL', 'POD_IP',
'PORTS', 'LABELS') and name:
if suffix in ('HOST', 'HOSTS', 'PORT', 'USE_PROXIES', 'PROTOCOL', 'SRV', 'URL', 'PROXY',
'CACERT', 'CERT', 'KEY', 'VERIFY', 'TOKEN', 'CHECKS', 'DC', 'REGISTER_SERVICE',
'SERVICE_CHECK_INTERVAL', 'NAMESPACE', 'CONTEXT', 'USE_ENDPOINTS', 'SCOPE_LABEL',
'ROLE_LABEL', 'POD_IP', 'PORTS', 'LABELS') and name:
value = os.environ.pop(param)
if suffix == 'PORT':
value = value and parse_int(value)
@@ -296,7 +296,7 @@ class Config(object):
value = value and _parse_list(value)
elif suffix == 'LABELS':
value = _parse_dict(value)
elif suffix == 'REGISTER_SERVICE':
elif suffix in ('USE_PROXIES', 'REGISTER_SERVICE'):
value = parse_bool(value)
if value:
ret[name.lower()][suffix.lower()] = value
patroni/dcs/etcd.py
+33 -27
View File
@@ -89,7 +89,7 @@ class Client(etcd.Client):
super(Client, self).__init__(read_timeout=config['retry_timeout'], **args)
self._config = config
self._load_machines_cache()
self._allow_reconnect = not self._use_proxies
self._allow_reconnect = True
def _build_request_parameters(self):
kwargs = {'headers': self._get_headers(), 'redirect': self.allow_redirect}
@@ -128,8 +128,11 @@ class Client(etcd.Client):
while True:
try:
response = self.http.request(self._MGET, self._base_uri + self.version_prefix + '/machines', **kwargs)
machines = [n.strip() for n in self._handle_server_response(response).data.decode('utf-8').split(',')]
data = self._handle_server_response(response).data.decode('utf-8')
machines = [m.strip() for m in data.split(',') if m.strip()]
logger.debug("Retrieved list of machines: %s", machines)
if not machines:
raise etcd.EtcdException
random.shuffle(machines)
for url in machines:
r = urlparse(url)
@@ -186,10 +189,8 @@ class Client(etcd.Client):
# Update machines_cache if previous attempt of update has failed
if self._update_machines_cache:
self._load_machines_cache()
elif time.time() - self._machines_cache_updated > self._machines_cache_ttl:
self._machines_cache = self.machines
if self._base_uri in self._machines_cache:
self._machines_cache.remove(self._base_uri)
elif not self._use_proxies and time.time() - self._machines_cache_updated > self._machines_cache_ttl:
self._refresh_machines_cache()
self._machines_cache_updated = time.time()
kwargs.update(self._build_request_parameters())
@@ -206,10 +207,8 @@ class Client(etcd.Client):
if response is False:
some_request_failed = True
if some_request_failed and not self._use_proxies:
self._machines_cache = self.machines
if self._base_uri in self._machines_cache:
self._machines_cache.remove(self._base_uri)
if some_request_failed:
self._refresh_machines_cache()
except etcd.EtcdConnectionFailed as e:
if isinstance(e, etcd.EtcdWatchTimedOut) and self._machines_cache:
self._base_uri = self._next_server()
@@ -268,6 +267,21 @@ class Client(etcd.Client):
return list(set(ret))
return [uri(self.protocol, host, port)]
def _get_machines_cache_from_config(self):
if 'proxy' in self._config:
return [uri(self.protocol, self._config['host'], self._config['port'])]
machines_cache = []
if 'srv' in self._config:
machines_cache = self._get_machines_cache_from_srv(self._config['srv'])
if not machines_cache and 'hosts' in self._config:
machines_cache = list(self._config['hosts'])
if not machines_cache and 'host' in self._config:
machines_cache = self._get_machines_cache_from_dns(self._config['host'], self._config['port'])
return machines_cache
def _load_machines_cache(self):
"""This method should fill up `_machines_cache` from scratch.
It could happen only in two cases:
@@ -279,19 +293,7 @@ class Client(etcd.Client):
if 'srv' not in self._config and 'host' not in self._config and 'hosts' not in self._config:
raise Exception('Neither srv, hosts, host nor url are defined in etcd section of config')
if self._use_proxies:
self._machines_cache = [uri(self.protocol, self._config['host'], self._config['port'])]
else:
self._machines_cache = []
if 'srv' in self._config:
self._machines_cache = self._get_machines_cache_from_srv(self._config['srv'])
if not self._machines_cache and 'hosts' in self._config:
self._machines_cache = list(self._config['hosts'])
if not self._machines_cache and 'host' in self._config:
self._machines_cache = self._get_machines_cache_from_dns(self._config['host'], self._config['port'])
self._machines_cache = self._get_machines_cache_from_config()
# Can not bootstrap list of etcd-cluster members, giving up
if not self._machines_cache:
@@ -299,14 +301,16 @@ class Client(etcd.Client):
# After filling up initial list of machines_cache we should ask etcd-cluster about actual list
self._base_uri = self._next_server()
self._machines_cache = self.machines
if self._base_uri in self._machines_cache:
self._machines_cache.remove(self._base_uri)
self._refresh_machines_cache()
self._update_machines_cache = False
self._machines_cache_updated = time.time()
def _refresh_machines_cache(self):
self._machines_cache = self._get_machines_cache_from_config() if self._use_proxies else self.machines
if self._base_uri in self._machines_cache:
self._machines_cache.remove(self._base_uri)
class Etcd(AbstractDCS):
@@ -426,6 +430,8 @@ class Etcd(AbstractDCS):
while not client:
try:
client = Client(config, dns_resolver)
if 'use_proxies' in config and not client.machines:
raise etcd.EtcdException
except etcd.EtcdException:
logger.info('waiting on etcd')
time.sleep(5)
tests/test_etcd.py
+17 -15
View File
@@ -148,13 +148,14 @@ def socket_getaddrinfo(*args):
def http_request(method, url, **kwargs):
if url == 'http://localhost:2379/timeout':
raise ReadTimeoutError(None, None, None)
ret = MockResponse()
if url == 'http://localhost:2379/v2/machines':
ret = MockResponse()
ret.content = 'http://localhost:2379,http://localhost:4001'
return ret
if url == 'http://localhost:2379/':
return MockResponse()
raise socket.error
elif url == 'http://localhost:4001/v2/machines':
ret.content = ''
elif url != 'http://localhost:2379/':
raise socket.error
return ret
class TestDnsCachingResolver(unittest.TestCase):
@@ -263,17 +264,18 @@ class TestEtcd(unittest.TestCase):
@patch('dns.resolver.query', dns_query)
def test_get_etcd_client(self):
with patch.object(Client, 'machines') as mock_machines:
with patch('time.sleep', Mock(side_effect=SleepException)),\
patch.object(Client, 'machines') as mock_machines:
mock_machines.__get__ = Mock(side_effect=etcd.EtcdException)
with patch('time.sleep', Mock(side_effect=SleepException)):
self.assertRaises(SleepException, self.etcd.get_etcd_client,
{'discovery_srv': 'test', 'retry_timeout': 10, 'cacert': '1', 'key': '1', 'cert': 1})
self.assertRaises(SleepException, self.etcd.get_etcd_client,
{'url': 'https://test:2379', 'retry_timeout': 10})
self.assertRaises(SleepException, self.etcd.get_etcd_client,
{'proxy': 'https://user:password@test:2379', 'retry_timeout': 10})
self.assertRaises(SleepException, self.etcd.get_etcd_client,
{'hosts': 'foo:4001,bar', 'retry_timeout': 10})
self.assertRaises(SleepException, self.etcd.get_etcd_client,
{'discovery_srv': 'test', 'retry_timeout': 10, 'cacert': '1', 'key': '1', 'cert': 1})
self.assertRaises(SleepException, self.etcd.get_etcd_client,
{'url': 'https://test:2379', 'retry_timeout': 10})
self.assertRaises(SleepException, self.etcd.get_etcd_client,
{'hosts': 'foo:4001,bar', 'retry_timeout': 10})
mock_machines.__get__ = Mock(return_value=[])
self.assertRaises(SleepException, self.etcd.get_etcd_client,
{'proxy': 'https://user:password@test:2379', 'retry_timeout': 10})
def test_get_cluster(self):
cluster = self.etcd.get_cluster()