cluster: add control-connection query fallback

docs/api/cassandra/cluster.rst

@@ -48,6 +48,8 @@ Clusters and Sessions
 
    .. autoattribute:: control_connection_timeout
 
+   .. autoattribute:: allow_control_connection_query_fallback
+
    .. autoattribute:: idle_heartbeat_interval
 
    .. autoattribute:: idle_heartbeat_timeout
@@ -106,6 +108,9 @@ Clusters and Sessions
 
    .. automethod:: set_meta_refresh_enabled
 
+.. autoclass:: ControlConnectionQueryFallback
+   :members:
+
 .. autoclass:: ExecutionProfile (load_balancing_policy=<object object>, retry_policy=None, consistency_level=ConsistencyLevel.LOCAL_ONE, serial_consistency_level=None, request_timeout=10.0, row_factory=<function named_tuple_factory>, speculative_execution_policy=None)
    :members:
    :exclude-members: consistency_level

tests/integration/cqlengine/model/test_model.py

@@ -259,10 +259,8 @@ class SensitiveModel(Model):
             rows[-1]
             rows[-1:]
 
-            # ignore DeprecationWarning('The loop argument is deprecated since Python 3.8, and scheduled for removal in Python 3.10.')
-            relevant_warnings = [warn for warn in w if "The loop argument is deprecated" not in str(warn.message)]
+            warning_messages = [str(warn.message) for warn in w]
 
-            assert "__table_name_case_sensitive__ will be removed in 4.0." in str(relevant_warnings[0].message)
-            assert "__table_name_case_sensitive__ will be removed in 4.0." in str(relevant_warnings[1].message)
-            assert "ModelQuerySet indexing with negative indices support will be removed in 4.0." in str(relevant_warnings[2].message)
-            assert "ModelQuerySet slicing with negative indices support will be removed in 4.0." in str(relevant_warnings[3].message)
+            assert sum("__table_name_case_sensitive__ will be removed in 4.0." in message for message in warning_messages) == 2
+            assert sum("ModelQuerySet indexing with negative indices support will be removed in 4.0." in message for message in warning_messages) == 1
+            assert sum("ModelQuerySet slicing with negative indices support will be removed in 4.0." in message for message in warning_messages) == 1

tests/integration/standard/conftest.py

@@ -37,6 +37,7 @@
     "test_ip_change": 4,
     "test_authentication": 4,
     "test_authentication_misconfiguration": 4,
+    "test_control_connection_query_fallback": 4,
     "test_custom_cluster": 4,
     "test_query": 4,
     # Group 5: tablets (destructive — decommissions a node)

tests/integration/standard/test_control_connection_query_fallback.py

@@ -0,0 +1,115 @@
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+import pytest
+
+from cassandra.cluster import ControlConnectionQueryFallback, NoHostAvailable
+
+from tests.integration import USE_CASS_EXTERNAL, TestCluster, local, remove_cluster, use_cluster
+
+
+_CLUSTER_NAME = "control_connection_query_fallback"
+_UNREACHABLE_BROADCAST_RPC_ADDRESS = "127.255.255.1"
+
+
+def setup_module():
+    if USE_CASS_EXTERNAL:
+        return
+
+    remove_cluster()
+
+    ccm_cluster = use_cluster(_CLUSTER_NAME, [1], start=False)
+    ccm_cluster.nodes["node1"].set_configuration_options(values={
+        "broadcast_rpc_address": _UNREACHABLE_BROADCAST_RPC_ADDRESS,
+    })
+    ccm_cluster.start(wait_for_binary_proto=True, wait_other_notice=True)
+
+
+def teardown_module():
+    if USE_CASS_EXTERNAL:
+        return
+
+    remove_cluster()
+
+
+@local
+class ControlConnectionQueryFallbackIntegrationTests(unittest.TestCase):
+
+    def setUp(self):
+        self.cluster = None
+
+    def tearDown(self):
+        if self.cluster is not None:
+            self.cluster.shutdown()
+
+    def _assert_unreachable_broadcast_rpc_metadata(self):
+        hosts = self.cluster.metadata.all_hosts()
+        assert len(hosts) == 1
+
+        host = hosts[0]
+        assert host.broadcast_rpc_address == _UNREACHABLE_BROADCAST_RPC_ADDRESS
+        assert host.endpoint.address == _UNREACHABLE_BROADCAST_RPC_ADDRESS
+        return host
+
+    def test_disabled_raises_when_broadcast_rpc_address_is_unreachable(self):
+        self.cluster = TestCluster(
+            allow_control_connection_query_fallback=ControlConnectionQueryFallback.Disabled,
+            connect_timeout=1,
+            monitor_reporting_enabled=False,
+        )
+
+        with pytest.raises(NoHostAvailable):
+            self.cluster.connect()
+
+        self._assert_unreachable_broadcast_rpc_metadata()
+        assert self.cluster.control_connection._connection is not None
+        assert self.cluster.get_all_pools() == []
+
+    def test_fallback_executes_queries_when_broadcast_rpc_address_is_unreachable(self):
+        self.cluster = TestCluster(
+            allow_control_connection_query_fallback=ControlConnectionQueryFallback.Fallback,
+            connect_timeout=1,
+            monitor_reporting_enabled=False,
+        )
+
+        session = self.cluster.connect()
+
+        self._assert_unreachable_broadcast_rpc_metadata()
+        assert session._initial_connect_futures
+        assert list(session.get_pools()) == []
+
+        row = session.execute(
+            "SELECT release_version, rpc_address FROM system.local WHERE key='local'").one()
+        assert str(row.rpc_address) == _UNREACHABLE_BROADCAST_RPC_ADDRESS
+        assert row.release_version
+
+    def test_no_node_pool_fallback_executes_queries_without_creating_pools(self):
+        self.cluster = TestCluster(
+            allow_control_connection_query_fallback=ControlConnectionQueryFallback.SkipPoolCreation,
+            connect_timeout=1,
+            monitor_reporting_enabled=False,
+        )
+
+        session = self.cluster.connect()
+
+        self._assert_unreachable_broadcast_rpc_metadata()
+        assert session._initial_connect_futures == set()
+        assert list(session.get_pools()) == []
+
+        row = session.execute(
+            "SELECT release_version, rpc_address FROM system.local WHERE key='local'").one()
+        assert str(row.rpc_address) == _UNREACHABLE_BROADCAST_RPC_ADDRESS
+        assert row.release_version

tests/unit/test_cluster.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 import unittest
 
+from concurrent.futures import Future
 import logging
 import socket
 from types import SimpleNamespace
@@ -22,9 +23,9 @@
 
 from cassandra import ConsistencyLevel, DriverException, Timeout, Unavailable, RequestExecutionException, ReadTimeout, WriteTimeout, CoordinationFailure, ReadFailure, WriteFailure, FunctionFailure, AlreadyExists,\
     InvalidRequest, Unauthorized, AuthenticationFailed, OperationTimedOut, UnsupportedOperation, RequestValidationException, ConfigurationException, ProtocolVersion
-from cassandra.cluster import _Scheduler, Session, Cluster, ResultSet, SchemaAgreementScope, default_lbp_factory, \
+from cassandra.cluster import _Scheduler, Session, Cluster, ResultSet, SchemaAgreementScope, ControlConnectionQueryFallback, default_lbp_factory, \
     ExecutionProfile, _ConfigMode, EXEC_PROFILE_DEFAULT
-from cassandra.connection import ConnectionBusy
+from cassandra.connection import ConnectionBusy, ConnectionException
 from cassandra.pool import Host
 from cassandra.policies import HostDistance, RetryPolicy, RoundRobinPolicy, DowngradingConsistencyRetryPolicy, SimpleConvictionPolicy
 from cassandra.query import SimpleStatement, named_tuple_factory, tuple_factory
@@ -186,6 +187,52 @@ def test_port_range(self):
             with pytest.raises(ValueError):
                 cluster = Cluster(contact_points=['127.0.0.1'], port=invalid_port)
 
