metadata: save hosts based on host_id insted of endpoint

also keep mapping between endpoints to host_ids, so we control
connection can still working with the inital endpoint,
while the hosts list only known nodes with thier host_id

Author: fruch

cassandra/cluster.py

@@ -2130,15 +2130,15 @@ def signal_connection_failure(self, host, connection_exc, is_host_addition, expe
             self.on_down(host, is_host_addition, expect_host_to_be_down)
         return is_down
 
-    def add_host(self, endpoint, datacenter=None, rack=None, signal=True, refresh_nodes=True):
+    def add_host(self, endpoint, datacenter=None, rack=None, signal=True, refresh_nodes=True, host_id=None):
         """
         Called when adding initial contact points and when the control
         connection subsequently discovers a new node.
         Returns a Host instance, and a flag indicating whether it was new in
         the metadata.
         Intended for internal use only.
         """
-        host, new = self.metadata.add_or_return_host(Host(endpoint, self.conviction_policy_factory, datacenter, rack))
+        host, new = self.metadata.add_or_return_host(Host(endpoint, self.conviction_policy_factory, datacenter, rack, host_id=host_id))
         if new and signal:
             log.info("New Cassandra host %r discovered", host)
             self.on_add(host, refresh_nodes)
@@ -3817,9 +3817,8 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
         partitioner = None
         token_map = {}
 
-        found_hosts = set()
+        found_host_ids = set()
         if local_result.parsed_rows:
-            found_hosts.add(connection.endpoint)
             local_rows = dict_factory(local_result.column_names, local_result.parsed_rows)
             local_row = local_rows[0]
             cluster_name = local_row["cluster_name"]
@@ -3833,7 +3832,9 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
                 datacenter = local_row.get("data_center")
                 rack = local_row.get("rack")
                 self._update_location_info(host, datacenter, rack)
+                host.endpoint = self._cluster.endpoint_factory.create(local_row)
                 host.host_id = local_row.get("host_id")
+                found_host_ids.add(host.host_id)
                 host.listen_address = local_row.get("listen_address")
                 host.listen_port = local_row.get("listen_port")
                 host.broadcast_address = _NodeInfo.get_broadcast_address(local_row)
@@ -3872,6 +3873,8 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
                 if partitioner and tokens:
                     token_map[host] = tokens
 
+                self._cluster.metadata.update_host(host, old_endpoint=connection.endpoint)
+                connection.original_endpoint = connection.endpoint = host.endpoint
         # Check metadata.partitioner to see if we haven't built anything yet. If
         # every node in the cluster was in the contact points, we won't discover
         # any new nodes, so we need this additional check.  (See PYTHON-90)
@@ -3884,24 +3887,26 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
                 continue
 
             endpoint = self._cluster.endpoint_factory.create(row)
+            host_id = row.get("host_id")
 
-            if endpoint in found_hosts:
-                log.warning("Found multiple hosts with the same endpoint (%s). Excluding peer %s", endpoint, row.get("peer"))
+            if host_id in found_host_ids:
+                log.warning("Found multiple hosts with the same host_id (%s). Excluding peer %s", host_id, row.get("peer"))
                 continue
 
-            found_hosts.add(endpoint)
+            found_host_ids.add(host_id)
 
             host = self._cluster.metadata.get_host(endpoint)
             datacenter = row.get("data_center")
             rack = row.get("rack")
+
             if host is None:
                 log.debug("[control connection] Found new host to connect to: %s", endpoint)
-                host, _ = self._cluster.add_host(endpoint, datacenter, rack, signal=True, refresh_nodes=False)
+                host, _ = self._cluster.add_host(endpoint, datacenter=datacenter, rack=rack, signal=True, refresh_nodes=False, host_id=host_id)
                 should_rebuild_token_map = True
             else:
                 should_rebuild_token_map |= self._update_location_info(host, datacenter, rack)
 
-            host.host_id = row.get("host_id")
+            host.host_id = host_id
             host.broadcast_address = _NodeInfo.get_broadcast_address(row)
             host.broadcast_port = _NodeInfo.get_broadcast_port(row)
             host.broadcast_rpc_address = _NodeInfo.get_broadcast_rpc_address(row)
@@ -3915,11 +3920,11 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
             if partitioner and tokens and self._token_meta_enabled:
                 token_map[host] = tokens
 
-        for old_host in self._cluster.metadata.all_hosts():
-            if old_host.endpoint.address != connection.endpoint and old_host.endpoint not in found_hosts:
+        for old_host_id, old_host in self._cluster.metadata.all_hosts_items():
+            if old_host_id not in found_host_ids:
                 should_rebuild_token_map = True
                 log.debug("[control connection] Removing host not found in peers metadata: %r", old_host)
-                self._cluster.remove_host(old_host)
+                self._cluster.metadata.remove_host_by_host_id(old_host_id)
 
         log.debug("[control connection] Finished fetching ring info")
         if partitioner and should_rebuild_token_map:

cassandra/metadata.py

@@ -124,6 +124,7 @@ def __init__(self):
         self.keyspaces = {}
         self.dbaas = False
         self._hosts = {}
+        self._host_id_by_endpoint = {}
         self._hosts_lock = RLock()
 
     def export_schema_as_string(self):
@@ -330,25 +331,39 @@ def add_or_return_host(self, host):
         """
         with self._hosts_lock:
             try:
-                return self._hosts[host.endpoint], False
+                return self._hosts[host.host_id], False
             except KeyError:
-                self._hosts[host.endpoint] = host
+                self._host_id_by_endpoint[host.endpoint] = host.host_id
+                self._hosts[host.host_id] = host
                 return host, True
 
     def remove_host(self, host):
         with self._hosts_lock:
-            return bool(self._hosts.pop(host.endpoint, False))
+            self._host_id_by_endpoint.pop(host.endpoint, False)
+            return bool(self._hosts.pop(host.host_id, False))
+
+    def remove_host_by_host_id(self, host_id):
+        with self._hosts_lock:
+            return bool(self._hosts.pop(host_id, False))
+
+    def update_host(self, host, old_endpoint):
+        host, created = self.add_or_return_host(host)
+        with self._hosts_lock:
+            self._host_id_by_endpoint.pop(old_endpoint, False)
+            self._host_id_by_endpoint[host.endpoint] = host.host_id
 
     def get_host(self, endpoint_or_address, port=None):
         """
         Find a host in the metadata for a specific endpoint. If a string inet address and port are passed,
         iterate all hosts to match the :attr:`~.pool.Host.broadcast_rpc_address` and
         :attr:`~.pool.Host.broadcast_rpc_port` attributes.
         """
-        if not isinstance(endpoint_or_address, EndPoint):
-            return self._get_host_by_address(endpoint_or_address, port)
+        with self._hosts_lock:
+            if not isinstance(endpoint_or_address, EndPoint):
+                return self._get_host_by_address(endpoint_or_address, port)
 
-        return self._hosts.get(endpoint_or_address)
+            host_id = self._host_id_by_endpoint.get(endpoint_or_address)
+            return self._hosts.get(host_id)
 
     def _get_host_by_address(self, address, port=None):
         for host in six.itervalues(self._hosts):
@@ -365,6 +380,10 @@ def all_hosts(self):
         with self._hosts_lock:
             return list(self._hosts.values())
 
+    def all_hosts_items(self):
+        with self._hosts_lock:
+            return list(self._hosts.items())
+
 
 REPLICATION_STRATEGY_CLASS_PREFIX = "org.apache.cassandra.locator."
 

cassandra/pool.py

@@ -22,6 +22,7 @@
 import time
 import random
 import copy
+import uuid
 from threading import Lock, RLock, Condition
 import weakref
 try:
@@ -174,6 +175,8 @@ def __init__(self, endpoint, conviction_policy_factory, datacenter=None, rack=No
 
         self.endpoint = endpoint if isinstance(endpoint, EndPoint) else DefaultEndPoint(endpoint)
         self.conviction_policy = conviction_policy_factory(self)
+        if not host_id:
+            host_id = uuid.uuid4()
         self.host_id = host_id
         self.set_location_info(datacenter, rack)
         self.lock = RLock()