cluster: add control-connection query fallback

Allow application queries to fall back to the control connection when the driver cannot populate normal pools, such as when the cluster is reached through a non-broadcast API address like a TCP proxy or a node public IP. This is a degraded path with poor throughput and higher churn-related error risk, so it should stay disabled in production.\n\nAlso propagate keyspace updates on the fallback path so USE statements keep the control connection in sync.

Author: dkropachev

cassandra/cluster.py

@@ -932,13 +932,17 @@ def default_retry_policy(self, policy):
 
     allow_control_connection_query_fallback = False
     """
-    Enables an opt-in degraded availability path for application queries.
+    Enables an opt-in degraded path for application queries.
 
     When :const:`True`, a request may be sent on the control connection if
-    the session has no usable node connection pools. This fallback is disabled
-    by default because the control connection is normally reserved for driver
-    metadata and event handling. It is not used for requests targeted to an
-    explicit host.
+    the session has no usable node connection pools. This is intended for
+    deployments that expose the cluster through a non-broadcast API address,
+    such as a TCP proxy or a node's public IP address, where the driver
+    cannot fill the normal pool set. Queries can still execute over the single
+    control connection, but throughput is poor and connection churn raises the
+    chance of request errors. Do not enable this in production.
+
+    This fallback is not used for requests targeted to an explicit host.
     """
 
     idle_heartbeat_interval = 30
@@ -1235,6 +1239,10 @@ def __init__(self,
         extablishing connection pools or refreshing metadata.
 
         Any of the mutable Cluster attributes may be set as keyword arguments to the constructor.
+        ``allow_control_connection_query_fallback`` is a degraded-availability
+        setting for cases where the driver reaches the cluster through a
+        non-broadcast API address and cannot populate the normal pools. It
+        should not be enabled in production.
         """
 
         # Handle port passed as string
@@ -3382,7 +3390,7 @@ def pool_finished_setting_keyspace(pool, host_errors):
                 errors[pool.host] = host_errors
 
             if not remaining_callbacks:
-                callback(host_errors)
+                callback(errors)
 
         for pool in tuple(self._pools.values()):
             pool._set_keyspace_for_all_conns(keyspace, pool_finished_setting_keyspace)
@@ -4883,6 +4891,8 @@ def _set_result(self, host, connection, pool, response):
             if isinstance(response, ResultMessage):
                 if response.kind == RESULT_KIND_SET_KEYSPACE:
                     session = getattr(self, 'session', None)
+                    if connection is not None:
+                        connection.keyspace = response.new_keyspace
                     # since we're running on the event loop thread, we need to
                     # use a non-blocking method for setting the keyspace on
                     # all connections in this session, otherwise the event

tests/unit/test_cluster.py

@@ -23,6 +23,7 @@
     InvalidRequest, Unauthorized, AuthenticationFailed, OperationTimedOut, UnsupportedOperation, RequestValidationException, ConfigurationException, ProtocolVersion
 from cassandra.cluster import _Scheduler, Session, Cluster, default_lbp_factory, \
     ExecutionProfile, _ConfigMode, EXEC_PROFILE_DEFAULT
+from cassandra.connection import 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
@@ -343,6 +344,32 @@ def test_set_keyspace_escapes_quotes(self, *_):
         assert query == 'USE simple_ks', (
             "Simple keyspace names should not be quoted, got: %r" % query)
 
+    @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):

tests/unit/test_response_future.py

@@ -422,6 +422,34 @@ def test_control_connection_fallback_disabled_by_default(self):
         with pytest.raises(NoHostAvailable):
             rf.result()
 
+    def test_control_connection_fallback_updates_connection_keyspace(self):
+        session = self.make_basic_session()
+        session.cluster.allow_control_connection_query_fallback = True
+        session.cluster._default_load_balancing_policy.make_query_plan.return_value = ['ip1']
+        session._pools = {}
+
+        def set_keyspace_for_all_pools(keyspace, callback):
+            session.keyspace = keyspace
+            callback({})
+
+        session._set_keyspace_for_all_pools.side_effect = set_keyspace_for_all_pools
+
+        connection = self.make_control_connection()
+        connection.keyspace = 'oldks'
+        session.cluster.control_connection._connection = connection
+        control_host = Mock(endpoint=connection.endpoint)
+        session.cluster.get_control_connection_host.return_value = control_host
+
+        rf = self.make_response_future(session)
+        assert rf.send_request()
+
+        result = Mock(spec=ResultMessage, kind=RESULT_KIND_SET_KEYSPACE, new_keyspace='newks')
+        connection.send_msg.call_args[1]['cb'](result)
+
+        assert connection.keyspace == 'newks'
+        assert session.keyspace == 'newks'
+        assert rf.result().current_rows == []
+
     def test_control_connection_fallback_when_no_usable_pools(self):
         session = self.make_basic_session()
         session.cluster.allow_control_connection_query_fallback = True