diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 57fcf46331..3787e48f1d 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -51,7 +51,8 @@ from cassandra.connection import (ClientRoutesEndPointFactory, ConnectionException, ConnectionShutdown, ConnectionHeartbeat, ProtocolVersionUnsupported, EndPoint, DefaultEndPoint, DefaultEndPointFactory, - SniEndPointFactory, ConnectionBusy, locally_supported_compressions) + SniEndPointFactory, UnixSocketEndPoint, + ConnectionBusy, locally_supported_compressions) from cassandra.cqltypes import UserType import cassandra.cqltypes as types from cassandra.encoder import Encoder @@ -2225,8 +2226,7 @@ def get_control_connection_host(self): Returns the control connection host metadata. """ connection = self.control_connection._connection - endpoint = connection.endpoint if connection else None - return self.metadata.get_host(endpoint) if endpoint else None + return self.control_connection._get_host_for_connection(connection) def refresh_schema_metadata(self, max_schema_agreement_wait=None): """ @@ -4131,6 +4131,7 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None, found_host_ids = set() found_endpoints = set() + local_row = None if local_result.parsed_rows: local_rows = dict_factory(local_result.column_names, local_result.parsed_rows) local_row = local_rows[0] @@ -4150,11 +4151,13 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None, if not self._is_valid_peer(row): continue - endpoint = self._cluster.endpoint_factory.create(row) + factory_endpoint = self._cluster.endpoint_factory.create(row) host_id = row.get("host_id") - if endpoint in found_endpoints: - log.warning("Found multiple hosts with the same endpoint(%s). Excluding peer %s - %s", endpoint, row.get("peer"), host_id) + # Use the factory endpoint for duplicate detection even when a Unix + # socket is retained as the route to the local host. + if factory_endpoint in found_endpoints: + log.warning("Found multiple hosts with the same endpoint(%s). Excluding peer %s - %s", factory_endpoint, row.get("peer"), host_id) continue if host_id in found_host_ids: @@ -4162,13 +4165,28 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None, continue found_host_ids.add(host_id) - found_endpoints.add(endpoint) + found_endpoints.add(factory_endpoint) + existing_host = self._cluster.metadata.get_host_by_host_id(host_id) + + # Host hashes depend on their endpoint, so never replace the route + # of an existing Host with or from a Unix socket. A newly discovered + # local Host keeps the socket which actually reached the node. + if (existing_host is not None and + isinstance(existing_host.endpoint, UnixSocketEndPoint)): + endpoint = existing_host.endpoint + elif (existing_host is None and row is local_row and + isinstance(connection.original_endpoint, + UnixSocketEndPoint)): + endpoint = connection.original_endpoint + else: + endpoint = factory_endpoint + host = self._cluster.metadata.get_host(endpoint) datacenter = row.get("data_center") rack = row.get("rack") if host is None: - host = self._cluster.metadata.get_host_by_host_id(host_id) + host = existing_host if host and host.endpoint != endpoint: log.debug("[control connection] Updating host ip from %s to %s for (%s)", host.endpoint, endpoint, host_id) reconnector = host.get_and_set_reconnection_handler(None) @@ -4198,6 +4216,9 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None, host.dse_workload = row.get("workload") host.dse_workloads = row.get("workloads") + if row is local_row: + connection._control_connection_host_id = host_id + tokens = row.get("tokens", None) if partitioner and tokens and self._token_meta_enabled: token_map[host] = tokens @@ -4465,8 +4486,15 @@ def _get_schema_mismatches(self, peers_result, local_result, local_address): continue endpoint = self._cluster.endpoint_factory.create(row) peer = self._cluster.metadata.get_host(endpoint) + if peer is None: + peer_by_host_id = self._cluster.metadata.get_host_by_host_id( + row.get('host_id')) + if (peer_by_host_id is not None and + isinstance(peer_by_host_id.endpoint, + UnixSocketEndPoint)): + peer = peer_by_host_id if peer and peer.is_up is not False: - versions[schema_ver].add(endpoint) + versions[schema_ver].add(peer.endpoint) if len(versions) == 1: log.debug("[control connection] Schemas match") @@ -4474,6 +4502,29 @@ def _get_schema_mismatches(self, peers_result, local_result, local_address): return dict((version, list(nodes)) for version, nodes in versions.items()) + def _get_host_for_connection(self, connection): + if connection is None: + return None + + host_id = getattr(connection, '_control_connection_host_id', None) + if host_id is not None: + host = self._cluster.metadata.get_host_by_host_id(host_id) + if host is not None: + return host + + return self._cluster.metadata.get_host(connection.endpoint) + + @staticmethod + def _connection_matches_host(connection, host): + if connection is None: + return False + + host_id = getattr(connection, '_control_connection_host_id', None) + if host_id is not None: + return host_id == host.host_id + + return connection.endpoint == host.endpoint + def _get_peers_query(self, peers_query_type, connection=None): """ Determine the peers query to use. @@ -4504,7 +4555,8 @@ 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) - original_endpoint_host = self._cluster.metadata.get_host(connection.original_endpoint) + original_endpoint_host = self._get_host_for_connection( + connection) 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 = ( @@ -4527,7 +4579,7 @@ def _signal_error(self): # try just signaling the cluster, as this will trigger a reconnect # as part of marking the host down if self._connection and self._connection.is_defunct: - host = self._cluster.metadata.get_host(self._connection.endpoint) + host = self._get_host_for_connection(self._connection) # host may be None if it's already been removed, but that indicates # that errors have already been reported, so we're fine if host: @@ -4545,7 +4597,7 @@ def on_up(self, host): def on_down(self, host): conn = self._connection - if conn and conn.endpoint == host.endpoint and \ + if self._connection_matches_host(conn, host) and \ self._reconnection_handler is None: log.debug("[control connection] Control connection host (%s) is " "considered down, starting reconnection", host) @@ -4558,7 +4610,7 @@ def on_add(self, host, refresh_nodes=True): def on_remove(self, host): c = self._connection - if c and c.endpoint == host.endpoint: + if self._connection_matches_host(c, host): log.debug("[control connection] Control connection host (%s) is being removed. Reconnecting", host) # refresh will be done on reconnect self.reconnect() diff --git a/cassandra/pool.py b/cassandra/pool.py index 1d90e3233f..acda00132f 100644 --- a/cassandra/pool.py +++ b/cassandra/pool.py @@ -29,7 +29,8 @@ from cassandra.util import WeakSet # NOQA from cassandra import AuthenticationFailed -from cassandra.connection import ConnectionException, EndPoint, DefaultEndPoint +from cassandra.connection import (ConnectionException, EndPoint, + DefaultEndPoint, UnixSocketEndPoint) from cassandra.policies import HostDistance log = logging.getLogger(__name__) @@ -241,6 +242,10 @@ def __hash__(self): return hash(self.endpoint) def __lt__(self, other): + self_is_unix = isinstance(self.endpoint, UnixSocketEndPoint) + other_is_unix = isinstance(other.endpoint, UnixSocketEndPoint) + if self_is_unix != other_is_unix: + return self_is_unix return self.endpoint < other.endpoint def __str__(self): diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py index fd62323f33..e8bd4be92f 100644 --- a/tests/unit/test_control_connection.py +++ b/tests/unit/test_control_connection.py @@ -19,9 +19,12 @@ from cassandra import OperationTimedOut, SchemaTargetType, SchemaChangeType from cassandra.protocol import ResultMessage, RESULT_KIND_ROWS -from cassandra.cluster import ControlConnection, _Scheduler, ProfileManager, EXEC_PROFILE_DEFAULT, ExecutionProfile +from cassandra.cluster import (Cluster, ControlConnection, _Scheduler, + ProfileManager, EXEC_PROFILE_DEFAULT, + ExecutionProfile) from cassandra.pool import Host -from cassandra.connection import EndPoint, DefaultEndPoint, DefaultEndPointFactory +from cassandra.connection import (ConnectionException, EndPoint, DefaultEndPoint, + DefaultEndPointFactory, UnixSocketEndPoint) from cassandra.policies import (SimpleConvictionPolicy, RoundRobinPolicy, ConstantReconnectionPolicy, IdentityTranslator) @@ -80,8 +83,8 @@ def add_or_return_host(self, host): def update_host(self, host, old_endpoint): host, created = self.add_or_return_host(host) - self._host_id_by_endpoint[host.endpoint] = host.host_id self._host_id_by_endpoint.pop(old_endpoint, False) + self._host_id_by_endpoint[host.endpoint] = host.host_id def all_hosts_items(self): return list(self.hosts.items()) @@ -205,6 +208,12 @@ def setUp(self): self.control_connection = ControlConnection(self.cluster, 1, 0, 0, 0) self.control_connection._connection = self.connection self.control_connection._time = self.time + self.cluster.control_connection = self.control_connection + + def _forget_local_host(self): + endpoint = DefaultEndPoint('192.168.1.0') + self.cluster.metadata._host_id_by_endpoint.pop(endpoint) + self.cluster.metadata.hosts.pop('uuid1') def test_wait_for_schema_agreement(self): """ @@ -330,6 +339,145 @@ def test_refresh_nodes_and_tokens(self): assert self.connection.wait_for_responses.call_count == 1 + def test_refresh_uses_control_endpoint_for_local_unix_host(self): + maintenance_endpoint = UnixSocketEndPoint('/tmp/maintenance.sock') + self._forget_local_host() + self.connection.endpoint = maintenance_endpoint + self.connection.original_endpoint = maintenance_endpoint + + self.control_connection.refresh_node_list_and_token_map() + + local_host = self.cluster.metadata.get_host_by_host_id('uuid1') + assert local_host.endpoint == maintenance_endpoint + assert local_host.broadcast_rpc_address == '192.168.1.0' + peer_host = self.cluster.metadata.get_host_by_host_id('uuid2') + assert peer_host.endpoint == DefaultEndPoint('192.168.1.1') + assert sorted([local_host, peer_host]) == \ + sorted([peer_host, local_host]) + + def test_refresh_checks_unix_local_advertised_endpoint_for_duplicates(self): + self._forget_local_host() + self.connection.endpoint = UnixSocketEndPoint('/tmp/maintenance.sock') + self.connection.original_endpoint = \ + UnixSocketEndPoint('/tmp/maintenance.sock') + self.connection.peer_results[1].append([ + '192.168.1.0', '10.0.0.4', 'a', 'dc1', 'rack1', + ['4', '104', '204'], 'uuid4']) + + self.control_connection.refresh_node_list_and_token_map() + + assert self.cluster.metadata.get_host_by_host_id('uuid4') is None + + def test_refresh_preserves_known_unix_endpoint_when_host_becomes_peer(self): + maintenance_endpoint = UnixSocketEndPoint('/tmp/maintenance.sock') + self._forget_local_host() + self.connection.endpoint = maintenance_endpoint + self.connection.original_endpoint = maintenance_endpoint + self.control_connection.refresh_node_list_and_token_map() + + local_results = ( + self.connection.local_results[0], + [['192.168.1.1', 'a', 'foocluster', 'dc1', 'rack1', + 'Murmur3Partitioner', '2.2.0', ['1', '101', '201'], + 'uuid2']]) + peer_results = ( + self.connection.peer_results[0], + [['192.168.1.0', '10.0.0.1', 'a', 'dc1', 'rack1', + ['0', '100', '200'], 'uuid1'], + ['192.168.1.2', '10.0.0.2', 'a', 'dc1', 'rack1', + ['2', '102', '202'], 'uuid3']]) + self.connection.endpoint = DefaultEndPoint('192.168.1.1') + self.connection.original_endpoint = self.connection.endpoint + + self.control_connection._refresh_node_list_and_token_map( + self.connection, + preloaded_results=_node_meta_results(local_results, peer_results)) + + local_host = self.cluster.metadata.get_host_by_host_id('uuid1') + assert local_host.endpoint == maintenance_endpoint + + peer_results[1][0][2] = 'b' + peers_response, local_response = _node_meta_results( + local_results, peer_results) + mismatches = self.control_connection._get_schema_mismatches( + peers_response, local_response, self.connection.endpoint) + assert maintenance_endpoint in mismatches['b'] + + def test_refresh_uses_factory_for_local_network_host(self): + self.connection.original_endpoint = DefaultEndPoint('proxy', 9999) + + self.control_connection.refresh_node_list_and_token_map() + + local_host = self.cluster.metadata.get_host_by_host_id('uuid1') + assert local_host.endpoint == DefaultEndPoint('192.168.1.0') + + def test_refresh_network_local_preserves_known_unix_endpoint(self): + maintenance_endpoint = UnixSocketEndPoint('/tmp/maintenance.sock') + self._forget_local_host() + self.connection.endpoint = maintenance_endpoint + self.connection.original_endpoint = \ + maintenance_endpoint + self.control_connection.refresh_node_list_and_token_map() + local_host = self.cluster.metadata.get_host_by_host_id('uuid1') + host_index = {local_host: object()} + + self.connection.endpoint = DefaultEndPoint('192.168.1.0') + self.connection.original_endpoint = self.connection.endpoint + self.control_connection.refresh_node_list_and_token_map() + + assert self.cluster.metadata.get_host_by_host_id('uuid1') is local_host + assert local_host.endpoint == maintenance_endpoint + assert host_index[local_host] is not None + assert Cluster.get_control_connection_host(self.cluster) is local_host + + connection_error = ConnectionException('control connection failed') + self.connection.is_defunct = True + self.connection.last_error = connection_error + self.cluster.signal_connection_failure = Mock() + self.cluster.executor.reset_mock() + + self.control_connection._signal_error() + + self.cluster.signal_connection_failure.assert_called_once_with( + local_host, connection_error, is_host_addition=False) + self.cluster.executor.submit.assert_not_called() + + self.control_connection.on_down(local_host) + + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_remove_matches_control_connection_by_host_id(self): + maintenance_endpoint = UnixSocketEndPoint('/tmp/maintenance.sock') + self._forget_local_host() + self.connection.endpoint = maintenance_endpoint + self.connection.original_endpoint = maintenance_endpoint + self.control_connection.refresh_node_list_and_token_map() + local_host = self.cluster.metadata.get_host_by_host_id('uuid1') + + self.connection.endpoint = DefaultEndPoint('192.168.1.0') + self.cluster.metadata.hosts.pop('uuid1') + self.cluster.metadata._host_id_by_endpoint.pop(maintenance_endpoint) + self.cluster.executor.reset_mock() + + self.control_connection.on_remove(local_host) + + self.cluster.executor.submit.assert_called_once_with( + self.control_connection._reconnect) + + def test_refresh_unix_local_preserves_known_network_endpoint(self): + maintenance_endpoint = UnixSocketEndPoint('/tmp/maintenance.sock') + local_host = self.cluster.metadata.get_host_by_host_id('uuid1') + host_index = {local_host: object()} + self.connection.endpoint = maintenance_endpoint + self.connection.original_endpoint = maintenance_endpoint + + self.control_connection.refresh_node_list_and_token_map() + + assert self.cluster.metadata.get_host_by_host_id('uuid1') is local_host + assert local_host.endpoint == DefaultEndPoint('192.168.1.0') + assert host_index[local_host] is not None + def test_refresh_nodes_and_tokens_with_invalid_peers(self): def refresh_and_validate_added_hosts(): self.connection.wait_for_responses = Mock(return_value=_node_meta_results(