cluster: add control connection query fallback

Author: dkropachev

cassandra/cluster.py

@@ -930,6 +930,17 @@ def default_retry_policy(self, policy):
     If set to :const:`None`, there will be no timeout for these queries.
     """
 
+    allow_control_connection_query_fallback = False
+    """
+    Enables an opt-in degraded availability 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.
+    """
+
     idle_heartbeat_interval = 30
     """
     Interval, in seconds, on which to heartbeat idle connections. This helps
@@ -1216,7 +1227,8 @@ def __init__(self,
                  metadata_request_timeout: Optional[float] = None,
                  column_encryption_policy=None,
                  application_info:Optional[ApplicationInfoBase]=None,
-                 client_routes_config:Optional[ClientRoutesConfig]=None
+                 client_routes_config:Optional[ClientRoutesConfig]=None,
+                 allow_control_connection_query_fallback:Optional[bool]=False
                  ):
         """
         ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as
@@ -1464,6 +1476,7 @@ def __init__(self,
         self.cql_version = cql_version
         self.max_schema_agreement_wait = max_schema_agreement_wait
         self.control_connection_timeout = control_connection_timeout
+        self.allow_control_connection_query_fallback = bool(allow_control_connection_query_fallback)
         self.metadata_request_timeout = self.control_connection_timeout if metadata_request_timeout is None else metadata_request_timeout
         self.idle_heartbeat_interval = idle_heartbeat_interval
         self.idle_heartbeat_timeout = idle_heartbeat_timeout
@@ -4439,6 +4452,7 @@ class ResponseFuture(object):
     _spec_execution_plan = NoSpeculativeExecutionPlan()
     _continuous_paging_session = None
     _host = None
+    _control_connection_query_attempted = False
     _TABLET_ROUTING_CTYPE = None
 
     _warned_timeout = False
@@ -4459,6 +4473,7 @@ def __init__(self, session, message, query, timeout, metrics=None, prepared_stat
         self._callback_lock = Lock()
         self._start_time = start_time or time.time()
         self._host = host
+        self._control_connection_query_attempted = False
         self._spec_execution_plan = speculative_execution_plan or self._spec_execution_plan
         self._make_query_plan()
         self._event = Event()
@@ -4537,11 +4552,16 @@ def _on_timeout(self, _attempts=0):
                         self._connection.orphaned_threshold_reached = True
 
                 pool.return_connection(self._connection, stream_was_orphaned=True)
+            elif getattr(self._connection, 'is_control_connection', False):
+                with self._connection.lock:
+                    self._connection.orphaned_request_ids.add(self._req_id)
+                    if len(self._connection.orphaned_request_ids) >= self._connection.orphaned_threshold:
+                        self._connection.orphaned_threshold_reached = True
 
         errors = self._errors
         if not errors:
             if self.is_schema_agreed:
-                key = str(self._current_host.endpoint) if self._current_host else 'no host queried before timeout'
+                key = str(self._get_host_endpoint(self._current_host)) if self._current_host else 'no host queried before timeout'
                 errors = {key: "Client request timeout. See Session.execute[_async](timeout)"}
             else:
                 connection = self.session.cluster.control_connection._connection
@@ -4599,14 +4619,111 @@ def send_request(self, error_no_hosts=True):
                 self._on_timeout()
                 return True
         if error_no_hosts:
+            if self._fallback_to_control_connection():
+                req_id = self._query_control_connection()
+                if req_id is not None:
+                    self._req_id = req_id
+                    return True
+
             self._set_final_exception(NoHostAvailable(
                 "Unable to complete the operation against any hosts", self._errors))
         return False
 
+    @staticmethod
+    def _get_host_endpoint(host):
+        return getattr(host, 'endpoint', host)
+
+    def _has_usable_node_pool(self):
+        try:
+            pools = tuple(self.session._pools.values())
+        except (AttributeError, TypeError):
+            return False
+
+        return any(pool and not pool.is_shutdown for pool in pools)
+
+    def _fallback_to_control_connection(self):
+        if getattr(self.session.cluster, 'allow_control_connection_query_fallback', False) is not True:
+            return False
+        if self._host or self._control_connection_query_attempted:
+            return False
+        return not self._has_usable_node_pool()
+
+    def _borrow_control_connection(self, connection):
+        with connection.lock:
+            if connection.in_flight >= connection.max_request_id:
+                raise NoConnectionsAvailable("All request IDs are currently in use")
+            connection.in_flight += 1
+            return connection.get_request_id()
+
+    def _release_control_connection_request(self, connection, request_id):
+        with connection.lock:
+            connection.in_flight -= 1
+            connection.request_ids.append(request_id)
+            connection._requests.pop(request_id, None)
+
+    def _handle_control_connection_response(self, connection, cb, response):
+        with connection.lock:
+            connection.in_flight -= 1
+        cb(response)
+
+    def _query_control_connection(self, message=None, cb=None, connection=None, host=None):
+        self._control_connection_query_attempted = True
+
+        if message is None:
+            message = self.message
+
+        if connection is None:
+            control_connection = self.session.cluster.control_connection
+            connection = control_connection._connection if control_connection else None
+        if not connection:
+            self._errors['control connection'] = ConnectionException("Control connection is not connected")
+            return None
+
+        if host is None:
+            host = self.session.cluster.get_control_connection_host() or connection.endpoint
+        self._current_host = host
+
+        request_id = None
+        request_sent = False
+        try:
+            request_id = self._borrow_control_connection(connection)
+            self._connection = connection
+            result_meta = self.prepared_statement.result_metadata if self.prepared_statement else []
+            if cb is None:
+                cb = partial(self._set_result, host, connection, None)
+            cb = partial(self._handle_control_connection_response, connection, cb)
+
+            log.debug("No usable node pools; falling back to control connection for host %s", host)
+            self.request_encoded_size = connection.send_msg(message, request_id, cb=cb,
+                                                            encoder=self._protocol_handler.encode_message,
+                                                            decoder=self._protocol_handler.decode_message,
+                                                            result_metadata=result_meta)
+            request_sent = True
+            self.attempted_hosts.append(host)
+            return request_id
+        except NoConnectionsAvailable as exc:
+            log.debug("Control connection is at capacity")
+            self._errors[host] = exc
+        except ConnectionBusy as exc:
+            log.debug("Control connection is busy")
+            self._errors[host] = exc
+        except Exception as exc:
+            log.debug("Error querying control connection", exc_info=True)
+            self._errors[host] = exc
+            if self._metrics is not None:
+                self._metrics.on_connection_error()
+        finally:
+            if request_id is not None and not request_sent:
+                self._release_control_connection_request(connection, request_id)
+
+        return None
+
     def _query(self, host, message=None, cb=None):
         if message is None:
             message = self.message
 
+        self._control_connection_query_attempted = False
+
         pool = self.session._pools.get(host)
         if not pool:
             self._errors[host] = ConnectionException("Host has been marked down or removed")
@@ -4717,12 +4834,17 @@ def start_fetching_next_page(self):
         self._event.clear()
         self._final_result = _NOT_SET
         self._final_exception = None
+        self._control_connection_query_attempted = False
         self._start_timer()
         self.send_request()
 
     def _reprepare(self, prepare_message, host, connection, pool):
         cb = partial(self.session.submit, self._execute_after_prepare, host, connection, pool)
-        request_id = self._query(host, prepare_message, cb=cb)
+        if pool is None and getattr(connection, 'is_control_connection', False):
+            request_id = self._query_control_connection(prepare_message, cb=cb,
+                                                        connection=connection, host=host)
+        else:
+            request_id = self._query(host, prepare_message, cb=cb)
         if request_id is None:
             # try to submit the original prepared statement on some other host
             self.send_request()
@@ -4940,7 +5062,10 @@ def _execute_after_prepare(self, host, connection, pool, response):
 
                 # use self._query to re-use the same host and
                 # at the same time properly borrow the connection
-                request_id = self._query(host)
+                if pool is None and getattr(connection, 'is_control_connection', False):
+                    request_id = self._query_control_connection(connection=connection, host=host)
+                else:
+                    request_id = self._query(host)
                 if request_id is None:
                     # this host errored out, move on to the next
                     self.send_request()
@@ -5053,6 +5178,11 @@ def _retry_task(self, reuse_connection, host):
             # to retry the operation
             return
 
+        if self._control_connection_query_attempted:
+            self._control_connection_query_attempted = False
+            self.send_request()
+            return
+
         if reuse_connection and self._query(host) is not None:
             return
 

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

tests/unit/test_cluster.py

@@ -184,6 +184,10 @@ 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_flag(self):
+        assert Cluster().allow_control_connection_query_fallback is False
+        assert Cluster(allow_control_connection_query_fallback=True).allow_control_connection_query_fallback is True
+
     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: