Allow to specify multiple hosts for etcd (#589)

This list will be used for initial discovery of etcd cluster members.
If for some reason during work this list of hosts has been exhausted (during work), Patroni will return to initial list.

In addition to that improve ipv6 compatibility by using a special function for splitting host and port.

Fixes https://github.com/zalando/patroni/issues/523

docs/ENVIRONMENT.rst

@@ -39,6 +39,7 @@ Consul
 Etcd
 ----
 -  **PATRONI\_ETCD\_HOST**: the host:port for the etcd endpoint.
+-  **PATRONI\_ETCD\_HOSTS**: list of etcd endpoints in format host1:port1,host2:port2,etc...
 -  **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.

docs/SETTINGS.rst

@@ -63,9 +63,10 @@ Most of the parameters are optional, but you have to specify one of the **host**
 
 Etcd
 ----
-Most of the parameters are optional, but you have to specify one of the **host**, **url**, **proxy** or **srv**
+Most of the parameters are optional, but you have to specify one of the **host**, **hosts**, **url**, **proxy** or **srv**
 
 -  **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.
 -  **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/api.py

@@ -483,7 +483,7 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread):
     def __initialize(self, config):
         self.__ssl_options = self.__get_ssl_options(config)
         self.__listen = config['listen']
-        host, port = config['listen'].split(':')
+        host, port = config['listen'].rsplit(':', 1)
         HTTPServer.__init__(self, (host, int(port)), RestApiHandler)
         Thread.__init__(self, target=self.serve_forever)
         self._set_fd_cloexec(self.socket)

patroni/dcs/consul.py

@@ -10,7 +10,7 @@ import urllib3
 from consul import ConsulException, NotFound, base
 from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState
 from patroni.exceptions import DCSError
-from patroni.utils import deep_compare, parse_bool, Retry, RetryFailedError
+from patroni.utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port
 from urllib3.exceptions import HTTPError
 from six.moves.urllib.parse import urlencode, urlparse
 from six.moves.http_client import HTTPException
@@ -148,7 +148,7 @@ class Consul(AbstractDCS):
             r = urlparse(config['url'])
             config.update({'scheme': r.scheme, 'host': r.hostname, 'port': r.port or 8500})
         elif 'host' in config:
-            host, port = (config.get('host', '127.0.0.1:8500') + ':8500').split(':')[:2]
+            host, port = split_host_port(config.get('host', '127.0.0.1:8500'), 8500)
             config['host'] = host
             if 'port' not in config:
                 config['port'] = int(port)

patroni/dcs/etcd.py

@@ -6,6 +6,7 @@ import os
 import urllib3.util.connection
 import random
 import requests
+import six
 import socket
 import time
 
@@ -13,7 +14,7 @@ from dns.exception import DNSException
 from dns import resolver
 from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState
 from patroni.exceptions import DCSError
-from patroni.utils import Retry, RetryFailedError
+from patroni.utils import Retry, RetryFailedError, split_host_port
 from urllib3.exceptions import HTTPError, ReadTimeoutError
 from requests.exceptions import RequestException
 from six.moves.queue import Queue
@@ -24,6 +25,10 @@ from threading import Thread
 logger = logging.getLogger(__name__)
 
 
+def uri(protocol, host, port, endpoint=''):
+    return '{0}://{1}:{2}{3}'.format(protocol, host, port, endpoint)
+
+
 class EtcdError(DCSError):
     pass
 
@@ -228,7 +233,7 @@ class Client(etcd.Client):
             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)
+                url = uri(protocol, host, port, endpoint)
                 if endpoint:
                     try:
                         response = requests.get(url, timeout=self.read_timeout, verify=False)
@@ -255,10 +260,10 @@ class Client(etcd.Client):
                 host, port = sa[:2]
                 if af == socket.AF_INET6:
                     host = '[{0}]'.format(host)
-                ret.append('{0}://{1}:{2}'.format(self.protocol, host, port))
+                ret.append(uri(self.protocol, host, port))
             if ret:
                 return list(set(ret))
-        return ['{0}://{1}:{2}'.format(self.protocol, host, port)]
+        return [uri(self.protocol, host, port)]
 
     def _load_machines_cache(self):
         """This method should fill up `_machines_cache` from scratch.
@@ -268,17 +273,20 @@ class Client(etcd.Client):
 
         self._update_machines_cache = True
 
-        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')
+        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 = ['{0}://{1}:{2}'.format(self.protocol, self._config['host'], self._config['port'])]
+            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'])
 
@@ -348,8 +356,20 @@ class Etcd(AbstractDCS):
             r = urlparse(config['url'])
             config.update({'protocol': r.scheme, 'host': r.hostname, 'port': r.port or 2379,
                            'username': r.username, 'password': r.password})
+        elif 'hosts' in config:
+            hosts = config.pop('hosts')
+            default_port = config.pop('port', 2379)
+            protocol = config.get('protocol', 'http')
+
+            if isinstance(hosts, six.string_types):
+                hosts = hosts.split(',')
+
+            config['hosts'] = []
+            for value in hosts:
+                if isinstance(value, six.string_types):
+                    config['hosts'].append(uri(protocol, *split_host_port(value, default_port)))
         elif 'host' in config:
-            host, port = (config['host'] + ':2379').split(':')[:2]
+            host, port = split_host_port(config['host'], 2379)
             config['host'] = host
             if 'port' not in config:
                 config['port'] = int(port)

patroni/postgresql.py

@@ -13,7 +13,7 @@ from collections import defaultdict
 from contextlib import contextmanager
 from patroni.callback_executor import CallbackExecutor
 from patroni.exceptions import PostgresConnectionException, PostgresException
-from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError, polling_loop
+from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError, polling_loop, split_host_port
 from patroni.postmaster import PostmasterProcess
 from six import string_types
 from six.moves.urllib.parse import quote_plus
@@ -215,8 +215,8 @@ class Postgresql(object):
 
     def get_server_parameters(self, config):
         parameters = config['parameters'].copy()
-        listen_addresses, port = (config['listen'] + ':5432').split(':')[:2]
-        parameters.update({'cluster_name': self.scope, 'listen_addresses': listen_addresses, 'port': port})
+        listen_addresses, port = split_host_port(config['listen'], 5432)
+        parameters.update({'cluster_name': self.scope, 'listen_addresses': listen_addresses, 'port': str(port)})
         if config.get('synchronous_mode', False):
             if self._synchronous_standby_names is None:
                 if config.get('synchronous_mode_strict', False):

patroni/utils.py

@@ -275,3 +275,9 @@ def polling_loop(timeout, interval=1):
         yield iteration
         iteration += 1
         time.sleep(interval)
+
+
+def split_host_port(value, default_port):
+    t = value.rsplit(':', 1)
+    t.append(default_port)
+    return t[0], int(t[1])

tests/test_etcd.py

@@ -262,6 +262,8 @@ class TestEtcd(unittest.TestCase):
                                   {'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})
 
     def test_get_cluster(self):
         self.assertIsInstance(self.etcd.get_cluster(), Cluster)