IPv6 support (#1122)

fixes https://github.com/zalando/patroni/issues/1121

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)
 

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
 

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

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

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

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:

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)
 

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)

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: