diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst index f94ac166..68ec7343 100644 --- a/docs/ENVIRONMENT.rst +++ b/docs/ENVIRONMENT.rst @@ -27,6 +27,12 @@ Consul Etcd ---- - **PATRONI\_ETCD\_HOST**: the host:port for the etcd endpoint. +- **PATRONI\_ETCD\_URL**: url for the etcd, in format: http(s)://(username:password@)host:port +- **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\_SRV**: Domain to search the SRV record(s) for cluster autodiscovery. +- **PATRONI\_ETCD\_CACERT**: The ca certificate. If pressent it will enable validation. +- **PATRONI\_ETCD\_CERT**: File with the client certificate +- **PATRONI\_ETCD\_KEY**: File with the client key. Can be empty if the key is part of certificate. Exhibitor --------- diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index 407d0b37..55c7492c 100644 --- a/docs/SETTINGS.rst +++ b/docs/SETTINGS.rst @@ -41,7 +41,17 @@ Consul Etcd ---- +Most of the parameters are optional, but you have to specify one of the **host**, **url**, **proxy** or **srv** - **host**: the host:port for the etcd endpoint. +- **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. +- **protocol**: (optional) http or https, if not specified http is used. If the **url** or **proxy** is specified - will take protocol from them. +- **username**: (optional) username for etcd authentication +- **password**: (optional) password for etcd authentication. +- **cacert**: (optional) The ca certificate. If pressent it will enable validation. +- **cert**: (optional) file with the client certificate +- **key**: (optional) file with the client key. Can be empty if the key is part of **cert**. Exhibitor --------- diff --git a/patroni/config.py b/patroni/config.py index c7b77536..db7a75f7 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -236,8 +236,9 @@ class Config(object): if param.startswith(Config.PATRONI_ENV_PREFIX): name, suffix = (param[8:].rsplit('_', 1) + [''])[:2] if name and suffix: - # PATRONI_(ETCD|CONSUL|ZOOKEEPER|EXHIBITOR|...)_(HOSTS?|PORT) - if suffix in ('HOST', 'HOSTS', 'PORT') and '_' not in name: + # PATRONI_(ETCD|CONSUL|ZOOKEEPER|EXHIBITOR|...)_(HOSTS?|PORT|..) + if suffix in ('HOST', 'HOSTS', 'PORT', 'SRV', 'URL', 'PROXY', 'CACERT', 'CERT', 'KEY') \ + and '_' not in name: value = os.environ.pop(param) if suffix == 'PORT': value = value and parse_int(value) diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 14dae7e4..1d09624d 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -15,6 +15,7 @@ from patroni.utils import Retry, RetryFailedError from urllib3.exceptions import HTTPError, ReadTimeoutError from requests.exceptions import RequestException from six.moves.http_client import HTTPException +from six.moves.urllib_parse import urlparse logger = logging.getLogger(__name__) @@ -26,10 +27,12 @@ class EtcdError(DCSError): class Client(etcd.Client): def __init__(self, config): - super(Client, self).__init__(read_timeout=config['retry_timeout']) + args = {p: config.get(p) for p in ('host', 'port', 'protocol', 'use_proxies', 'username', 'password', + 'cert', 'ca_cert') if config.get(p)} + super(Client, self).__init__(read_timeout=config['retry_timeout'], **args) self._config = config self._load_machines_cache() - self._allow_reconnect = True + self._allow_reconnect = not self._use_proxies def _build_request_parameters(self): kwargs = {'headers': self._get_headers(), 'redirect': self.allow_redirect} @@ -147,40 +150,48 @@ class Client(etcd.Client): @staticmethod def get_srv_record(host): try: - return [(str(r.target).rstrip('.'), r.port) for r in resolver.query('_etcd-server._tcp.' + host, 'SRV')] + return [(r.target.to_text(True), r.port) for r in resolver.query(host, 'SRV')] except DNSException: logger.exception('Can not resolve SRV for %s', host) return [] - def _get_machines_cache_from_srv(self, discovery_srv): + def _get_machines_cache_from_srv(self, srv): """Fetch list of etcd-cluster member by resolving _etcd-server._tcp. SRV record. This record should contain list of host and peer ports which could be used to run 'GET http://{host}:{port}/members' request (peer protocol)""" ret = [] - for host, port in self.get_srv_record(discovery_srv): - url = '{0}://{1}:{2}/members'.format(self._protocol, host, port) - try: - response = requests.get(url, timeout=self.read_timeout) - if response.ok: - for member in response.json(): - ret.extend(member['clientURLs']) - break - except RequestException: - logger.exception('GET %s', url) + for r in ['-client-ssl', '-client', '-ssl', '', '-server-ssl', '-server']: + protocol = 'https' if '-ssl' in r else 'http' + endpoint = '/members' if '-server' in r else '' + for host, port in self.get_srv_record('_etcd{0}._tcp.{1}'.format(r, srv)): + url = '{0}://{1}:{2}{3}'.format(protocol, host, port, endpoint) + if endpoint: + try: + response = requests.get(url, timeout=self.read_timeout, verify=False) + if response.ok: + for member in response.json(): + ret.extend(member['clientURLs']) + break + except RequestException: + logger.exception('GET %s', url) + else: + ret.append(url) + if ret: + self._protocol = protocol + break return list(set(ret)) - def _get_machines_cache_from_dns(self, addr): + def _get_machines_cache_from_dns(self, host, port): """One host might be resolved into multiple ip addresses. We will make list out of it""" ret = [] - host, port = addr.split(':') try: for r in set(socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)): - ret.append('{0}://{1}:{2}'.format(self._protocol, r[4][0], r[4][1])) + ret.append('{0}://{1}:{2}'.format(self.protocol, *r[4])) except socket.error: logger.exception('Can not resolve %s', host) - return list(set(ret)) if ret else ['{0}://{1}:{2}'.format(self._protocol, host, port)] + return list(set(ret)) if ret else ['{0}://{1}:{2}'.format(self.protocol, host, port)] def _load_machines_cache(self): """This method should fill up `_machines_cache` from scratch. @@ -190,16 +201,19 @@ class Client(etcd.Client): self._update_machines_cache = True - if 'discovery_srv' not in self._config and 'host' not in self._config: - raise Exception('Neither discovery_srv nor host are defined in etcd section of config') + if 'srv' not in self._config and 'host' not in self._config: + raise Exception('Neither srv nor host url are defined in etcd section of config') - self._machines_cache = [] + if self._use_proxies: + self._machines_cache = ['{0}://{1}:{2}'.format(self.protocol, self._config['host'], self._config['port'])] + else: + self._machines_cache = [] - if 'discovery_srv' in self._config: - self._machines_cache = self._get_machines_cache_from_srv(self._config['discovery_srv']) + if 'srv' in self._config: + self._machines_cache = self._get_machines_cache_from_srv(self._config['srv']) - if not self._machines_cache and 'host' in self._config: - self._machines_cache = self._get_machines_cache_from_dns(self._config['host']) + 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']) # Can not bootstrap list of etcd-cluster members, giving up if not self._machines_cache: @@ -245,6 +259,29 @@ class Etcd(AbstractDCS): @staticmethod def get_etcd_client(config): + if 'proxy' in config: + config['use_proxies'] = True + config['url'] = config['proxy'] + + if 'url' in config: + r = urlparse(config['url']) + config.update({'protocol': r.scheme, 'host': r.hostname, 'port': r.port or 2379, + 'username': r.username, 'password': r.password}) + elif 'host' in config: + host, port = (config['host'] + ':2379').split(':')[:2] + config['host'] = host + if 'port' not in config: + config['port'] = int(port) + + if config.get('cacert'): + config['ca_cert'] = config.pop('cacert') + + if config.get('key') and config.get('cert'): + config['cert'] = (config['cert'], config['key']) + + for p in ('discovery_srv', 'srv_domain'): + if p in config: + config['srv'] = config.pop(p) client = None while not client: try: diff --git a/tests/test_config.py b/tests/test_config.py index 94e5f5fc..f06d1eb0 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -40,6 +40,12 @@ class TestConfig(unittest.TestCase): 'PATRONI_POSTGRESQL_DATA_DIR': 'data/postgres0', 'PATRONI_POSTGRESQL_PGPASS': '/tmp/pgpass0', 'PATRONI_ETCD_HOST': '127.0.0.1:2379', + 'PATRONI_ETCD_URL': 'https://127.0.0.1:2379', + 'PATRONI_ETCD_PROXY': 'http://127.0.0.1:2379', + 'PATRONI_ETCD_SRV': 'test', + 'PATRONI_ETCD_CACERT': '/cacert', + 'PATRONI_ETCD_CERT': '/cert', + 'PATRONI_ETCD_KEY': '/key', 'PATRONI_CONSUL_HOST': '127.0.0.1:8500', 'PATRONI_ZOOKEEPER_HOSTS': "'host1:2181','host2:2181'", 'PATRONI_EXHIBITOR_HOSTS': 'host1,host2', diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 3f38bc86..6742f66a 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -114,17 +114,17 @@ class SleepException(Exception): pass -class MockSRV(object): - port = 2380 - target = '127.0.0.1' - - def dns_query(name, _): + if '-server' not in name or '-ssl' in name: + return [] if name == '_etcd-server._tcp.blabla': return [] elif name == '_etcd-server._tcp.exception': raise DNSException() - return [MockSRV()] + srv = Mock() + srv.port = 2380 + srv.target.to_text.return_value = 'localhost' if name == '_etcd-server._tcp.foobar' else '127.0.0.1' + return [srv] def socket_getaddrinfo(*args): @@ -155,7 +155,7 @@ class TestClient(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.client = Client({'discovery_srv': 'test', 'retry_timeout': 3}) + self.client = Client({'srv': 'test', 'retry_timeout': 3}) self.client.http.request = http_request self.client.http.request_encode_body = http_request @@ -188,22 +188,23 @@ class TestClient(unittest.TestCase): self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', 'GET') def test_get_srv_record(self): - self.assertEquals(self.client.get_srv_record('blabla'), []) - self.assertEquals(self.client.get_srv_record('exception'), []) + self.assertEquals(self.client.get_srv_record('_etcd-server._tcp.blabla'), []) + self.assertEquals(self.client.get_srv_record('_etcd-server._tcp.exception'), []) def test__get_machines_cache_from_srv(self): + self.client._get_machines_cache_from_srv('foobar') self.client.get_srv_record = Mock(return_value=[('localhost', 2380)]) self.client._get_machines_cache_from_srv('blabla') def test__get_machines_cache_from_dns(self): - self.client._get_machines_cache_from_dns('error:2379') + self.client._get_machines_cache_from_dns('error', 2379) @patch.object(Client, 'machines') def test__load_machines_cache(self, mock_machines): mock_machines.__get__ = Mock(return_value=['http://localhost:2379']) self.client._config = {} self.assertRaises(Exception, self.client._load_machines_cache) - self.client._config = {'discovery_srv': 'blabla'} + self.client._config = {'srv': 'blabla'} self.assertRaises(etcd.EtcdException, self.client._load_machines_cache) @@ -228,7 +229,11 @@ class TestEtcd(unittest.TestCase): 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}) + {'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}) def test_get_cluster(self): self.assertIsInstance(self.etcd.get_cluster(), Cluster)