diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 26725cfedc..a3b57388b9 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -58,6 +58,17 @@ Others come through unchanged. Previously ``DRIVER_NAME`` and ``DRIVER_VERSION`` could be overridden, which misreported the driver to the server for the life of the connection and, in the clients table, to the operator reading the row. +* ``Cluster.prepare_on_all_hosts`` now defaults to unset instead of ``True``. In multi-DC + deployments eager preparation previously ran on every pooled host, including remote hosts + that are rarely or never queried. Left unset, a ``Session`` now eagerly prepares on all + hosts only during a short warm-up window after it connects (``prepare_on_all_hosts_warmup_seconds``, + default 15s), when hosts have just been discovered and many different statements are likely + to hit many different hosts in quick succession; afterwards it falls back to the lazy + behavior (``prepare_on_all_hosts=False``), since steady-state traffic for a given prepared + statement usually concentrates on a stable subset of replicas via token-aware routing. + Passing ``prepare_on_all_hosts=True`` or ``False`` explicitly disables the warm-up and pins + the old, unconditional behavior for the life of the cluster. An ``UNPREPARED`` response + still triggers on-demand reprepare and retry, so correctness is unaffected either way. * ``PreparedStatement.result_metadata`` and ``PreparedStatement.result_metadata_id`` are now read-only. They are replaced together by ``PreparedStatement.update_result_metadata()``, so a request can never observe a metadata diff --git a/cassandra/cluster.py b/cassandra/cluster.py index d858f5835e..8c5aa8673d 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -985,13 +985,57 @@ def default_retry_policy(self, policy): establish connection pools. This can cause a rush of connections and queries if not mitigated with this factor. """ - prepare_on_all_hosts = True + _prepare_on_all_hosts = False + _prepare_on_all_hosts_explicit = False + + @property + def prepare_on_all_hosts(self): + """ + Specifies whether statements should be prepared on all hosts, or just one. + + When enabled, statements are eagerly prepared on every host with an open connection pool. In multi-DC + deployments this includes remote hosts that are rarely or never queried on the happy path; preparing on them + is purely a latency optimization, since an ``UNPREPARED`` response always triggers on-demand reprepare and + retry. It can be enabled on long-running applications with numerous clients preparing statements on startup, + where a randomized initial condition of the load balancing policy can be expected to distribute prepares from + different clients across the cluster. + + If left unset (the default), a :class:`.Session` instead applies :attr:`.prepare_on_all_hosts_warmup_seconds`: + it behaves as if this were ``True`` for a short warm-up window right after the session connects, then as if + ``False`` afterwards. Explicitly assigning ``True`` or ``False``, whether to the :class:`.Cluster` + constructor or to this attribute at any later point, disables the warm-up behavior and pins this to the + given value for the lifetime of the cluster. + """ + return self._prepare_on_all_hosts + + @prepare_on_all_hosts.setter + def prepare_on_all_hosts(self, value): + self._prepare_on_all_hosts = value + self._prepare_on_all_hosts_explicit = True + + prepare_on_all_hosts_warmup_seconds = 15 """ - Specifies whether statements should be prepared on all hosts, or just one. + Length, in seconds, of the warm-up window used to decide whether :meth:`.Session.prepare` eagerly prepares + on all pooled hosts, when :attr:`.prepare_on_all_hosts` was not explicitly set by the caller. + + Right after a :class:`.Session` connects, hosts have just been discovered and different callers/tests + typically prepare many different statements against many different hosts in quick succession; eagerly + broadcasting each prepare avoids a burst of ``UNPREPARED``/reprepare/retry round trips during that period. + In steady state, query traffic for a given prepared statement usually concentrates on a stable subset of + replicas (via token-aware routing), so broadcasting to every host is normally wasted work, and the driver + falls back to lazy on-demand reprepare (the same behavior as ``prepare_on_all_hosts=False``). - This can reasonably be disabled on long-running applications with numerous clients preparing statements on startup, - where a randomized initial condition of the load balancing policy can be expected to distribute prepares from - different clients across the cluster. + The window is measured from when the :class:`.Session` finished establishing its initial connection pools, + not from the first call to :meth:`.Session.prepare`. An application that waits well past connect before + ever calling ``prepare()`` (lazy-first-use) will not benefit from the warm-up window, since by then hosts + are no longer "freshly discovered" and the startup thundering-herd risk this is meant to mitigate has + already passed. + + Setting this to zero (or a falsy value) disables the warm-up behavior entirely, equivalent to leaving + :attr:`.prepare_on_all_hosts` at its unset default with no warm-up: statements are never eagerly broadcast + unless the flag is set explicitly. + + Has no effect when :attr:`.prepare_on_all_hosts` was explicitly set by the caller. """ reprepare_on_up = True @@ -1205,7 +1249,7 @@ def __init__(self, schema_metadata_page_size=1000, address_translator=None, status_event_refresh_window=2, - prepare_on_all_hosts=True, + prepare_on_all_hosts=_NOT_SET, reprepare_on_up=True, execution_profiles=None, allow_beta_protocol_version=False, @@ -1222,7 +1266,8 @@ def __init__(self, application_info:Optional[ApplicationInfoBase]=None, client_routes_config:Optional[ClientRoutesConfig]=None, allow_control_connection_query_fallback:Optional[ControlConnectionQueryFallback]=ControlConnectionQueryFallback.Disabled, - driver_config_reporting_enabled=True + driver_config_reporting_enabled=True, + prepare_on_all_hosts_warmup_seconds=15 ): """ ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as @@ -1491,7 +1536,12 @@ def __init__(self, self.topology_event_refresh_window = topology_event_refresh_window self.status_event_refresh_window = status_event_refresh_window self.connect_timeout = connect_timeout - self.prepare_on_all_hosts = prepare_on_all_hosts + if prepare_on_all_hosts is _NOT_SET: + self._prepare_on_all_hosts = False + self._prepare_on_all_hosts_explicit = False + else: + self.prepare_on_all_hosts = prepare_on_all_hosts + self.prepare_on_all_hosts_warmup_seconds = prepare_on_all_hosts_warmup_seconds self.reprepare_on_up = reprepare_on_up self.shard_aware_options = ShardAwareOptions(opts=shard_aware_options) @@ -1786,6 +1836,8 @@ def connect(self, keyspace=None, wait_for_all_pools=False): session = self._new_session(keyspace) if wait_for_all_pools: wait_futures(session._initial_connect_futures) + # reset so the warm-up window starts after all pools are up, not just the first + session._connect_time = time.time() self._set_default_dbaas_consistency(session) @@ -2652,6 +2704,9 @@ def __init__(self, cluster, hosts, keyspace=None): raise NoHostAvailable(msg, [h.address for h in hosts]) self.session_id = uuid.uuid4() + # marks when this session finished its initial pool setup; used to gauge whether we're + # still in the post-connect warm-up window for prepare_on_all_hosts (see _should_prepare_on_all_hosts) + self._connect_time = time.time() if self.cluster.column_encryption_policy is not None: try: @@ -3248,7 +3303,7 @@ def prepare(self, query, custom_payload=None, keyspace=None): self.cluster.add_prepared(response.query_id, prepared_statement) - if self.cluster.prepare_on_all_hosts: + if self._should_prepare_on_all_hosts(): host = future._current_host try: self.prepare_on_all_hosts(prepared_statement.query_string, host, prepared_keyspace) @@ -3257,6 +3312,23 @@ def prepare(self, query, custom_payload=None, keyspace=None): return prepared_statement + def _should_prepare_on_all_hosts(self): + """ + Decide whether this prepare() call should eagerly broadcast to all pooled hosts. + + If the user explicitly set Cluster.prepare_on_all_hosts, that choice always wins. Otherwise, act as + if it were True during the post-connect warm-up window (see prepare_on_all_hosts_warmup_seconds) and + False afterwards. + """ + cluster = self.cluster + if cluster._prepare_on_all_hosts_explicit: + return cluster.prepare_on_all_hosts + + warmup_seconds = cluster.prepare_on_all_hosts_warmup_seconds + if not warmup_seconds: + return False + return (time.time() - self._connect_time) <= warmup_seconds + def prepare_on_all_hosts(self, query, excluded_host, keyspace=None): """ Prepare the given query on all hosts, excluding ``excluded_host``. diff --git a/tests/integration/standard/test_query.py b/tests/integration/standard/test_query.py index 5f1d5bfc19..b342a4ceb4 100644 --- a/tests/integration/standard/test_query.py +++ b/tests/integration/standard/test_query.py @@ -527,6 +527,37 @@ def test_prepare_on_all_hosts(self): session.execute(select_statement, (1, ), host=host) assert 2 == mock_handler.get_message_count('debug', "Re-preparing") + def test_prepare_on_all_hosts_default_and_explicit_true(self): + """ + Regression test for the prepare_on_all_hosts default flip to False. + + test_prepare_on_all_hosts above pins prepare_on_all_hosts=False explicitly, so it + can't catch a regression in the class attribute or constructor default. Disable the + warm-up shim (warmup_seconds=0) so the unset default is exercised deterministically, + and also cover the explicit True opt-in to eager preparation. + """ + with MockLoggingHandler().set_module_name(cluster.__name__) as mock_handler: + clus = TestCluster(reprepare_on_up=False, prepare_on_all_hosts_warmup_seconds=0) + self.addCleanup(clus.shutdown) + assert clus.prepare_on_all_hosts is False + + session = clus.connect(wait_for_all_pools=True) + select_statement = session.prepare("SELECT k FROM test3rf.test WHERE k = ?") + for host in clus.metadata.all_hosts(): + session.execute(select_statement, (1, ), host=host) + assert 2 == mock_handler.get_message_count('debug', "Re-preparing") + + with MockLoggingHandler().set_module_name(cluster.__name__) as mock_handler: + clus = TestCluster(prepare_on_all_hosts=True, reprepare_on_up=False) + self.addCleanup(clus.shutdown) + assert clus.prepare_on_all_hosts is True + + session = clus.connect(wait_for_all_pools=True) + select_statement = session.prepare("SELECT k FROM test3rf.test WHERE k = ?") + for host in clus.metadata.all_hosts(): + session.execute(select_statement, (1, ), host=host) + assert 0 == mock_handler.get_message_count('debug', "Re-preparing") + def test_prepare_batch_statement(self): """ Test to validate a prepared statement used inside a batch statement is correctly handled diff --git a/tests/integration/standard/test_shard_aware.py b/tests/integration/standard/test_shard_aware.py index 6daba6e26f..654ddb20df 100644 --- a/tests/integration/standard/test_shard_aware.py +++ b/tests/integration/standard/test_shard_aware.py @@ -71,7 +71,10 @@ def verify_same_shard_in_tracing(self, results, shard_name): assert shard_name in event.thread_name assert 'querying locally' in "\n".join([event.description for event in events]) - trace_id = results.response_future.get_query_trace_ids()[0] + # Use the last trace id: prepare_on_all_hosts defaults to False now, so a query + # against a host that hasn't prepared the statement yet can get UNPREPARED and + # retry, which appends an earlier, incomplete trace before the one that matters. + trace_id = results.response_future.get_query_trace_ids()[-1] traces = self.session.execute("SELECT * FROM system_traces.events WHERE session_id = %s", (trace_id,)) events = [event for event in traces] for event in events: diff --git a/tests/integration/standard/test_tablets.py b/tests/integration/standard/test_tablets.py index 0491b15f3f..4a37b7a84f 100644 --- a/tests/integration/standard/test_tablets.py +++ b/tests/integration/standard/test_tablets.py @@ -37,7 +37,10 @@ def verify_hosts_in_tracing(self, results, expected): assert len(host_set) == expected assert 'locally' in "\n".join([event.description for event in events]) - trace_id = results.response_future.get_query_trace_ids()[0] + # Use the last trace id: prepare_on_all_hosts defaults to False now, so a query + # against a host that hasn't prepared the statement yet can get UNPREPARED and + # retry, which appends an earlier, incomplete trace before the one that matters. + trace_id = results.response_future.get_query_trace_ids()[-1] traces = self.session.execute("SELECT * FROM system_traces.events WHERE session_id = %s", (trace_id,)) events = [event for event in traces] host_set = set() @@ -63,7 +66,8 @@ def verify_same_shard_in_tracing(self, results): assert len(shard_set) == 1 assert 'locally' in "\n".join([event.description for event in events]) - trace_id = results.response_future.get_query_trace_ids()[0] + # See verify_hosts_in_tracing: use the last trace id, not the first. + trace_id = results.response_future.get_query_trace_ids()[-1] traces = self.session.execute("SELECT * FROM system_traces.events WHERE session_id = %s", (trace_id,)) events = [event for event in traces] shard_set = set() diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 74ed346c68..c5a00bdb0b 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -14,8 +14,11 @@ import unittest from concurrent.futures import Future +import inspect +import itertools import logging import socket +import time from types import SimpleNamespace from unittest.mock import patch, Mock @@ -190,6 +193,41 @@ def test_port_range(self): with pytest.raises(ValueError): cluster = Cluster(contact_points=['127.0.0.1'], port=invalid_port) + def test_prepare_on_all_hosts_defaults_to_unset(self): + cluster = Cluster() + self.addCleanup(cluster.shutdown) + assert cluster.prepare_on_all_hosts is False + assert cluster._prepare_on_all_hosts_explicit is False + + explicit_cluster = Cluster(prepare_on_all_hosts=True) + self.addCleanup(explicit_cluster.shutdown) + assert explicit_cluster.prepare_on_all_hosts is True + assert explicit_cluster._prepare_on_all_hosts_explicit is True + + def test_reprepare_on_up_keeps_its_positional_slot(self): + # guards against prepare_on_all_hosts_warmup_seconds shifting reprepare_on_up's + # position and silently breaking callers that pass it positionally; the names + # below are the pre-existing params, in their pre-existing order, up to and + # including reprepare_on_up (see e.g. commit 3583fdcf's Cluster.__init__) + pre_existing_params_up_to_reprepare_on_up = [ + 'contact_points', 'port', 'compression', 'auth_provider', 'load_balancing_policy', + 'reconnection_policy', 'default_retry_policy', 'conviction_policy_factory', + 'metrics_enabled', 'connection_class', 'ssl_options', 'sockopts', 'cql_version', + 'protocol_version', 'executor_threads', 'max_schema_agreement_wait', + 'control_connection_timeout', 'idle_heartbeat_interval', 'schema_event_refresh_window', + 'topology_event_refresh_window', 'connect_timeout', 'schema_metadata_enabled', + 'token_metadata_enabled', 'schema_metadata_page_size', 'address_translator', + 'status_event_refresh_window', 'prepare_on_all_hosts', + ] + sig = inspect.signature(Cluster.__init__) + args = [sig.parameters[name].default for name in pre_existing_params_up_to_reprepare_on_up] + args.append(False) # positionally where reprepare_on_up used to be (and must still be) + cluster = Cluster(*args) + self.addCleanup(cluster.shutdown) + assert cluster.reprepare_on_up is False + assert cluster.prepare_on_all_hosts is False + assert cluster._prepare_on_all_hosts_explicit is False + def test_control_connection_query_fallback_modes(self): default_cluster = Cluster() self.addCleanup(default_cluster.shutdown) @@ -421,6 +459,140 @@ def test_connection_factory_ignores_a_caller_supplied_session_id_and_reporter(se assert factory.call_args.kwargs['driver_config_reporter'] is None +class PrepareOnAllHostsWarmupTest(unittest.TestCase): + """ + Covers the post-connect warm-up window that decides whether Session.prepare() + eagerly broadcasts to all pooled hosts when Cluster.prepare_on_all_hosts was + left unset. See Session._should_prepare_on_all_hosts. + """ + + def _make_session(self, **cluster_kwargs): + cluster = Cluster(**cluster_kwargs) + self.addCleanup(cluster.shutdown) + host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4()) + host.set_up() + cluster.metadata.add_or_return_host(host) + return Session(cluster, [host]) + + @mock_session_pools + def test_within_warmup_window_prepares_eagerly_by_default(self, *_): + session = self._make_session() + session._connect_time = time.time() + + assert session._should_prepare_on_all_hosts() is True + + @mock_session_pools + def test_after_warmup_window_falls_back_to_lazy_by_default(self, *_): + session = self._make_session() + session._connect_time = time.time() - session.cluster.prepare_on_all_hosts_warmup_seconds - 1 + + assert session._should_prepare_on_all_hosts() is False + + @mock_session_pools + def test_explicit_true_is_respected_even_after_warmup_elapses(self, *_): + session = self._make_session(prepare_on_all_hosts=True) + session._connect_time = time.time() - session.cluster.prepare_on_all_hosts_warmup_seconds - 1 + + assert session._should_prepare_on_all_hosts() is True + + @mock_session_pools + def test_explicit_false_is_respected_even_within_warmup_window(self, *_): + session = self._make_session(prepare_on_all_hosts=False) + session._connect_time = time.time() + + assert session._should_prepare_on_all_hosts() is False + + @mock_session_pools + def test_runtime_assignment_after_construction_is_respected(self, *_): + session = self._make_session() + session._connect_time = time.time() - session.cluster.prepare_on_all_hosts_warmup_seconds - 1 + + session.cluster.prepare_on_all_hosts = True + assert session._should_prepare_on_all_hosts() is True + + session._connect_time = time.time() + session.cluster.prepare_on_all_hosts = False + assert session._should_prepare_on_all_hosts() is False + + @mock_session_pools + def test_zero_warmup_seconds_disables_eager_behavior(self, *_): + session = self._make_session(prepare_on_all_hosts_warmup_seconds=0) + session._connect_time = time.time() + + assert session._should_prepare_on_all_hosts() is False + + @mock_session_pools + def test_prepare_uses_should_prepare_on_all_hosts_decision(self, *_): + session = self._make_session() + session._connect_time = time.time() + + message = Mock(query_id=b'qid', bind_metadata=[], pk_indexes=[], column_metadata=[], + result_metadata_id=None, is_lwt=False) + future = Mock() + future.result.return_value.one.return_value = message + future._current_host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4()) + + with patch('cassandra.cluster.ResponseFuture', return_value=future), \ + patch.object(session.cluster, 'add_prepared'), \ + patch.object(Session, 'prepare_on_all_hosts') as prepare_on_all_hosts: + session.prepare("SELECT * FROM t") + + assert prepare_on_all_hosts.call_count == 1 + + session._connect_time = time.time() - session.cluster.prepare_on_all_hosts_warmup_seconds - 1 + with patch('cassandra.cluster.ResponseFuture', return_value=future), \ + patch.object(session.cluster, 'add_prepared'), \ + patch.object(Session, 'prepare_on_all_hosts') as prepare_on_all_hosts: + session.prepare("SELECT * FROM t") + + assert prepare_on_all_hosts.call_count == 0 + + @mock_session_pools + def test_wait_for_all_pools_starts_warmup_window_after_slow_pool_connects(self, *_): + # Cluster.connect(wait_for_all_pools=True) must reset the warm-up window's start + # to after Session.__init__'s own wait (first pool up) finishes waiting for every + # initial pool, not leave it at Session.__init__'s earlier timestamp. + cluster = Cluster() + self.addCleanup(cluster.shutdown) + host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4()) + host.set_up() + cluster.metadata.add_or_return_host(host) + + # A monotonically increasing counter, not a fixed-size list: DEBUG-level log + # records also call time.time() on some Python versions, so the exact number + # of calls before connect()'s reset can't be pinned down to a fixed count. + timestamps = itertools.count(100.0, 100.0) + with patch.object(cluster.control_connection, 'connect'), \ + patch.object(cluster, '_populate_hosts'), \ + patch.object(cluster.profile_manager, 'check_supported'), \ + patch('cassandra.cluster.time.time', side_effect=lambda: next(timestamps)): + session = cluster.connect(wait_for_all_pools=True) + + # connect()'s reset must be later than the first timestamp ever handed out + # (Session.__init__'s own stamp), proving it didn't just keep that early value. + assert session._connect_time > 100.0 + + @mock_session_pools + def test_prepare_all_queries_on_host_up_is_unaffected_by_flag_or_warmup(self, *_): + # Cluster._prepare_all_queries (the reprepare_on_up path for late-joining hosts) + # is a separate mechanism from prepare_on_all_hosts/warmup and must keep firing + # regardless of either. + session = self._make_session(prepare_on_all_hosts=False, prepare_on_all_hosts_warmup_seconds=0) + session._connect_time = time.time() - 1000 + cluster = session.cluster + + prepared_statement = Mock(query_string="SELECT * FROM t", keyspace=None) + cluster._prepared_statements = {b'qid': prepared_statement} + + new_host = Host("127.0.0.2", SimpleConvictionPolicy, host_id=uuid.uuid4()) + new_host.set_up() + + with patch.object(cluster, 'connection_factory') as connection_factory: + cluster._prepare_all_queries(new_host) + + assert connection_factory.call_count == 1 + + class SchedulerTest(unittest.TestCase): # TODO: this suite could be expanded; for now just adding a test covering a ticket