diff --git a/patroni/api.py b/patroni/api.py index a0977201..894064f5 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -7,11 +7,12 @@ import traceback import dateutil.parser import datetime import os +import socket from patroni.postgresql import PostgresConnectionException from patroni.postgresql.misc import postgres_version_to_int, PostgresException from patroni.utils import deep_compare, parse_bool, patch_config, Retry, \ - RetryFailedError, parse_int, split_host_port, tzutc + RetryFailedError, parse_int, split_host_port, tzutc, uri from six.moves.BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer from six.moves.socketserver import ThreadingMixIn from threading import Thread @@ -532,8 +533,34 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): def __set_config_parameters(self, config): self.__auth_key = base64.b64encode(config['auth'].encode('utf-8')).decode('utf-8') if 'auth' in config else None - self.connection_string = '{0}://{1}/patroni'.format(self.__protocol, - config.get('connect_address') or self.__listen) + self.connection_string = uri(self.__protocol, config.get('connect_address') or self.__listen, 'patroni') + + @staticmethod + def __has_dual_stack(): + if hasattr(socket, 'AF_INET6') and hasattr(socket, 'IPPROTO_IPV6') and hasattr(socket, 'IPV6_V6ONLY'): + sock = None + try: + sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) + sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, False) + return True + except socket.error as e: + logger.debug('Error when working with ipv6 socket: %s', e) + finally: + if sock: + sock.close() + return False + + def __httpserver_init(self, host, port): + dual_stack = self.__has_dual_stack() + if host == '': + host = None + + info = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM, 0, socket.AI_PASSIVE) + # in case dual stack is not supported we want IPv4 to be preferred over IPv6 + info.sort(key=lambda x: x[0] == socket.AF_INET, reverse=not dual_stack) + + self.address_family = info[0][0] + HTTPServer.__init__(self, info[0][-1][:2], RestApiHandler) def __initialize(self, config): try: @@ -548,7 +575,7 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): self.__listen = config['listen'] self.__ssl_options = self.__get_ssl_options(config) - HTTPServer.__init__(self, (host, port), RestApiHandler) + self.__httpserver_init(host, port) Thread.__init__(self, target=self.serve_forever) self._set_fd_cloexec(self.socket) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 7b34abd2..624e29d4 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -14,7 +14,7 @@ import time from collections import defaultdict, namedtuple from copy import deepcopy from patroni.exceptions import PatroniException -from patroni.utils import parse_bool +from patroni.utils import parse_bool, uri from random import randint from six.moves.urllib_parse import urlparse, urlunparse, parse_qsl from threading import Event, Lock @@ -134,10 +134,7 @@ class Member(namedtuple('Member', 'index,name,session,data')): return conn_url if conn_kwargs: - conn_url = 'postgresql://{host}:{port}'.format( - host=conn_kwargs.get('host'), - port=conn_kwargs.get('port', 5432), - ) + conn_url = uri('postgresql', (conn_kwargs.get('host'), conn_kwargs.get('port', 5432))) self.data['conn_url'] = conn_url return conn_url diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 10d71655..8147ec55 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -11,7 +11,7 @@ import urllib3 from consul import ConsulException, NotFound, base from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory from patroni.exceptions import DCSError -from patroni.utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port +from patroni.utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri from urllib3.exceptions import HTTPError from six.moves.urllib.parse import urlencode, urlparse, quote from six.moves.http_client import HTTPException @@ -40,7 +40,7 @@ class HTTPClient(object): def __init__(self, host='127.0.0.1', port=8500, token=None, scheme='http', verify=True, cert=None, ca_cert=None): self.token = token self._read_timeout = 10 - self.base_uri = '{0}://{1}:{2}'.format(scheme, host, port) + self.base_uri = uri(scheme, (host, port)) kwargs = {} if cert: if isinstance(cert, tuple): diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index dc4a369f..600f0aff 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -14,7 +14,7 @@ from dns.exception import DNSException from dns import resolver from patroni.dcs import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory from patroni.exceptions import DCSError -from patroni.utils import Retry, RetryFailedError, split_host_port +from patroni.utils import Retry, RetryFailedError, split_host_port, uri from urllib3.exceptions import HTTPError, ReadTimeoutError from requests.exceptions import RequestException from six.moves.queue import Queue @@ -25,10 +25,6 @@ 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 @@ -238,7 +234,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 = uri(protocol, host, port, endpoint) + url = uri(protocol, (host, port), endpoint) if endpoint: try: response = requests.get(url, timeout=self.read_timeout, verify=False) @@ -260,19 +256,14 @@ class Client(etcd.Client): 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""" if self.protocol == 'http': - ret = [] - for af, _, _, _, sa in self._dns_resolver.resolve(host, port): - host, port = sa[:2] - if af == socket.AF_INET6: - host = '[{0}]'.format(host) - ret.append(uri(self.protocol, host, port)) + ret = map(lambda res: uri(self.protocol, res[-1][:2]), self._dns_resolver.resolve(host, port)) if ret: return list(set(ret)) - return [uri(self.protocol, host, port)] + 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'])] + return [uri(self.protocol, (self._config['host'], self._config['port']))] machines_cache = [] if 'srv' in self._config: @@ -377,7 +368,7 @@ class Etcd(AbstractDCS): config['hosts'] = [] for value in hosts: if isinstance(value, six.string_types): - config['hosts'].append(uri(protocol, *split_host_port(value, default_port))) + config['hosts'].append(uri(protocol, split_host_port(value, default_port))) elif 'host' in config: host, port = split_host_port(config['host'], 2379) config['host'] = host diff --git a/patroni/dcs/exhibitor.py b/patroni/dcs/exhibitor.py index 216dbabc..16568daf 100644 --- a/patroni/dcs/exhibitor.py +++ b/patroni/dcs/exhibitor.py @@ -4,6 +4,7 @@ import requests import time from patroni.dcs.zookeeper import ZooKeeper +from patroni.utils import uri from requests.exceptions import RequestException logger = logging.getLogger(__name__) @@ -46,9 +47,8 @@ class ExhibitorEnsembleProvider(object): def _query_exhibitors(self, exhibitors): random.shuffle(exhibitors) for host in exhibitors: - uri = 'http://{0}:{1}{2}'.format(host, self._exhibitor_port, self._uri_path) try: - response = requests.get(uri, timeout=self.TIMEOUT) + response = requests.get(uri('http', (host, self._exhibitor_port), self._uri_path), timeout=self.TIMEOUT) return response.json() except RequestException: pass diff --git a/patroni/postgresql/bootstrap.py b/patroni/postgresql/bootstrap.py index 5b79ce36..e9ca8cbd 100644 --- a/patroni/postgresql/bootstrap.py +++ b/patroni/postgresql/bootstrap.py @@ -5,7 +5,7 @@ import tempfile import time from patroni.dcs import RemoteMember -from patroni.utils import deep_compare +from patroni.utils import deep_compare, uri from six import string_types from six.moves.urllib.parse import quote_plus @@ -135,14 +135,14 @@ class Bootstrap(object): r['host'] = 'localhost' # set it to localhost to write into pgpass if 'user' in r: - user = r['user'] + '@' + user = r['user'] else: user = '' if 'password' in r: import getpass r.setdefault('user', os.environ.get('PGUSER', getpass.getuser())) - connstring = 'postgres://{0}{1}:{2}/{3}'.format(user, host, r['port'], r['database']) + connstring = uri('postgres', (host, r['port']), r['database'], user) env = self._postgresql.write_pgpass(r) if 'password' in r else None try: @@ -175,7 +175,7 @@ class Bootstrap(object): if clone_member and clone_member.conn_url: r = clone_member.conn_kwargs(self._postgresql.config.replication) - connstring = 'postgres://{user}@{host}:{port}/{database}'.format(**r) + connstring = uri('postgres', (r['host'], r['port']), r['database'], r['user']) # add the credentials to connect to the replica origin to pgpass. env = self._postgresql.write_pgpass(r) else: diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 45a15c7a..3f88dce5 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -8,7 +8,7 @@ import stat from requests.structures import CaseInsensitiveDict from six.moves.urllib_parse import urlparse, parse_qsl, unquote -from ..utils import compare_values, parse_bool, parse_int, split_host_port +from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri logger = logging.getLogger(__name__) @@ -474,8 +474,8 @@ class ConfigHandler(object): self._local_address = local_address self.local_replication_address = {'host': tcp_local_address, 'port': port} - self._postgresql.connection_string = 'postgres://{0}/{1}'.format( - self._config.get('connect_address') or tcp_local_address + ':' + port, self._postgresql.database) + netloc = self._config.get('connect_address') or tcp_local_address + ':' + port + self._postgresql.connection_string = uri('postgres', netloc, self._postgresql.database) self._postgresql.set_connection_kwargs(self.local_connect_kwargs) diff --git a/patroni/utils.py b/patroni/utils.py index b35ee1aa..6fd48ba0 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -342,3 +342,13 @@ def split_host_port(value, default_port): t = value.rsplit(':', 1) t.append(default_port) return t[0], int(t[1]) + + +def uri(proto, netloc, path='', user=None): + host, port = netloc if isinstance(netloc, (list, tuple)) else split_host_port(netloc, 0) + if host and ':' in host and host[0] != '[' and host[-1] != ']': + host = '[{0}]'.format(host) + port = ':{0}'.format(port) if port else '' + path = '/{0}'.format(path) if path and not path.startswith('/') else path + user = '{0}@'.format(user) if user else '' + return '{0}://{1}{2}{3}{4}'.format(proto, user, host, port, path) diff --git a/tests/test_api.py b/tests/test_api.py index 17758a4c..327dbf8e 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -2,6 +2,7 @@ import datetime import json import psycopg2 import unittest +import socket from mock import Mock, PropertyMock, patch from patroni.api import RestApiHandler, RestApiServer @@ -413,7 +414,8 @@ class TestRestApiServer(unittest.TestCase): srv = MockRestApiServer(lambda a1, a2, a3: None, '') self.assertRaises(ValueError, srv.reload_config, bad_config) self.assertRaises(ValueError, srv.reload_config, {}) - srv.reload_config({'listen': '127.0.0.2:8008'}) + with patch.object(socket.socket, 'setsockopt', Mock(side_effect=socket.error)): + srv.reload_config({'listen': ':8008'}) def test_handle_error(self): try: