cluster: add control-connection query fallback

Add an opt-in control-connection fallback for application queries when the driver cannot populate normal node pools, which happens in deployments that expose the cluster through a non-broadcast IP address such as a TCP proxy or a node public IP. In that mode the driver can still execute queries over the single control connection, but throughput is poor and connection churn increases the chance of request errors. This option is intentionally disabled by default and should not be used in production.

Also propagate keyspace updates on the fallback path so USE keeps the control connection in sync.

Tests:
- tests/unit/test_cluster.py::ClusterTest::test_set_keyspace_for_all_pools_reports_all_errors
- tests/unit/test_response_future.py::ResponseFutureTests::test_control_connection_fallback_updates_connection_keyspace

Author: dkropachev

cassandra/cluster.py

@@ -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

docs/api/cassandra/cluster.rst

@@ -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/cqlengine/model/test_model.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/conftest.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/integration/standard/test_control_connection_query_fallback.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:
@@ -551,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')
 
+    @mock_session_pools
+    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]}
+
 class ProtocolVersionTests(unittest.TestCase):
 
     def test_protocol_downgrade_test(self):