Respect publish host when present on a node.
Closes #251. Thanks danpilch and bowensong for the patch!
This commit is contained in:
+14
-10
@@ -1,4 +1,3 @@
|
||||
import re
|
||||
import time
|
||||
from itertools import chain
|
||||
|
||||
@@ -8,9 +7,6 @@ from .serializer import JSONSerializer, Deserializer, DEFAULT_SERIALIZERS
|
||||
from .exceptions import ConnectionError, TransportError, SerializationError, \
|
||||
ConnectionTimeout, ImproperlyConfigured
|
||||
|
||||
# get ip/port from "127.0.0.1:9200"
|
||||
ADDRESS_RE = re.compile(r'^(?P<host>[\.:0-9a-f]*):(?P<port>[0-9]+)?$')
|
||||
|
||||
|
||||
def get_host_info(node_info, host):
|
||||
"""
|
||||
@@ -218,15 +214,23 @@ class Transport(object):
|
||||
raise
|
||||
|
||||
hosts = []
|
||||
address = self.connection_class.transport_schema + '_address'
|
||||
address_key = self.connection_class.transport_schema + '_address'
|
||||
for n in node_info['nodes'].values():
|
||||
match = ADDRESS_RE.search(n.get(address, ''))
|
||||
if not match:
|
||||
host = {}
|
||||
address = n.get(address_key, '')
|
||||
if '/' in address:
|
||||
host['host'], address = address.split('/', 1)
|
||||
|
||||
# malformed address
|
||||
if ':' not in address:
|
||||
continue
|
||||
|
||||
host = match.groupdict()
|
||||
if 'port' in host:
|
||||
host['port'] = int(host['port'])
|
||||
ip, port = address.rsplit(':', 1)
|
||||
|
||||
# use the ip if not overridden by publish_host
|
||||
host.setdefault('host', ip)
|
||||
host['port'] = int(port)
|
||||
|
||||
host = self.host_info_callback(n, host)
|
||||
if host is not None:
|
||||
hosts.append(host)
|
||||
|
||||
@@ -38,6 +38,26 @@ CLUSTER_NODES = '''{
|
||||
}
|
||||
}'''
|
||||
|
||||
CLUSTER_NODE_PUBLISH_HOST = '''{
|
||||
"ok" : true,
|
||||
"cluster_name" : "super_cluster",
|
||||
"nodes" : {
|
||||
"wE_6OGBNSjGksbONNncIbg" : {
|
||||
"name": "Thunderbird",
|
||||
"transport_address": "obsidian/192.168.1.60:9300",
|
||||
"host": "192.168.1.60",
|
||||
"ip": "192.168.1.60",
|
||||
"version": "2.1.0",
|
||||
"build": "72cd1f1",
|
||||
"http_address": "obsidian/192.168.1.60:9200",
|
||||
"attributes": {
|
||||
"testattr": "test"
|
||||
}
|
||||
}
|
||||
}
|
||||
}'''
|
||||
|
||||
|
||||
class TestHostsInfoCallback(TestCase):
|
||||
def test_master_only_nodes_are_ignored(self):
|
||||
nodes = [
|
||||
@@ -155,6 +175,14 @@ class TestTransport(TestCase):
|
||||
self.assertEquals(1, len(t.connection_pool.connections))
|
||||
self.assertEquals('http://1.1.1.1:123', t.get_connection().host)
|
||||
|
||||
def test_sniff_will_pick_up_published_host(self):
|
||||
t = Transport([{'data': CLUSTER_NODE_PUBLISH_HOST}], connection_class=DummyConnection)
|
||||
t.sniff_hosts()
|
||||
|
||||
self.assertEquals(1, len(t.connection_pool.connections))
|
||||
self.assertEquals('http://obsidian:9200', t.get_connection().host)
|
||||
|
||||
|
||||
def test_sniff_on_start_fetches_and_uses_nodes_list_for_its_schema(self):
|
||||
class DummyThriftConnection(DummyConnection):
|
||||
transport_schema = 'thrift'
|
||||
|
||||
Reference in New Issue
Block a user