Merge pull request #372 from toddpalino/master

Fully resolve and randomize the client host list at connect time

Author: bbangert

CHANGES.rst

@@ -1,6 +1,19 @@
 Changelog
 =========
 
+2.X.Y (TBD)
+-----------
+
+Features
+********
+
+Bug Handling
+************
+- #372: fully resolve multiple records for hosts in the zookeeper connection string
+
+Documentation
+*************
+
 2.2.1 (2015-06-17)
 ------------------
 

ensure-zookeeper-env.sh

@@ -6,7 +6,7 @@ HERE=`pwd`
 ZOO_BASE_DIR="$HERE/zookeeper"
 ZOOKEEPER_VERSION=${ZOOKEEPER_VERSION:-3.4.6}
 ZOOKEEPER_PATH="$ZOO_BASE_DIR/$ZOOKEEPER_VERSION"
-ZOO_MIRROR_URL="http://apache.osuosl.org/"
+ZOO_MIRROR_URL="http://archive.apache.org/dist"
 
 
 function download_zookeeper(){

kazoo/client.py

@@ -366,10 +366,12 @@ def set_hosts(self, hosts, randomize_hosts=None):
 
         """
 
-        if randomize_hosts is None:
-            randomize_hosts = self.randomize_hosts
+        # Change the client setting for randomization if specified
+        if randomize_hosts is not None:
+            self.randomize_hosts = randomize_hosts
 
-        self.hosts, chroot = collect_hosts(hosts, randomize_hosts)
+        # Randomizing the list will be done at connect time
+        self.hosts, chroot = collect_hosts(hosts)
 
         if chroot:
             new_chroot = normpath(chroot)

kazoo/hosts.py

@@ -1,9 +1,7 @@
-import random
-
 from six.moves import urllib_parse
 
 
-def collect_hosts(hosts, randomize=True):
+def collect_hosts(hosts):
     """Collect a set of hosts and an optional chroot from a string."""
     host_ports, chroot = hosts.partition("/")[::2]
     chroot = "/" + chroot if chroot else None
@@ -19,7 +17,4 @@ def collect_hosts(hosts, randomize=True):
         port = int(res.port) if res.port else 2181
         result.append((host.strip(), port))
 
-    if randomize:
-        random.shuffle(result)
-
     return result, chroot

kazoo/protocol/connection.py

@@ -478,10 +478,31 @@ def zk_loop(self):
             self.client._session_callback(KeeperState.CLOSED)
             self.logger.log(BLATHER, 'Connection stopped')
 
+    def _expand_client_hosts(self):
+        # Expand the entire list in advance so we can randomize it if needed
+        host_ports = []
+        for host, port in self.client.hosts:
+            try:
+                for rhost in socket.getaddrinfo(host.strip(), port, 0, 0, socket.IPPROTO_TCP):
+                    host_ports.append((rhost[4][0], rhost[4][1]))
+            except socket.gaierror as e:
+                # Skip hosts that don't resolve
+                self.logger.warning("Cannot resolve %s: %s", host.strip(), e)
+                pass
+        if self.client.randomize_hosts:
+            random.shuffle(host_ports)
+        return host_ports
+
     def _connect_loop(self, retry):
         # Iterate through the hosts a full cycle before starting over
         status = None
-        for host, port in self.client.hosts:
+        host_ports = self._expand_client_hosts()
+
+        # Check for an empty hostlist, indicating none resolved
+        if len(host_ports) == 0:
+            return STOP_CONNECTING
+
+        for host, port in host_ports:
             if self.client._stopped.is_set():
                 status = STOP_CONNECTING
                 break