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
 
@@ -3817,7 +3801,10 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
             tokens = local_row.get("tokens")
 
             host = self._cluster.metadata.get_host(connection.original_endpoint)
-            if host:
+            if not host:
+                log.info("[control connection] Local host %s not found in metadata, adding it", connection.original_endpoint)
+                peers_result.append(local_row)
+            else:
                 datacenter = local_row.get("data_center")
                 rack = local_row.get("rack")
                 self._update_location_info(host, datacenter, rack)
@@ -4173,8 +4160,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

@@ -139,6 +139,10 @@ def export_schema_as_string(self):
     def refresh(self, connection, timeout, target_type=None, change_type=None, fetch_size=None,
                 metadata_request_timeout=None, **kwargs):
 
+        # If the host is not in metadata, we can't proceed, hosts should be added after succesfully establishing control connection
+        if not self.get_host(connection.original_endpoint):
+            return
+
         server_version = self.get_host(connection.original_endpoint).release_version
         dse_version = self.get_host(connection.original_endpoint).dse_version
         parser = get_schema_parser(connection, server_version, dse_version, timeout, metadata_request_timeout, fetch_size)

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