diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 068be2e048..b5dc522e7f 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -27,6 +27,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 7260bd08b6..d0cc9a07dc 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -984,13 +984,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 @@ -1204,7 +1248,8 @@ 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, + prepare_on_all_hosts_warmup_seconds=15, reprepare_on_up=True, execution_profiles=None, allow_beta_protocol_version=False, @@ -1480,7 +1525,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) @@ -2642,6 +2692,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: @@ -3238,7 +3291,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) @@ -3247,6 +3300,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_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 35dc354465..ad5ba47a16 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -16,6 +16,7 @@ from concurrent.futures import Future import logging import socket +import time from types import SimpleNamespace from unittest.mock import patch, Mock @@ -381,6 +382,115 @@ 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_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