Varios configuration parameters for etcd (#358)

* Add https and auth support for etcd

Also implement support of PATRONI_ETCD_URL and PATRONI_ETCD_SRV
environment variables

* Implement etcd.proxy etcd.cacert, etcd.cert and etcd.key support

Now it should be possible to set up fully encrypted connection to etcd
with authorization.

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

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

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)
+                    # PATRONI_(ETCD|CONSUL|ZOOKEEPER|EXHIBITOR|...)_(HOSTS?|PORT|..)
-                    if suffix in ('HOST', 'HOSTS', 'PORT') and '_' not in name:
+                    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)

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):
+        for r in ['-client-ssl', '-client', '-ssl', '', '-server-ssl', '-server']:
-            url = '{0}://{1}:{2}/members'.format(self._protocol, host, port)
+            protocol = 'https' if '-ssl' in r else 'http'
-            try:
+            endpoint = '/members' if '-server' in r else ''
-                response = requests.get(url, timeout=self.read_timeout)
+            for host, port in self.get_srv_record('_etcd{0}._tcp.{1}'.format(r, srv)):
-                if response.ok:
+                url = '{0}://{1}:{2}{3}'.format(protocol, host, port, endpoint)
-                    for member in response.json():
+                if endpoint:
-                        ret.extend(member['clientURLs'])
+                    try:
-                    break
+                        response = requests.get(url, timeout=self.read_timeout, verify=False)
-            except RequestException:
+                        if response.ok:
-                logger.exception('GET %s', url)
+                            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:
+        if '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')
+            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:
+            if 'srv' in self._config:
-            self._machines_cache = self._get_machines_cache_from_srv(self._config['discovery_srv'])
+                self._machines_cache = self._get_machines_cache_from_srv(self._config['srv'])
 
-        if not self._machines_cache and 'host' in self._config:
+            if not self._machines_cache and 'host' in self._config:
-            self._machines_cache = self._get_machines_cache_from_dns(self._config['host'])
+                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:

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',

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('_etcd-server._tcp.blabla'), [])
-        self.assertEquals(self.client.get_srv_record('exception'), [])
+        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)