From b6425cab85c546dc6148ef99008f4b2043bd4662 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 4 Jan 2018 10:25:06 +0100 Subject: [PATCH] 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 | 1 + docs/SETTINGS.rst | 3 ++- patroni/api.py | 2 +- patroni/dcs/consul.py | 4 ++-- patroni/dcs/etcd.py | 36 ++++++++++++++++++++++++++++-------- patroni/postgresql.py | 6 +++--- patroni/utils.py | 6 ++++++ tests/test_etcd.py | 2 ++ 8 files changed, 45 insertions(+), 15 deletions(-) diff --git a/docs/ENVIRONMENT.rst b/docs/ENVIRONMENT.rst index 50a73fa9..a5a66aa1 100644 --- a/docs/ENVIRONMENT.rst +++ b/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. diff --git a/docs/SETTINGS.rst b/docs/SETTINGS.rst index 35d4a941..d2598bb7 100644 --- a/docs/SETTINGS.rst +++ b/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. diff --git a/patroni/api.py b/patroni/api.py index 8684a82a..807de770 100644 --- a/patroni/api.py +++ b/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) diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index d941a2c0..d1120099 100644 --- a/patroni/dcs/consul.py +++ b/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) diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index bd06f977..940654da 100644 --- a/patroni/dcs/etcd.py +++ b/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) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 0480c4d9..581d4bff 100644 --- a/patroni/postgresql.py +++ b/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): diff --git a/patroni/utils.py b/patroni/utils.py index 89d69797..83ff0c31 100644 --- a/patroni/utils.py +++ b/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]) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 90f523d1..03c4dd5e 100644 --- a/tests/test_etcd.py +++ b/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)