+    def test_control_connection_query_fallback_modes(self):
+        assert Cluster().allow_control_connection_query_fallback is ControlConnectionQueryFallback.Disabled
+        with pytest.raises(TypeError):
+            Cluster(allow_control_connection_query_fallback=False)
+        with pytest.raises(TypeError):
+            Cluster(allow_control_connection_query_fallback=True)
+        assert (
+            Cluster(allow_control_connection_query_fallback=ControlConnectionQueryFallback.Fallback)
+            .allow_control_connection_query_fallback
+            is ControlConnectionQueryFallback.Fallback
+        )
+        assert Cluster(
+            allow_control_connection_query_fallback=ControlConnectionQueryFallback.SkipPoolCreation
+        ).allow_control_connection_query_fallback is ControlConnectionQueryFallback.SkipPoolCreation
+
+    def test_control_connection_query_fallback_no_node_pool_mode_skips_pool_creation(self):
+        cluster = Cluster(
+            allow_control_connection_query_fallback=ControlConnectionQueryFallback.SkipPoolCreation,
+            monitor_reporting_enabled=False,
+        )
+        host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4())
+
+        with patch.object(Session, "add_or_renew_pool") as mocked_add_or_renew_pool:
+            session = Session(cluster, [host])
+
+        mocked_add_or_renew_pool.assert_not_called()
+        assert session._initial_connect_futures == set()
+        assert session._pools == {}
+        assert session.update_created_pools() == set()
+
+    def test_control_connection_query_fallback_fallback_tolerates_empty_initial_pools(self):
+        cluster = Cluster(
+            allow_control_connection_query_fallback=ControlConnectionQueryFallback.Fallback,
+            monitor_reporting_enabled=False,
+        )
+        host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4())
+        future = Future()
+        future.set_result(False)
+
+        with patch.object(Session, "add_or_renew_pool", return_value=future) as mocked_add_or_renew_pool:
+            session = Session(cluster, [host])
+
+        mocked_add_or_renew_pool.assert_called_once_with(host, is_host_addition=False)
+        assert session._initial_connect_futures == {future}
+        assert session._pools == {}
+
     def test_compression_autodisabled_without_libraries(self):
         with patch.dict('cassandra.cluster.locally_supported_compressions', {}, clear=True):
             with patch('cassandra.cluster.log') as patched_logger:
@@ -454,6 +501,7 @@ def test_set_keyspace_escapes_quotes(self, *_):
             "Simple keyspace names should not be quoted, got: %r" % query)
 
     @mock_session_pools
+<<<<<<< HEAD
     def test_wait_for_schema_agreement_default_scope_queries_all_connected_hosts(self, *_):
         session, hosts, _ = self._new_schema_agreement_session(
             ["a", "a"],
@@ -550,6 +598,32 @@ def test_wait_for_schema_agreement_rejects_unknown_scope(self, *_):
 
         with pytest.raises(ValueError):
             session.wait_for_schema_agreement(wait_time=1, scope='planet')
+=======
+    def test_set_keyspace_for_all_pools_reports_all_errors(self, *_):
+        cluster = Cluster()
+        session = Session(
+            cluster,
+            [Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4())],
+        )
+
+        pool1 = Mock(host='host1')
+        pool2 = Mock(host='host2')
+        keyspace_error = ConnectionException("boom")
+
+        pool1._set_keyspace_for_all_conns.side_effect = (
+            lambda keyspace, callback: callback(pool1, [keyspace_error])
+        )
+        pool2._set_keyspace_for_all_conns.side_effect = (
+            lambda keyspace, callback: callback(pool2, [])
+        )
+        session._pools = {'host1': pool1, 'host2': pool2}
+
+        callback = Mock()
+        session._set_keyspace_for_all_pools('ks', callback)
+
+        callback.assert_called_once()
+        assert callback.call_args.args[0] == {'host1': [keyspace_error]}
+>>>>>>> 40719aea4 (cluster: add control-connection query fallback)
 
 class ProtocolVersionTests(unittest.TestCase):