Don't create Host instances with random host_id

Previously, we used endpoints provided to the cluster to create Host
instances with random host_ids in order to populate the LBP before the
ControlConnection was established. This logic led to creating many
connections that were opened and then quickly closed, because once we
learned the correct host_ids from system.peers, we removed the old Hosts
with random IDs and created new ones with the proper host_ids.

This commit introduces a new approach. To establish the ControlConnection,
we now use only the resolved contact points from the cluster configuration.
Only after a successful connection do we populate Host information in the
LBP. If the LBP is already initialized during ControlConnection
reconnection, we reuse the existing values.

Author: sylwiaszunejko

cassandra/cluster.py

@@ -1683,14 +1683,7 @@ def protocol_downgrade(self, host_endpoint, previous_version):
                     "http://datastax.github.io/python-driver/api/cassandra/cluster.html#cassandra.cluster.Cluster.protocol_version", self.protocol_version, new_version, host_endpoint)
         self.protocol_version = new_version
 
-    def _add_resolved_hosts(self):
-        for endpoint in self.endpoints_resolved:
-            host, new = self.add_host(endpoint, signal=False)
-            if new:
-                host.set_up()
-                for listener in self.listeners:
-                    listener.on_add(host)
-
+    def _populate_hosts(self):
         self.profile_manager.populate(
             weakref.proxy(self), self.metadata.all_hosts())
         self.load_balancing_policy.populate(
@@ -1718,16 +1711,9 @@ def connect(self, keyspace=None, wait_for_all_pools=False):
                 self.connection_class.initialize_reactor()
                 _register_cluster_shutdown(self)
 
-                self._add_resolved_hosts()
-
                 try:
                     self.control_connection.connect()
-
-                    # we set all contact points up for connecting, but we won't infer state after this
-                    for endpoint in self.endpoints_resolved:
-                        h = self.metadata.get_host(endpoint)
-                        if h and self.profile_manager.distance(h) == HostDistance.IGNORED:
-                            h.is_up = None
+                    self._populate_hosts()
 
                     log.debug("Control connection created")
                 except Exception:
@@ -3535,20 +3521,18 @@ def _set_new_connection(self, conn):
             log.debug("[control connection] Closing old connection %r, replacing with %r", old, conn)
             old.close()
 
-    def _connect_host_in_lbp(self):
+    def _try_connect_to_hosts(self):
         errors = {}
-        lbp = (
-            self._cluster.load_balancing_policy
-            if self._cluster._config_mode == _ConfigMode.LEGACY else
-            self._cluster._default_load_balancing_policy
-        )
 
-        for host in lbp.make_query_plan():
+        lbp = self._cluster.load_balancing_policy \
+            if self._cluster._config_mode == _ConfigMode.LEGACY else self._cluster._default_load_balancing_policy
+
+        for endpoint in chain((host.endpoint for host in lbp.make_query_plan()), self._cluster.endpoints_resolved):
             try:
-                return (self._try_connect(host.endpoint), None)
+                return (self._try_connect(endpoint), None)
             except Exception as exc:
-                errors[str(host.endpoint)] = exc
-                log.warning("[control connection] Error connecting to %s:", host, exc_info=True)
+                errors[str(endpoint)] = exc
+                log.warning("[control connection] Error connecting to %s:", endpoint, exc_info=True)
             if self._is_shutdown:
                 raise DriverException("[control connection] Reconnection in progress during shutdown")
 
@@ -3563,16 +3547,16 @@ def _reconnect_internal(self):
         to the exception that was raised when an attempt was made to open
         a connection to that host.
         """
-        (conn, _) = self._connect_host_in_lbp()
+        (conn, _) = self._try_connect_to_hosts()
         if conn is not None:
             return conn
 
         # Try to re-resolve hostnames as a fallback when all hosts are unreachable
         self._cluster._resolve_hostnames()
 
-        self._cluster._add_resolved_hosts()
+        self._cluster._populate_hosts()
 
-        (conn, errors) = self._connect_host_in_lbp()
+        (conn, errors) = self._try_connect_to_hosts()
         if conn is not None:
             return conn
 
@@ -3813,68 +3797,8 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
             cluster_name = local_row["cluster_name"]
             self._cluster.metadata.cluster_name = cluster_name
 
-            partitioner = local_row.get("partitioner")
-            tokens = local_row.get("tokens")
-
-            host = self._cluster.metadata.get_host(connection.original_endpoint)
-            if host:
-                datacenter = local_row.get("data_center")
-                rack = local_row.get("rack")
-                self._update_location_info(host, datacenter, rack)
-
-                # support the use case of connecting only with public address
-                if isinstance(self._cluster.endpoint_factory, SniEndPointFactory):
-                    new_endpoint = self._cluster.endpoint_factory.create(local_row)
-
-                    if new_endpoint.address:
-                        host.endpoint = new_endpoint
-
-                host.host_id = local_row.get("host_id")
-
-                found_host_ids.add(host.host_id)
-                found_endpoints.add(host.endpoint)
-
-                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)
-                host.broadcast_port = _NodeInfo.get_broadcast_port(local_row)
-
-                host.broadcast_rpc_address = _NodeInfo.get_broadcast_rpc_address(local_row)
-                host.broadcast_rpc_port = _NodeInfo.get_broadcast_rpc_port(local_row)
-                if host.broadcast_rpc_address is None:
-                    if self._token_meta_enabled:
-                        # local rpc_address is not available, use the connection endpoint
-                        host.broadcast_rpc_address = connection.endpoint.address
-                        host.broadcast_rpc_port = connection.endpoint.port
-                    else:
-                        # local rpc_address has not been queried yet, try to fetch it
-                        # separately, which might fail because C* < 2.1.6 doesn't have rpc_address
-                        # in system.local. See CASSANDRA-9436.
-                        local_rpc_address_query = QueryMessage(
-                            query=maybe_add_timeout_to_query(self._SELECT_LOCAL_NO_TOKENS_RPC_ADDRESS, self._metadata_request_timeout),
-                            consistency_level=ConsistencyLevel.ONE)
-                        success, local_rpc_address_result = connection.wait_for_response(
-                            local_rpc_address_query, timeout=self._timeout, fail_on_error=False)
-                        if success:
-                            row = dict_factory(
-                                local_rpc_address_result.column_names,
-                                local_rpc_address_result.parsed_rows)
-                            host.broadcast_rpc_address = _NodeInfo.get_broadcast_rpc_address(row[0])
-                            host.broadcast_rpc_port = _NodeInfo.get_broadcast_rpc_port(row[0])
-                        else:
-                            host.broadcast_rpc_address = connection.endpoint.address
-                            host.broadcast_rpc_port = connection.endpoint.port
-
-                host.release_version = local_row.get("release_version")
-                host.dse_version = local_row.get("dse_version")
-                host.dse_workload = local_row.get("workload")
-                host.dse_workloads = local_row.get("workloads")
-
-                if partitioner and tokens:
-                    token_map[host] = tokens
+            peers_result.insert(0, local_row)
 
-                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)
@@ -4173,8 +4097,9 @@ def _get_peers_query(self, peers_query_type, connection=None):
                 query_template = (self._SELECT_SCHEMA_PEERS_TEMPLATE
                                   if peers_query_type == self.PeersQueryType.PEERS_SCHEMA
                                   else self._SELECT_PEERS_NO_TOKENS_TEMPLATE)
-                host_release_version = self._cluster.metadata.get_host(connection.original_endpoint).release_version
-                host_dse_version = self._cluster.metadata.get_host(connection.original_endpoint).dse_version
+                original_endpoint_host = self._cluster.metadata.get_host(connection.original_endpoint)
+                host_release_version = None if original_endpoint_host is None else original_endpoint_host.release_version
+                host_dse_version = None if original_endpoint_host is None else original_endpoint_host.dse_version
                 uses_native_address_query = (
                     host_dse_version and Version(host_dse_version) >= self._MINIMUM_NATIVE_ADDRESS_DSE_VERSION)
 

cassandra/metadata.py

@@ -3481,7 +3481,7 @@ def group_keys_by_replica(session, keyspace, table, keys):
     :class:`~.NO_VALID_REPLICA`
 
     Example usage::
-        
+
         >>> result = group_keys_by_replica(
         ...     session, "system", "peers",
         ...     (("127.0.0.1", ), ("127.0.0.2", )))

cassandra/pool.py

@@ -176,7 +176,7 @@ 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()
+            raise ValueError("host_id may not be None")
         self.host_id = host_id
         self.set_location_info(datacenter, rack)
         self.lock = RLock()