From 2c44ec8d0a6199f713621a1f3dd5aadd95c66651 Mon Sep 17 00:00:00 2001 From: Emmanuel T Odeke Date: Wed, 30 Apr 2025 22:21:32 +0300 Subject: [PATCH 1/4] chore(x-goog-spanner-request-id): plug in functionality after test scaffolding This change chops down the load of the large changes for x-goog-spanner-request-id. It depends on PR #1366 and should only be merged after that PR. Updates #1261 Requires PR #1366 --- google/cloud/spanner_v1/database.py | 97 ++++- google/cloud/spanner_v1/pool.py | 8 +- google/cloud/spanner_v1/session.py | 39 +- google/cloud/spanner_v1/snapshot.py | 134 ++++-- .../cloud/spanner_v1/testing/interceptors.py | 2 +- google/cloud/spanner_v1/transaction.py | 120 ++++-- tests/unit/test_batch.py | 37 ++ tests/unit/test_database.py | 336 ++++++++++++--- tests/unit/test_pool.py | 29 ++ tests/unit/test_session.py | 384 ++++++++++++++++-- tests/unit/test_snapshot.py | 162 ++++++-- tests/unit/test_spanner.py | 283 ++++++++++--- tests/unit/test_transaction.py | 55 ++- 13 files changed, 1399 insertions(+), 287 deletions(-) diff --git a/google/cloud/spanner_v1/database.py b/google/cloud/spanner_v1/database.py index 03c6e5119f..c4bdfa7c07 100644 --- a/google/cloud/spanner_v1/database.py +++ b/google/cloud/spanner_v1/database.py @@ -53,6 +53,7 @@ from google.cloud.spanner_v1._helpers import ( _metadata_with_prefix, _metadata_with_leader_aware_routing, + _metadata_with_request_id, ) from google.cloud.spanner_v1.batch import Batch from google.cloud.spanner_v1.batch import MutationGroups @@ -151,6 +152,9 @@ class Database(object): _spanner_api: SpannerClient = None + __transport_lock = threading.Lock() + __transports_to_channel_id = dict() + def __init__( self, database_id, @@ -188,6 +192,7 @@ def __init__( self._instance._client.default_transaction_options ) self._proto_descriptors = proto_descriptors + self._channel_id = 0 # It'll be created when _spanner_api is created. if pool is None: pool = BurstyPool(database_role=database_role) @@ -446,8 +451,26 @@ def spanner_api(self): client_info=client_info, client_options=client_options, ) + + with self.__transport_lock: + transport = self._spanner_api._transport + channel_id = self.__transports_to_channel_id.get(transport, None) + if channel_id is None: + channel_id = len(self.__transports_to_channel_id) + 1 + self.__transports_to_channel_id[transport] = channel_id + self._channel_id = channel_id + return self._spanner_api + def metadata_with_request_id(self, nth_request, nth_attempt, prior_metadata=[]): + return _metadata_with_request_id( + self._nth_client_id, + self._channel_id, + nth_request, + nth_attempt, + prior_metadata, + ) + def __eq__(self, other): if not isinstance(other, self.__class__): return NotImplemented @@ -490,7 +513,10 @@ def create(self): database_dialect=self._database_dialect, proto_descriptors=self._proto_descriptors, ) - future = api.create_database(request=request, metadata=metadata) + future = api.create_database( + request=request, + metadata=self.metadata_with_request_id(self._next_nth_request, 1, metadata), + ) return future def exists(self): @@ -506,7 +532,12 @@ def exists(self): metadata = _metadata_with_prefix(self.name) try: - api.get_database_ddl(database=self.name, metadata=metadata) + api.get_database_ddl( + database=self.name, + metadata=self.metadata_with_request_id( + self._next_nth_request, 1, metadata + ), + ) except NotFound: return False return True @@ -523,10 +554,16 @@ def reload(self): """ api = self._instance._client.database_admin_api metadata = _metadata_with_prefix(self.name) - response = api.get_database_ddl(database=self.name, metadata=metadata) + response = api.get_database_ddl( + database=self.name, + metadata=self.metadata_with_request_id(self._next_nth_request, 1, metadata), + ) self._ddl_statements = tuple(response.statements) self._proto_descriptors = response.proto_descriptors - response = api.get_database(name=self.name, metadata=metadata) + response = api.get_database( + name=self.name, + metadata=self.metadata_with_request_id(self._next_nth_request, 1, metadata), + ) self._state = DatabasePB.State(response.state) self._create_time = response.create_time self._restore_info = response.restore_info @@ -571,7 +608,10 @@ def update_ddl(self, ddl_statements, operation_id="", proto_descriptors=None): proto_descriptors=proto_descriptors, ) - future = api.update_database_ddl(request=request, metadata=metadata) + future = api.update_database_ddl( + request=request, + metadata=self.metadata_with_request_id(self._next_nth_request, 1, metadata), + ) return future def update(self, fields): @@ -609,7 +649,9 @@ def update(self, fields): metadata = _metadata_with_prefix(self.name) future = api.update_database( - database=database_pb, update_mask=field_mask, metadata=metadata + database=database_pb, + update_mask=field_mask, + metadata=self.metadata_with_request_id(self._next_nth_request, 1, metadata), ) return future @@ -622,7 +664,10 @@ def drop(self): """ api = self._instance._client.database_admin_api metadata = _metadata_with_prefix(self.name) - api.drop_database(database=self.name, metadata=metadata) + api.drop_database( + database=self.name, + metadata=self.metadata_with_request_id(self._next_nth_request, 1, metadata), + ) def execute_partitioned_dml( self, @@ -711,7 +756,13 @@ def execute_pdml(): with SessionCheckout(self._pool) as session: add_span_event(span, "Starting BeginTransaction") txn = api.begin_transaction( - session=session.name, options=txn_options, metadata=metadata + session=session.name, + options=txn_options, + metadata=self.metadata_with_request_id( + self._next_nth_request, + 1, + metadata, + ), ) txn_selector = TransactionSelector(id=txn.id) @@ -724,6 +775,7 @@ def execute_pdml(): query_options=query_options, request_options=request_options, ) + method = functools.partial( api.execute_streaming_sql, metadata=metadata, @@ -736,6 +788,7 @@ def execute_pdml(): metadata=metadata, transaction_selector=txn_selector, observability_options=self.observability_options, + request_id_manager=self, ) result_set = StreamedResultSet(iterator) @@ -745,6 +798,16 @@ def execute_pdml(): return _retry_on_aborted(execute_pdml, DEFAULT_RETRY_BACKOFF)() + @property + def _next_nth_request(self): + if self._instance and self._instance._client: + return self._instance._client._next_nth_request + return 1 + + @property + def _nth_client_id(self): + return self._instance._client._nth_client_id + def session(self, labels=None, database_role=None): """Factory to create a session for this database. @@ -965,7 +1028,8 @@ def restore(self, source): ) future = api.restore_database( request=request, - metadata=metadata, + # TODO: Infer the channel_id being used. + metadata=self.metadata_with_request_id(self._next_nth_request, 1, metadata), ) return future @@ -1034,7 +1098,10 @@ def list_database_roles(self, page_size=None): parent=self.name, page_size=page_size, ) - return api.list_database_roles(request=request, metadata=metadata) + return api.list_database_roles( + request=request, + metadata=self.metadata_with_request_id(self._next_nth_request, 1, metadata), + ) def table(self, table_id): """Factory to create a table object within this database. @@ -1118,7 +1185,10 @@ def get_iam_policy(self, policy_version=None): requested_policy_version=policy_version ), ) - response = api.get_iam_policy(request=request, metadata=metadata) + response = api.get_iam_policy( + request=request, + metadata=self.metadata_with_request_id(self._next_nth_request, 1, metadata), + ) return response def set_iam_policy(self, policy): @@ -1140,7 +1210,10 @@ def set_iam_policy(self, policy): resource=self.name, policy=policy, ) - response = api.set_iam_policy(request=request, metadata=metadata) + response = api.set_iam_policy( + request=request, + metadata=self.metadata_with_request_id(self._next_nth_request, 1, metadata), + ) return response @property diff --git a/google/cloud/spanner_v1/pool.py b/google/cloud/spanner_v1/pool.py index 0c4dd5a63b..0bc0135ba0 100644 --- a/google/cloud/spanner_v1/pool.py +++ b/google/cloud/spanner_v1/pool.py @@ -256,7 +256,9 @@ def bind(self, database): ) resp = api.batch_create_sessions( request=request, - metadata=metadata, + metadata=database.metadata_with_request_id( + database._next_nth_request, 1, metadata + ), ) add_span_event( @@ -561,7 +563,9 @@ def bind(self, database): while returned_session_count < self.size: resp = api.batch_create_sessions( request=request, - metadata=metadata, + metadata=database.metadata_with_request_id( + database._next_nth_request, 1, metadata + ), ) add_span_event( diff --git a/google/cloud/spanner_v1/session.py b/google/cloud/spanner_v1/session.py index d5feb2ef1a..2e3492892b 100644 --- a/google/cloud/spanner_v1/session.py +++ b/google/cloud/spanner_v1/session.py @@ -170,7 +170,9 @@ def create(self): ), MetricsCapture(): session_pb = api.create_session( request=request, - metadata=metadata, + metadata=self._database.metadata_with_request_id( + self._database._next_nth_request, 1, metadata + ), ) self._session_id = session_pb.name.split("/")[-1] @@ -195,7 +197,8 @@ def exists(self): current_span, "Checking if Session exists", {"session.id": self._session_id} ) - api = self._database.spanner_api + database = self._database + api = database.spanner_api metadata = _metadata_with_prefix(self._database.name) if self._database._route_to_leader_enabled: metadata.append( @@ -212,7 +215,12 @@ def exists(self): metadata=metadata, ) as span, MetricsCapture(): try: - api.get_session(name=self.name, metadata=metadata) + api.get_session( + name=self.name, + metadata=database.metadata_with_request_id( + database._next_nth_request, 1, metadata + ), + ) if span: span.set_attribute("session_found", True) except NotFound: @@ -242,8 +250,11 @@ def delete(self): current_span, "Deleting Session", {"session.id": self._session_id} ) - api = self._database.spanner_api - metadata = _metadata_with_prefix(self._database.name) + database = self._database + api = database.spanner_api + metadata = database.metadata_with_request_id( + database._next_nth_request, 1, _metadata_with_prefix(database.name) + ) observability_options = getattr(self._database, "observability_options", None) with trace_call( "CloudSpanner.DeleteSession", @@ -255,7 +266,10 @@ def delete(self): observability_options=observability_options, metadata=metadata, ), MetricsCapture(): - api.delete_session(name=self.name, metadata=metadata) + api.delete_session( + name=self.name, + metadata=metadata, + ) def ping(self): """Ping the session to keep it alive by executing "SELECT 1". @@ -264,10 +278,17 @@ def ping(self): """ if self._session_id is None: raise ValueError("Session ID not set by back-end") - api = self._database.spanner_api - metadata = _metadata_with_prefix(self._database.name) + database = self._database + api = database.spanner_api request = ExecuteSqlRequest(session=self.name, sql="SELECT 1") - api.execute_sql(request=request, metadata=metadata) + api.execute_sql( + request=request, + metadata=database.metadata_with_request_id( + database._next_nth_request, + 1, + _metadata_with_prefix(database.name), + ), + ) self._last_use_time = datetime.now() def snapshot(self, **kw): diff --git a/google/cloud/spanner_v1/snapshot.py b/google/cloud/spanner_v1/snapshot.py index 3b18d2c855..e00a253232 100644 --- a/google/cloud/spanner_v1/snapshot.py +++ b/google/cloud/spanner_v1/snapshot.py @@ -61,6 +61,7 @@ def _restart_on_unavailable( transaction=None, transaction_selector=None, observability_options=None, + request_id_manager=None, ): """Restart iteration after :exc:`.ServiceUnavailable`. @@ -81,6 +82,11 @@ def _restart_on_unavailable( resume_token = b"" item_buffer = [] + def next_nth_request(): + return getattr(request_id_manager, "_next_nth_request", 0) + + nth_request = next_nth_request() + if transaction is not None: transaction_selector = transaction._make_txn_selector() elif transaction_selector is None: @@ -90,6 +96,7 @@ def _restart_on_unavailable( request.transaction = transaction_selector iterator = None + attempt = 0 while True: try: @@ -101,7 +108,13 @@ def _restart_on_unavailable( observability_options=observability_options, metadata=metadata, ), MetricsCapture(): - iterator = method(request=request, metadata=metadata) + attempt += 1 + iterator = method( + request=request, + metadata=request_id_manager.metadata_with_request_id( + nth_request, attempt, metadata + ), + ) for item in iterator: item_buffer.append(item) # Setting the transaction id because the transaction begin was inlined for first rpc. @@ -129,7 +142,14 @@ def _restart_on_unavailable( if transaction is not None: transaction_selector = transaction._make_txn_selector() request.transaction = transaction_selector - iterator = method(request=request) + nth_request = next_nth_request() + attempt = 1 + iterator = method( + request=request, + metadata=request_id_manager.metadata_with_request_id( + nth_request, attempt, metadata + ), + ) continue except InternalServerError as exc: resumable_error = any( @@ -149,8 +169,15 @@ def _restart_on_unavailable( request.resume_token = resume_token if transaction is not None: transaction_selector = transaction._make_txn_selector() + nth_request = next_nth_request() + attempt = 1 request.transaction = transaction_selector - iterator = method(request=request) + iterator = method( + request=request, + metadata=request_id_manager.metadata_with_request_id( + nth_request, attempt, metadata + ), + ) continue if len(item_buffer) == 0: @@ -329,6 +356,7 @@ def read( data_boost_enabled=data_boost_enabled, directed_read_options=directed_read_options, ) + restart = functools.partial( api.streaming_read, request=request, @@ -352,6 +380,7 @@ def read( trace_attributes, transaction=self, observability_options=observability_options, + request_id_manager=self._session._database, ) self._read_request_count += 1 if self._multi_use: @@ -375,6 +404,7 @@ def read( trace_attributes, transaction=self, observability_options=observability_options, + request_id_manager=self._session._database, ) self._read_request_count += 1 @@ -562,13 +592,16 @@ def execute_sql( data_boost_enabled=data_boost_enabled, directed_read_options=directed_read_options, ) - restart = functools.partial( - api.execute_streaming_sql, - request=request, - metadata=metadata, - retry=retry, - timeout=timeout, - ) + + def wrapped_restart(*args, **kwargs): + restart = functools.partial( + api.execute_streaming_sql, + request=request, + metadata=kwargs.get("metadata", metadata), + retry=retry, + timeout=timeout, + ) + return restart(*args, **kwargs) trace_attributes = {"db.statement": sql} observability_options = getattr(database, "observability_options", None) @@ -577,7 +610,7 @@ def execute_sql( # lock is added to handle the inline begin for first rpc with self._lock: return self._get_streamed_result_set( - restart, + wrapped_restart, request, metadata, trace_attributes, @@ -587,7 +620,7 @@ def execute_sql( ) else: return self._get_streamed_result_set( - restart, + wrapped_restart, request, metadata, trace_attributes, @@ -615,6 +648,7 @@ def _get_streamed_result_set( trace_attributes, transaction=self, observability_options=observability_options, + request_id_manager=self._session._database, ) self._read_request_count += 1 self._execute_sql_count += 1 @@ -718,15 +752,25 @@ def partition_read( observability_options=getattr(database, "observability_options", None), metadata=metadata, ), MetricsCapture(): - method = functools.partial( - api.partition_read, - request=request, - metadata=metadata, - retry=retry, - timeout=timeout, - ) + nth_request = getattr(database, "_next_nth_request", 0) + counters = dict(attempt=0) + + def attempt_tracking_method(): + counters["attempt"] += 1 + all_metadata = database.metadata_with_request_id( + nth_request, counters["attempt"], metadata + ) + method = functools.partial( + api.partition_read, + request=request, + metadata=all_metadata, + retry=retry, + timeout=timeout, + ) + return method() + response = _retry( - method, + attempt_tracking_method, allowed_exceptions={InternalServerError: _check_rst_stream_error}, ) @@ -822,15 +866,25 @@ def partition_query( observability_options=getattr(database, "observability_options", None), metadata=metadata, ), MetricsCapture(): - method = functools.partial( - api.partition_query, - request=request, - metadata=metadata, - retry=retry, - timeout=timeout, - ) + nth_request = getattr(database, "_next_nth_request", 0) + counters = dict(attempt=0) + + def attempt_tracking_method(): + counters["attempt"] += 1 + all_metadata = database.metadata_with_request_id( + nth_request, counters["attempt"], metadata + ) + method = functools.partial( + api.partition_query, + request=request, + metadata=all_metadata, + retry=retry, + timeout=timeout, + ) + return method() + response = _retry( - method, + attempt_tracking_method, allowed_exceptions={InternalServerError: _check_rst_stream_error}, ) @@ -969,14 +1023,24 @@ def begin(self): observability_options=getattr(database, "observability_options", None), metadata=metadata, ), MetricsCapture(): - method = functools.partial( - api.begin_transaction, - session=self._session.name, - options=txn_selector.begin, - metadata=metadata, - ) + nth_request = getattr(database, "_next_nth_request", 0) + counters = dict(attempt=0) + + def attempt_tracking_method(): + counters["attempt"] += 1 + all_metadata = database.metadata_with_request_id( + nth_request, counters["attempt"], metadata + ) + method = functools.partial( + api.begin_transaction, + session=self._session.name, + options=txn_selector.begin, + metadata=all_metadata, + ) + return method() + response = _retry( - method, + attempt_tracking_method, allowed_exceptions={InternalServerError: _check_rst_stream_error}, ) self._transaction_id = response.id diff --git a/google/cloud/spanner_v1/testing/interceptors.py b/google/cloud/spanner_v1/testing/interceptors.py index bf5e271e26..71b77e4d16 100644 --- a/google/cloud/spanner_v1/testing/interceptors.py +++ b/google/cloud/spanner_v1/testing/interceptors.py @@ -72,7 +72,7 @@ def reset(self): class XGoogRequestIDHeaderInterceptor(ClientInterceptor): # TODO:(@odeke-em): delete this guard when PR #1367 is merged. - X_GOOG_REQUEST_ID_FUNCTIONALITY_MERGED = False + X_GOOG_REQUEST_ID_FUNCTIONALITY_MERGED = True def __init__(self): self._unary_req_segments = [] diff --git a/google/cloud/spanner_v1/transaction.py b/google/cloud/spanner_v1/transaction.py index 2f52aaa144..8adbf8e108 100644 --- a/google/cloud/spanner_v1/transaction.py +++ b/google/cloud/spanner_v1/transaction.py @@ -32,6 +32,7 @@ from google.cloud.spanner_v1 import ExecuteSqlRequest from google.cloud.spanner_v1 import TransactionSelector from google.cloud.spanner_v1 import TransactionOptions +from google.cloud.spanner_v1._helpers import AtomicCounter from google.cloud.spanner_v1.snapshot import _SnapshotBase from google.cloud.spanner_v1.batch import _BatchBase from google.cloud.spanner_v1._opentelemetry_tracing import add_span_event, trace_call @@ -181,12 +182,20 @@ def begin(self): observability_options=observability_options, metadata=metadata, ) as span, MetricsCapture(): - method = functools.partial( - api.begin_transaction, - session=self._session.name, - options=txn_options, - metadata=metadata, - ) + attempt = AtomicCounter(0) + nth_request = database._next_nth_request + + def wrapped_method(*args, **kwargs): + attempt.increment() + method = functools.partial( + api.begin_transaction, + session=self._session.name, + options=txn_options, + metadata=database.metadata_with_request_id( + nth_request, attempt.value, metadata + ), + ) + return method(*args, **kwargs) def beforeNextRetry(nthRetry, delayInSeconds): add_span_event( @@ -196,7 +205,7 @@ def beforeNextRetry(nthRetry, delayInSeconds): ) response = _retry( - method, + wrapped_method, allowed_exceptions={InternalServerError: _check_rst_stream_error}, beforeNextRetry=beforeNextRetry, ) @@ -217,6 +226,7 @@ def rollback(self): database._route_to_leader_enabled ) ) + observability_options = getattr(database, "observability_options", None) with trace_call( f"CloudSpanner.{type(self).__name__}.rollback", @@ -224,16 +234,26 @@ def rollback(self): observability_options=observability_options, metadata=metadata, ), MetricsCapture(): - method = functools.partial( - api.rollback, - session=self._session.name, - transaction_id=self._transaction_id, - metadata=metadata, - ) + attempt = AtomicCounter(0) + nth_request = database._next_nth_request + + def wrapped_method(*args, **kwargs): + attempt.increment() + method = functools.partial( + api.rollback, + session=self._session.name, + transaction_id=self._transaction_id, + metadata=database.metadata_with_request_id( + nth_request, attempt.value, metadata + ), + ) + return method(*args, **kwargs) + _retry( - method, + wrapped_method, allowed_exceptions={InternalServerError: _check_rst_stream_error}, ) + self.rolled_back = True del self._session._transaction @@ -306,11 +326,19 @@ def commit( add_span_event(span, "Starting Commit") - method = functools.partial( - api.commit, - request=request, - metadata=metadata, - ) + attempt = AtomicCounter(0) + nth_request = database._next_nth_request + + def wrapped_method(*args, **kwargs): + attempt.increment() + method = functools.partial( + api.commit, + request=request, + metadata=database.metadata_with_request_id( + nth_request, attempt.value, metadata + ), + ) + return method(*args, **kwargs) def beforeNextRetry(nthRetry, delayInSeconds): add_span_event( @@ -320,7 +348,7 @@ def beforeNextRetry(nthRetry, delayInSeconds): ) response = _retry( - method, + wrapped_method, allowed_exceptions={InternalServerError: _check_rst_stream_error}, beforeNextRetry=beforeNextRetry, ) @@ -469,19 +497,27 @@ def execute_update( last_statement=last_statement, ) - method = functools.partial( - api.execute_sql, - request=request, - metadata=metadata, - retry=retry, - timeout=timeout, - ) + nth_request = database._next_nth_request + attempt = AtomicCounter(0) + + def wrapped_method(*args, **kwargs): + attempt.increment() + method = functools.partial( + api.execute_sql, + request=request, + metadata=database.metadata_with_request_id( + nth_request, attempt.value, metadata + ), + retry=retry, + timeout=timeout, + ) + return method(*args, **kwargs) if self._transaction_id is None: # lock is added to handle the inline begin for first rpc with self._lock: response = self._execute_request( - method, + wrapped_method, request, metadata, f"CloudSpanner.{type(self).__name__}.execute_update", @@ -499,7 +535,7 @@ def execute_update( self._transaction_id = response.metadata.transaction.id else: response = self._execute_request( - method, + wrapped_method, request, metadata, f"CloudSpanner.{type(self).__name__}.execute_update", @@ -611,19 +647,27 @@ def batch_update( last_statements=last_statement, ) - method = functools.partial( - api.execute_batch_dml, - request=request, - metadata=metadata, - retry=retry, - timeout=timeout, - ) + nth_request = database._next_nth_request + attempt = AtomicCounter(0) + + def wrapped_method(*args, **kwargs): + attempt.increment() + method = functools.partial( + api.execute_batch_dml, + request=request, + metadata=database.metadata_with_request_id( + nth_request, attempt.value, metadata + ), + retry=retry, + timeout=timeout, + ) + return method(*args, **kwargs) if self._transaction_id is None: # lock is added to handle the inline begin for first rpc with self._lock: response = self._execute_request( - method, + wrapped_method, request, metadata, "CloudSpanner.DMLTransaction", @@ -642,7 +686,7 @@ def batch_update( break else: response = self._execute_request( - method, + wrapped_method, request, metadata, "CloudSpanner.DMLTransaction", diff --git a/tests/unit/test_batch.py b/tests/unit/test_batch.py index 355ce20520..33130dd98d 100644 --- a/tests/unit/test_batch.py +++ b/tests/unit/test_batch.py @@ -37,6 +37,9 @@ from google.cloud.spanner_v1.keyset import KeySet from google.rpc.status_pb2 import Status +from google.cloud.spanner_v1._helpers import ( + _metadata_with_request_id, +) TABLE_NAME = "citizens" COLUMNS = ["email", "first_name", "last_name", "age"] @@ -579,10 +582,17 @@ def _test_batch_write_with_request_options( "traceparent is missing in metadata", ) + # expected_metadata.append( + # ( + # "x-goog-spanner-request-id", + # f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + # ) + # ) # Remove traceparent from actual metadata for comparison filtered_metadata = [item for item in metadata if item[0] != "traceparent"] self.assertEqual(filtered_metadata, expected_metadata) + if request_options is None: expected_request_options = RequestOptions() elif type(request_options) is dict: @@ -635,12 +645,39 @@ def session_id(self): class _Database(object): + name = "testing" + _route_to_leader_enabled = True + NTH_CLIENT = 1 + def __init__(self, enable_end_to_end_tracing=False): self.name = "testing" self._route_to_leader_enabled = True if enable_end_to_end_tracing: self.observability_options = dict(enable_end_to_end_tracing=True) self.default_transaction_options = DefaultTransactionOptions() + self._nth_request = 0 + + @property + def _next_nth_request(self): + self._nth_request += 1 + return self._nth_request + + @property + def _nth_client_id(self): + return 1 + + def metadata_with_request_id(self, nth_request, nth_attempt, prior_metadata=[]): + return _metadata_with_request_id( + self._nth_client_id, + self._channel_id, + nth_request, + nth_attempt, + prior_metadata, + ) + + @property + def _channel_id(self): + return 1 class _FauxSpannerAPI: diff --git a/tests/unit/test_database.py b/tests/unit/test_database.py index c7ed5a0e3d..17f9c78e6c 100644 --- a/tests/unit/test_database.py +++ b/tests/unit/test_database.py @@ -30,6 +30,11 @@ DirectedReadOptions, DefaultTransactionOptions, ) +from google.cloud.spanner_v1._helpers import ( + AtomicCounter, + _metadata_with_request_id, +) +from google.cloud.spanner_v1.request_id_header import REQ_RAND_PROCESS_ID DML_WO_PARAM = """ DELETE FROM citizens @@ -549,7 +554,13 @@ def test_create_grpc_error(self): api.create_database.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_create_already_exists(self): @@ -576,7 +587,13 @@ def test_create_already_exists(self): api.create_database.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_create_instance_not_found(self): @@ -602,7 +619,13 @@ def test_create_instance_not_found(self): api.create_database.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_create_success(self): @@ -638,7 +661,13 @@ def test_create_success(self): api.create_database.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_create_success_w_encryption_config_dict(self): @@ -675,7 +704,13 @@ def test_create_success_w_encryption_config_dict(self): api.create_database.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_create_success_w_proto_descriptors(self): @@ -710,7 +745,13 @@ def test_create_success_w_proto_descriptors(self): api.create_database.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_exists_grpc_error(self): @@ -728,7 +769,13 @@ def test_exists_grpc_error(self): api.get_database_ddl.assert_called_once_with( database=self.DATABASE_NAME, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_exists_not_found(self): @@ -745,7 +792,13 @@ def test_exists_not_found(self): api.get_database_ddl.assert_called_once_with( database=self.DATABASE_NAME, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_exists_success(self): @@ -764,7 +817,13 @@ def test_exists_success(self): api.get_database_ddl.assert_called_once_with( database=self.DATABASE_NAME, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_reload_grpc_error(self): @@ -782,7 +841,13 @@ def test_reload_grpc_error(self): api.get_database_ddl.assert_called_once_with( database=self.DATABASE_NAME, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_reload_not_found(self): @@ -800,7 +865,13 @@ def test_reload_not_found(self): api.get_database_ddl.assert_called_once_with( database=self.DATABASE_NAME, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_reload_success(self): @@ -859,11 +930,23 @@ def test_reload_success(self): api.get_database_ddl.assert_called_once_with( database=self.DATABASE_NAME, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) api.get_database.assert_called_once_with( name=self.DATABASE_NAME, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), + ], ) def test_update_ddl_grpc_error(self): @@ -889,7 +972,13 @@ def test_update_ddl_grpc_error(self): api.update_database_ddl.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_update_ddl_not_found(self): @@ -915,7 +1004,13 @@ def test_update_ddl_not_found(self): api.update_database_ddl.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_update_ddl(self): @@ -942,7 +1037,13 @@ def test_update_ddl(self): api.update_database_ddl.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_update_ddl_w_operation_id(self): @@ -969,7 +1070,13 @@ def test_update_ddl_w_operation_id(self): api.update_database_ddl.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_update_success(self): @@ -995,7 +1102,13 @@ def test_update_success(self): api.update_database.assert_called_once_with( database=expected_database, update_mask=field_mask, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_update_ddl_w_proto_descriptors(self): @@ -1023,7 +1136,13 @@ def test_update_ddl_w_proto_descriptors(self): api.update_database_ddl.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_drop_grpc_error(self): @@ -1041,7 +1160,13 @@ def test_drop_grpc_error(self): api.drop_database.assert_called_once_with( database=self.DATABASE_NAME, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_drop_not_found(self): @@ -1059,7 +1184,13 @@ def test_drop_not_found(self): api.drop_database.assert_called_once_with( database=self.DATABASE_NAME, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_drop_success(self): @@ -1076,7 +1207,13 @@ def test_drop_success(self): api.drop_database.assert_called_once_with( database=self.DATABASE_NAME, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def _execute_partitioned_dml_helper( @@ -1149,17 +1286,33 @@ def _execute_partitioned_dml_helper( exclude_txn_from_change_streams=exclude_txn_from_change_streams, ) - api.begin_transaction.assert_called_with( - session=session.name, - options=txn_options, - metadata=[ - ("google-cloud-resource-prefix", database.name), - ("x-goog-spanner-route-to-leader", "true"), - ], - ) if retried: + api.begin_transaction.assert_called_with( + session=session.name, + options=txn_options, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.3.1", + ), + ], + ) self.assertEqual(api.begin_transaction.call_count, 2) else: + api.begin_transaction.assert_called_with( + session=session.name, + options=txn_options, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], + ) self.assertEqual(api.begin_transaction.call_count, 1) if params: @@ -1191,18 +1344,11 @@ def _execute_partitioned_dml_helper( request_options=expected_request_options, ) - api.execute_streaming_sql.assert_any_call( - request=expected_request, - metadata=[ - ("google-cloud-resource-prefix", database.name), - ("x-goog-spanner-route-to-leader", "true"), - ], - ) if retried: expected_retry_transaction = TransactionSelector( id=self.RETRY_TRANSACTION_ID ) - expected_request = ExecuteSqlRequest( + expected_request_with_retry = ExecuteSqlRequest( session=self.SESSION_NAME, sql=dml, transaction=expected_retry_transaction, @@ -1211,15 +1357,47 @@ def _execute_partitioned_dml_helper( query_options=expected_query_options, request_options=expected_request_options, ) - api.execute_streaming_sql.assert_called_with( + + self.assertEqual( + api.execute_streaming_sql.call_args_list, + [ + mock.call( + request=expected_request, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), + ], + ), + mock.call( + request=expected_request_with_retry, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.4.1", + ), + ], + ), + ], + ) + self.assertEqual(api.execute_streaming_sql.call_count, 2) + else: + api.execute_streaming_sql.assert_any_call( request=expected_request, metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), ], ) - self.assertEqual(api.execute_streaming_sql.call_count, 2) - else: self.assertEqual(api.execute_streaming_sql.call_count, 1) def test_execute_partitioned_dml_wo_params(self): @@ -1490,7 +1668,13 @@ def test_restore_grpc_error(self): api.restore_database.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_restore_not_found(self): @@ -1516,7 +1700,13 @@ def test_restore_not_found(self): api.restore_database.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_restore_success(self): @@ -1553,7 +1743,13 @@ def test_restore_success(self): api.restore_database.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_restore_success_w_encryption_config_dict(self): @@ -1594,7 +1790,13 @@ def test_restore_success_w_encryption_config_dict(self): api.restore_database.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_restore_w_invalid_encryption_config_dict(self): @@ -1741,7 +1943,13 @@ def test_list_database_roles_grpc_error(self): api.list_database_roles.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_list_database_roles_defaults(self): @@ -1762,7 +1970,13 @@ def test_list_database_roles_defaults(self): api.list_database_roles.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) self.assertIsNotNone(resp) @@ -3113,6 +3327,8 @@ def _make_database_admin_api(): class _Client(object): + NTH_CLIENT = AtomicCounter() + def __init__( self, project=TestDatabase.PROJECT_ID, @@ -3135,6 +3351,12 @@ def __init__( self.directed_read_options = directed_read_options self.default_transaction_options = default_transaction_options self.observability_options = observability_options + self._nth_client_id = _Client.NTH_CLIENT.increment() + self._nth_request = AtomicCounter() + + @property + def _next_nth_request(self): + return self._nth_request.increment() class _Instance(object): @@ -3153,6 +3375,7 @@ def __init__(self, name): class _Database(object): log_commit_stats = False _route_to_leader_enabled = True + NTH_CLIENT_ID = AtomicCounter() def __init__(self, name, instance=None): self.name = name @@ -3163,6 +3386,25 @@ def __init__(self, name, instance=None): self.logger = mock.create_autospec(Logger, instance=True) self._directed_read_options = None self.default_transaction_options = DefaultTransactionOptions() + self._nth_request = AtomicCounter() + self._nth_client_id = _Database.NTH_CLIENT_ID.increment() + + @property + def _next_nth_request(self): + return self._nth_request.increment() + + def metadata_with_request_id(self, nth_request, nth_attempt, prior_metadata=[]): + return _metadata_with_request_id( + self._nth_client_id, + self._channel_id, + nth_request, + nth_attempt, + prior_metadata, + ) + + @property + def _channel_id(self): + return 1 class _Pool(object): diff --git a/tests/unit/test_pool.py b/tests/unit/test_pool.py index 768f8482f3..8069f806d8 100644 --- a/tests/unit/test_pool.py +++ b/tests/unit/test_pool.py @@ -19,6 +19,11 @@ from datetime import datetime, timedelta import mock +from google.cloud.spanner_v1._helpers import ( + _metadata_with_request_id, + AtomicCounter, +) + from google.cloud.spanner_v1._opentelemetry_tracing import trace_call from tests._helpers import ( OpenTelemetryBase, @@ -1193,6 +1198,9 @@ def session_id(self): class _Database(object): + NTH_REQUEST = AtomicCounter() + NTH_CLIENT_ID = AtomicCounter() + def __init__(self, name): self.name = name self._sessions = [] @@ -1247,6 +1255,27 @@ def session(self, **kwargs): def observability_options(self): return dict(db_name=self.name) + @property + def _next_nth_request(self): + return self.NTH_REQUEST.increment() + + @property + def _nth_client_id(self): + return self.NTH_CLIENT_ID.increment() + + def metadata_with_request_id(self, nth_request, nth_attempt, prior_metadata=[]): + return _metadata_with_request_id( + self._nth_client_id, + self._channel_id, + nth_request, + nth_attempt, + prior_metadata, + ) + + @property + def _channel_id(self): + return 1 + class _Queue(object): _size = 1 diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index d72c01f5ab..b80d6bd18a 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -49,6 +49,11 @@ from google.protobuf.struct_pb2 import Struct, Value from google.cloud.spanner_v1.batch import Batch from google.cloud.spanner_v1 import DefaultTransactionOptions +from google.cloud.spanner_v1.request_id_header import REQ_RAND_PROCESS_ID +from google.cloud.spanner_v1._helpers import ( + AtomicCounter, + _metadata_with_request_id, +) def _make_rpc_error(error_cls, trailing_metadata=None): @@ -57,6 +62,37 @@ def _make_rpc_error(error_cls, trailing_metadata=None): return error_cls("error", errors=(grpc_error,)) +NTH_CLIENT_ID = AtomicCounter() + + +def inject_into_mock_database(mockdb): + setattr(mockdb, "_nth_request", AtomicCounter()) + nth_client_id = NTH_CLIENT_ID.increment() + setattr(mockdb, "_nth_client_id", nth_client_id) + channel_id = 1 + setattr(mockdb, "_channel_id", channel_id) + + def metadata_with_request_id(nth_request, nth_attempt, prior_metadata=[]): + nth_req = nth_request.fget(mockdb) + return _metadata_with_request_id( + nth_client_id, + channel_id, + nth_req, + nth_attempt, + prior_metadata, + ) + + setattr(mockdb, "metadata_with_request_id", metadata_with_request_id) + + @property + def _next_nth_request(self): + return self._nth_request.increment() + + setattr(mockdb, "_next_nth_request", _next_nth_request) + + return mockdb + + class TestSession(OpenTelemetryBase): PROJECT_ID = "project-id" INSTANCE_ID = "instance-id" @@ -95,6 +131,7 @@ def _make_database( database.database_role = database_role database._route_to_leader_enabled = True database.default_transaction_options = default_transaction_options + inject_into_mock_database(database) return database @@ -191,6 +228,10 @@ def test_create_w_database_role(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -226,6 +267,10 @@ def test_create_session_span_annotations(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -253,6 +298,10 @@ def test_create_wo_database_role(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -281,6 +330,10 @@ def test_create_ok(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -311,6 +364,10 @@ def test_create_w_labels(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -358,6 +415,10 @@ def test_exists_hit(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -386,6 +447,10 @@ def test_exists_hit_wo_span(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -406,6 +471,10 @@ def test_exists_miss(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -433,6 +502,10 @@ def test_exists_miss_wo_span(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -454,6 +527,10 @@ def test_exists_error(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -486,7 +563,13 @@ def test_ping_hit(self): gax_api.execute_sql.assert_called_once_with( request=request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_ping_miss(self): @@ -507,7 +590,13 @@ def test_ping_miss(self): gax_api.execute_sql.assert_called_once_with( request=request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_ping_error(self): @@ -528,7 +617,13 @@ def test_ping_error(self): gax_api.execute_sql.assert_called_once_with( request=request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) def test_delete_wo_session_id(self): @@ -552,7 +647,13 @@ def test_delete_hit(self): gax_api.delete_session.assert_called_once_with( name=self.SESSION_NAME, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) attrs = {"session.id": session._session_id, "session.name": session.name} @@ -575,7 +676,13 @@ def test_delete_miss(self): gax_api.delete_session.assert_called_once_with( name=self.SESSION_NAME, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) attrs = {"session.id": session._session_id, "session.name": session.name} @@ -600,7 +707,13 @@ def test_delete_error(self): gax_api.delete_session.assert_called_once_with( name=self.SESSION_NAME, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) attrs = {"session.id": session._session_id, "session.name": session.name} @@ -936,6 +1049,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) request = CommitRequest( @@ -949,6 +1066,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), ], ) @@ -1000,6 +1121,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -1052,10 +1177,25 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], - ) - ] - * 2, + ), + mock.call( + session=self.SESSION_NAME, + options=expected_options, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.3.1", + ), + ], + ), + ], ) request = CommitRequest( session=self.SESSION_NAME, @@ -1071,10 +1211,24 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), ], - ) - ] - * 2, + ), + mock.call( + request=request, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.4.1", + ), + ], + ), + ], ) def test_run_in_transaction_w_abort_w_retry_metadata(self): @@ -1137,10 +1291,25 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], + ), + mock.call( + session=self.SESSION_NAME, + options=expected_options, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.3.1", + ), ], - ) - ] - * 2, + ), + ], ) request = CommitRequest( session=self.SESSION_NAME, @@ -1156,10 +1325,24 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), + ], + ), + mock.call( + request=request, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.4.1", + ), ], - ) - ] - * 2, + ), + ], ) def test_run_in_transaction_w_callback_raises_abort_wo_metadata(self): @@ -1221,6 +1404,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) request = CommitRequest( @@ -1234,6 +1421,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), ], ) @@ -1297,6 +1488,10 @@ def _time(_results=[1, 1.5]): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) request = CommitRequest( @@ -1310,6 +1505,10 @@ def _time(_results=[1, 1.5]): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), ], ) @@ -1369,10 +1568,37 @@ def _time(_results=[1, 2, 4, 8]): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], + ), + mock.call( + session=self.SESSION_NAME, + options=expected_options, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.3.1", + ), ], - ) - ] - * 3, + ), + mock.call( + session=self.SESSION_NAME, + options=expected_options, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.5.1", + ), + ], + ), + ], ) request = CommitRequest( session=self.SESSION_NAME, @@ -1388,10 +1614,35 @@ def _time(_results=[1, 2, 4, 8]): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), ], - ) - ] - * 3, + ), + mock.call( + request=request, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.4.1", + ), + ], + ), + mock.call( + request=request, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.6.1", + ), + ], + ), + ], ) def test_run_in_transaction_w_commit_stats_success(self): @@ -1440,6 +1691,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) request = CommitRequest( @@ -1454,6 +1709,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), ], ) database.logger.info.assert_called_once_with( @@ -1502,6 +1761,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) request = CommitRequest( @@ -1516,6 +1779,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), ], ) database.logger.info.assert_not_called() @@ -1568,6 +1835,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) request = CommitRequest( @@ -1581,6 +1852,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), ], ) @@ -1633,6 +1908,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) request = CommitRequest( @@ -1646,6 +1925,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), ], ) @@ -1719,10 +2002,25 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], - ) - ] - * 2, + ), + mock.call( + session=self.SESSION_NAME, + options=expected_options, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.3.1", + ), + ], + ), + ], ) request = CommitRequest( session=self.SESSION_NAME, @@ -1738,10 +2036,24 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), + ], + ), + mock.call( + request=request, + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.4.1", + ), ], - ) - ] - * 2, + ), + ], ) def test_run_in_transaction_w_isolation_level_at_request(self): @@ -1773,6 +2085,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -1807,6 +2123,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -1845,6 +2165,10 @@ def unit_of_work(txn, *args, **kw): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) diff --git a/tests/unit/test_snapshot.py b/tests/unit/test_snapshot.py index 11fc0135d1..7b3ad679a9 100644 --- a/tests/unit/test_snapshot.py +++ b/tests/unit/test_snapshot.py @@ -24,7 +24,11 @@ HAS_OPENTELEMETRY_INSTALLED, enrich_with_otel_scope, ) +from google.cloud.spanner_v1._helpers import ( + _metadata_with_request_id, +) from google.cloud.spanner_v1.param_types import INT64 +from google.cloud.spanner_v1.request_id_header import REQ_RAND_PROCESS_ID from google.api_core.retry import Retry TABLE_NAME = "citizens" @@ -135,6 +139,7 @@ def _call_fut( session, attributes, transaction=derived, + request_id_manager=None if not session else session._database, ) def _make_item(self, value, resume_token=b"", metadata=None): @@ -153,9 +158,17 @@ def test_iteration_w_empty_raw(self): database.spanner_api = self._make_spanner_api() session = _Session(database) derived = self._makeDerived(session) - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) self.assertEqual(list(resumable), []) - restart.assert_called_once_with(request=request, metadata=None) + restart.assert_called_once_with( + request=request, + metadata=[ + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ) + ], + ) self.assertNoSpans() def test_iteration_w_non_empty_raw(self): @@ -167,9 +180,17 @@ def test_iteration_w_non_empty_raw(self): database.spanner_api = self._make_spanner_api() session = _Session(database) derived = self._makeDerived(session) - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) self.assertEqual(list(resumable), list(ITEMS)) - restart.assert_called_once_with(request=request, metadata=None) + restart.assert_called_once_with( + request=request, + metadata=[ + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ) + ], + ) self.assertNoSpans() def test_iteration_w_raw_w_resume_tken(self): @@ -186,9 +207,17 @@ def test_iteration_w_raw_w_resume_tken(self): database.spanner_api = self._make_spanner_api() session = _Session(database) derived = self._makeDerived(session) - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) self.assertEqual(list(resumable), list(ITEMS)) - restart.assert_called_once_with(request=request, metadata=None) + restart.assert_called_once_with( + request=request, + metadata=[ + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ) + ], + ) self.assertNoSpans() def test_iteration_w_raw_raising_unavailable_no_token(self): @@ -207,7 +236,7 @@ def test_iteration_w_raw_raising_unavailable_no_token(self): database.spanner_api = self._make_spanner_api() session = _Session(database) derived = self._makeDerived(session) - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) self.assertEqual(list(resumable), list(ITEMS)) self.assertEqual(len(restart.mock_calls), 2) self.assertEqual(request.resume_token, b"") @@ -234,7 +263,7 @@ def test_iteration_w_raw_raising_retryable_internal_error_no_token(self): database.spanner_api = self._make_spanner_api() session = _Session(database) derived = self._makeDerived(session) - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) self.assertEqual(list(resumable), list(ITEMS)) self.assertEqual(len(restart.mock_calls), 2) self.assertEqual(request.resume_token, b"") @@ -256,10 +285,18 @@ def test_iteration_w_raw_raising_non_retryable_internal_error_no_token(self): database.spanner_api = self._make_spanner_api() session = _Session(database) derived = self._makeDerived(session) - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) with self.assertRaises(InternalServerError): list(resumable) - restart.assert_called_once_with(request=request, metadata=None) + restart.assert_called_once_with( + request=request, + metadata=[ + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ) + ], + ) self.assertNoSpans() def test_iteration_w_raw_raising_unavailable(self): @@ -278,7 +315,7 @@ def test_iteration_w_raw_raising_unavailable(self): database.spanner_api = self._make_spanner_api() session = _Session(database) derived = self._makeDerived(session) - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) self.assertEqual(list(resumable), list(FIRST + LAST)) self.assertEqual(len(restart.mock_calls), 2) self.assertEqual(request.resume_token, RESUME_TOKEN) @@ -295,7 +332,7 @@ def test_iteration_w_raw_raising_retryable_internal_error(self): fail_after=True, error=InternalServerError( "Received unexpected EOS on DATA frame from server" - ) + ), ) after = _MockIterator(*LAST) request = mock.Mock(test="test", spec=["test", "resume_token"]) @@ -304,7 +341,7 @@ def test_iteration_w_raw_raising_retryable_internal_error(self): database.spanner_api = self._make_spanner_api() session = _Session(database) derived = self._makeDerived(session) - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) self.assertEqual(list(resumable), list(FIRST + LAST)) self.assertEqual(len(restart.mock_calls), 2) self.assertEqual(request.resume_token, RESUME_TOKEN) @@ -326,10 +363,18 @@ def test_iteration_w_raw_raising_non_retryable_internal_error(self): database.spanner_api = self._make_spanner_api() session = _Session(database) derived = self._makeDerived(session) - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) with self.assertRaises(InternalServerError): list(resumable) - restart.assert_called_once_with(request=request, metadata=None) + restart.assert_called_once_with( + request=request, + metadata=[ + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ) + ], + ) self.assertNoSpans() def test_iteration_w_raw_raising_unavailable_after_token(self): @@ -347,7 +392,7 @@ def test_iteration_w_raw_raising_unavailable_after_token(self): database.spanner_api = self._make_spanner_api() session = _Session(database) derived = self._makeDerived(session) - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) self.assertEqual(list(resumable), list(FIRST + SECOND)) self.assertEqual(len(restart.mock_calls), 2) self.assertEqual(request.resume_token, RESUME_TOKEN) @@ -370,7 +415,7 @@ def test_iteration_w_raw_w_multiuse(self): session = _Session(database) derived = self._makeDerived(session) derived._multi_use = True - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) self.assertEqual(list(resumable), list(FIRST)) self.assertEqual(len(restart.mock_calls), 1) begin_count = sum( @@ -401,7 +446,7 @@ def test_iteration_w_raw_raising_unavailable_w_multiuse(self): session = _Session(database) derived = self._makeDerived(session) derived._multi_use = True - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) self.assertEqual(list(resumable), list(SECOND)) self.assertEqual(len(restart.mock_calls), 2) begin_count = sum( @@ -440,7 +485,7 @@ def test_iteration_w_raw_raising_unavailable_after_token_w_multiuse(self): derived = self._makeDerived(session) derived._multi_use = True - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) self.assertEqual(list(resumable), list(FIRST + SECOND)) self.assertEqual(len(restart.mock_calls), 2) @@ -467,7 +512,7 @@ def test_iteration_w_raw_raising_retryable_internal_error_after_token(self): fail_after=True, error=InternalServerError( "Received unexpected EOS on DATA frame from server" - ) + ), ) after = _MockIterator(*SECOND) request = mock.Mock(test="test", spec=["test", "resume_token"]) @@ -476,7 +521,7 @@ def test_iteration_w_raw_raising_retryable_internal_error_after_token(self): database.spanner_api = self._make_spanner_api() session = _Session(database) derived = self._makeDerived(session) - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) self.assertEqual(list(resumable), list(FIRST + SECOND)) self.assertEqual(len(restart.mock_calls), 2) self.assertEqual(request.resume_token, RESUME_TOKEN) @@ -497,10 +542,18 @@ def test_iteration_w_raw_raising_non_retryable_internal_error_after_token(self): database.spanner_api = self._make_spanner_api() session = _Session(database) derived = self._makeDerived(session) - resumable = self._call_fut(derived, restart, request) + resumable = self._call_fut(derived, restart, request, session=session) with self.assertRaises(InternalServerError): list(resumable) - restart.assert_called_once_with(request=request, metadata=None) + restart.assert_called_once_with( + request=request, + metadata=[ + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ) + ], + ) self.assertNoSpans() def test_iteration_w_span_creation(self): @@ -777,7 +830,13 @@ def _read_helper( ) api.streaming_read.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], retry=retry, timeout=timeout, ) @@ -1026,7 +1085,13 @@ def _execute_sql_helper( ) api.execute_streaming_sql.assert_called_once_with( request=expected_request, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), + ], timeout=timeout, retry=retry, ) @@ -1199,6 +1264,10 @@ def _partition_read_helper( metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], retry=retry, timeout=timeout, @@ -1378,6 +1447,10 @@ def _partition_query_helper( metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], retry=retry, timeout=timeout, @@ -1774,7 +1847,13 @@ def test_begin_ok_exact_staleness(self): api.begin_transaction.assert_called_once_with( session=session.name, options=expected_txn_options, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) self.assertSpanAttributes( @@ -1810,7 +1889,13 @@ def test_begin_ok_exact_strong(self): api.begin_transaction.assert_called_once_with( session=session.name, options=expected_txn_options, - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], ) self.assertSpanAttributes( @@ -1835,6 +1920,7 @@ def __init__(self): class _Database(object): def __init__(self, directed_read_options=None): self.name = "testing" + self._nth_request = 0 self._instance = _Instance() self._route_to_leader_enabled = True self._directed_read_options = directed_read_options @@ -1843,6 +1929,28 @@ def __init__(self, directed_read_options=None): def observability_options(self): return dict(db_name=self.name) + @property + def _next_nth_request(self): + self._nth_request += 1 + return self._nth_request + + @property + def _nth_client_id(self): + return 1 + + def metadata_with_request_id(self, nth_request, nth_attempt, prior_metadata=[]): + return _metadata_with_request_id( + self._nth_client_id, + self._channel_id, + nth_request, + nth_attempt, + prior_metadata, + ) + + @property + def _channel_id(self): + return 1 + class _Session(object): def __init__(self, database=None, name=TestSnapshot.SESSION_NAME): diff --git a/tests/unit/test_spanner.py b/tests/unit/test_spanner.py index 8bd95c7228..46ab95b7d9 100644 --- a/tests/unit/test_spanner.py +++ b/tests/unit/test_spanner.py @@ -41,7 +41,11 @@ _make_value_pb, _merge_query_options, ) - +from google.cloud.spanner_v1._helpers import ( + AtomicCounter, + _metadata_with_request_id, +) +from google.cloud.spanner_v1.request_id_header import REQ_RAND_PROCESS_ID import mock from google.api_core import gapic_v1 @@ -522,6 +526,10 @@ def test_transaction_should_include_begin_with_first_update(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], ) @@ -537,6 +545,10 @@ def test_transaction_should_include_begin_with_first_query(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], timeout=TIMEOUT, retry=RETRY, @@ -554,6 +566,10 @@ def test_transaction_should_include_begin_with_first_read(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], retry=RETRY, timeout=TIMEOUT, @@ -570,6 +586,10 @@ def test_transaction_should_include_begin_with_first_batch_update(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], retry=RETRY, timeout=TIMEOUT, @@ -595,6 +615,10 @@ def test_transaction_should_include_begin_w_exclude_txn_from_change_streams_with metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], ) @@ -621,6 +645,10 @@ def test_transaction_should_include_begin_w_isolation_level_with_first_update( metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], ) @@ -639,6 +667,10 @@ def test_transaction_should_use_transaction_id_if_error_with_first_batch_update( metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], retry=RETRY, timeout=TIMEOUT, @@ -653,6 +685,10 @@ def test_transaction_should_use_transaction_id_if_error_with_first_batch_update( metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.2.1", + ), ], ) @@ -669,6 +705,10 @@ def test_transaction_should_use_transaction_id_returned_by_first_query(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], ) @@ -682,6 +722,10 @@ def test_transaction_should_use_transaction_id_returned_by_first_query(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.2.1", + ), ], ) @@ -698,6 +742,10 @@ def test_transaction_should_use_transaction_id_returned_by_first_update(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], ) @@ -711,6 +759,10 @@ def test_transaction_should_use_transaction_id_returned_by_first_update(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.2.1", + ), ], ) @@ -732,6 +784,10 @@ def test_transaction_execute_sql_w_directed_read_options(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], retry=gapic_v1.method.DEFAULT, timeout=gapic_v1.method.DEFAULT, @@ -755,6 +811,10 @@ def test_transaction_streaming_read_w_directed_read_options(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], retry=RETRY, timeout=TIMEOUT, @@ -771,6 +831,10 @@ def test_transaction_should_use_transaction_id_returned_by_first_read(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], retry=RETRY, timeout=TIMEOUT, @@ -782,6 +846,10 @@ def test_transaction_should_use_transaction_id_returned_by_first_read(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.2.1", + ), ], retry=RETRY, timeout=TIMEOUT, @@ -798,6 +866,10 @@ def test_transaction_should_use_transaction_id_returned_by_first_batch_update(se metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], retry=RETRY, timeout=TIMEOUT, @@ -810,6 +882,10 @@ def test_transaction_should_use_transaction_id_returned_by_first_batch_update(se metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.2.1", + ), ], retry=RETRY, timeout=TIMEOUT, @@ -850,6 +926,10 @@ def test_transaction_for_concurrent_statement_should_begin_one_transaction_with_ metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), ], ) @@ -860,6 +940,10 @@ def test_transaction_for_concurrent_statement_should_begin_one_transaction_with_ metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.2.1", + ), ], ) @@ -868,6 +952,10 @@ def test_transaction_for_concurrent_statement_should_begin_one_transaction_with_ metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.3.1", + ), ], retry=RETRY, timeout=TIMEOUT, @@ -903,6 +991,7 @@ def test_transaction_for_concurrent_statement_should_begin_one_transaction_with_ thread.join() self._execute_update_helper(transaction=transaction, api=api) + self.assertEqual(api.execute_sql.call_count, 1) api.execute_sql.assert_any_call( request=self._execute_update_expected_request(database, begin=False), @@ -911,32 +1000,47 @@ def test_transaction_for_concurrent_statement_should_begin_one_transaction_with_ metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.3.1", + ), ], ) - api.execute_batch_dml.assert_any_call( - request=self._batch_update_expected_request(), - metadata=[ - ("google-cloud-resource-prefix", database.name), - ("x-goog-spanner-route-to-leader", "true"), - ], - retry=RETRY, - timeout=TIMEOUT, - ) - - api.execute_batch_dml.assert_any_call( - request=self._batch_update_expected_request(begin=False), - metadata=[ - ("google-cloud-resource-prefix", database.name), - ("x-goog-spanner-route-to-leader", "true"), + self.assertEqual(api.execute_batch_dml.call_count, 2) + print("\033[34marg_list", api.execute_batch_dml.call_args_list, "\033[00m") + self.assertEqual( + api.execute_batch_dml.call_args_list, + [ + mock.call( + request=self._batch_update_expected_request(), + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], + retry=RETRY, + timeout=TIMEOUT, + ), + mock.call( + request=self._batch_update_expected_request(begin=False), + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), + ], + retry=RETRY, + timeout=TIMEOUT, + ), ], - retry=RETRY, - timeout=TIMEOUT, ) - self.assertEqual(api.execute_sql.call_count, 1) - self.assertEqual(api.execute_batch_dml.call_count, 2) - def test_transaction_for_concurrent_statement_should_begin_one_transaction_with_read( self, ): @@ -977,27 +1081,43 @@ def test_transaction_for_concurrent_statement_should_begin_one_transaction_with_ metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.3.1", + ), ], ) - api.streaming_read.assert_any_call( - request=self._read_helper_expected_request(), - metadata=[ - ("google-cloud-resource-prefix", database.name), - ("x-goog-spanner-route-to-leader", "true"), - ], - retry=RETRY, - timeout=TIMEOUT, - ) - - api.streaming_read.assert_any_call( - request=self._read_helper_expected_request(begin=False), - metadata=[ - ("google-cloud-resource-prefix", database.name), - ("x-goog-spanner-route-to-leader", "true"), + self.assertEqual( + api.streaming_read.call_args_list, + [ + mock.call( + request=self._read_helper_expected_request(), + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], + retry=RETRY, + timeout=TIMEOUT, + ), + mock.call( + request=self._read_helper_expected_request(begin=False), + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), + ], + retry=RETRY, + timeout=TIMEOUT, + ), ], - retry=RETRY, - timeout=TIMEOUT, ) self.assertEqual(api.execute_sql.call_count, 1) @@ -1043,27 +1163,43 @@ def test_transaction_for_concurrent_statement_should_begin_one_transaction_with_ metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.3.1", + ), ], ) - req = self._execute_sql_expected_request(database) - api.execute_streaming_sql.assert_any_call( - request=req, - metadata=[ - ("google-cloud-resource-prefix", database.name), - ("x-goog-spanner-route-to-leader", "true"), - ], - retry=RETRY, - timeout=TIMEOUT, - ) - api.execute_streaming_sql.assert_any_call( - request=self._execute_sql_expected_request(database, begin=False), - metadata=[ - ("google-cloud-resource-prefix", database.name), - ("x-goog-spanner-route-to-leader", "true"), + self.assertEqual( + api.execute_streaming_sql.call_args_list, + [ + mock.call( + request=self._execute_sql_expected_request(database), + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), + ], + retry=RETRY, + timeout=TIMEOUT, + ), + mock.call( + request=self._execute_sql_expected_request(database, begin=False), + metadata=[ + ("google-cloud-resource-prefix", database.name), + ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.2.1", + ), + ], + retry=RETRY, + timeout=TIMEOUT, + ), ], - retry=RETRY, - timeout=TIMEOUT, ) self.assertEqual(api.execute_sql.call_count, 1) @@ -1079,19 +1215,33 @@ def test_transaction_should_execute_sql_with_route_to_leader_disabled(self): api.execute_streaming_sql.assert_called_once_with( request=self._execute_sql_expected_request(database=database), - metadata=[("google-cloud-resource-prefix", database.name)], + metadata=[ + ("google-cloud-resource-prefix", database.name), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", + ), + ], timeout=TIMEOUT, retry=RETRY, ) class _Client(object): + NTH_CLIENT = AtomicCounter() + def __init__(self): from google.cloud.spanner_v1 import ExecuteSqlRequest self._query_options = ExecuteSqlRequest.QueryOptions(optimizer_version="1") self.directed_read_options = None self.default_transaction_options = DefaultTransactionOptions() + self._nth_client_id = _Client.NTH_CLIENT.increment() + self._nth_request = AtomicCounter() + + @property + def _next_nth_request(self): + return self._nth_request.increment() class _Instance(object): @@ -1107,6 +1257,27 @@ def __init__(self): self._directed_read_options = None self.default_transaction_options = DefaultTransactionOptions() + @property + def _next_nth_request(self): + return self._instance._client._next_nth_request + + @property + def _nth_client_id(self): + return self._instance._client._nth_client_id + + def metadata_with_request_id(self, nth_request, nth_attempt, prior_metadata=[]): + return _metadata_with_request_id( + self._nth_client_id, + self._channel_id, + nth_request, + nth_attempt, + prior_metadata, + ) + + @property + def _channel_id(self): + return 1 + class _Session(object): _transaction = None diff --git a/tests/unit/test_transaction.py b/tests/unit/test_transaction.py index ff4743f1f6..64fafcae46 100644 --- a/tests/unit/test_transaction.py +++ b/tests/unit/test_transaction.py @@ -25,6 +25,7 @@ AtomicCounter, _metadata_with_request_id, ) +from google.cloud.spanner_v1.request_id_header import REQ_RAND_PROCESS_ID from tests._helpers import ( HAS_OPENTELEMETRY_INSTALLED, @@ -201,11 +202,10 @@ def test_begin_ok(self): [ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), - # TODO(@odeke-em): enable with PR #1367. - # ( - # "x-goog-spanner-request-id", - # f"1.{REQ_RAND_PROCESS_ID}.{_Client.NTH_CLIENT.value}.1.1.1", - # ), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{_Client.NTH_CLIENT.value}.1.1.1", + ), ], ) @@ -310,11 +310,10 @@ def test_rollback_ok(self): [ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), - # TODO(@odeke-em): enable with PR #1367. - # ( - # "x-goog-spanner-request-id", - # f"1.{REQ_RAND_PROCESS_ID}.{_Client.NTH_CLIENT.value}.1.1.1", - # ), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{_Client.NTH_CLIENT.value}.1.1.1", + ), ], ) @@ -506,11 +505,10 @@ def _commit_helper( [ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), - # TODO(@odeke-em): enable with PR #1367. - # ( - # "x-goog-spanner-request-id", - # f"1.{REQ_RAND_PROCESS_ID}.{_Client.NTH_CLIENT.value}.1.1.1", - # ), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{_Client.NTH_CLIENT.value}.1.1.1", + ), ], ) self.assertEqual(actual_request_options, expected_request_options) @@ -685,11 +683,10 @@ def _execute_update_helper( metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), - # TODO(@odeke-em): enable with PR #1367. - # ( - # "x-goog-spanner-request-id", - # f"1.{REQ_RAND_PROCESS_ID}.{_Client.NTH_CLIENT.value}.1.1.1", - # ), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{_Client.NTH_CLIENT.value}.1.1.1", + ), ], ) @@ -883,11 +880,10 @@ def _batch_update_helper( metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), - # TODO(@odeke-em): enable with PR #1367. - # ( - # "x-goog-spanner-request-id", - # f"1.{REQ_RAND_PROCESS_ID}.{_Client.NTH_CLIENT.value}.1.1.1", - # ), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{_Client.NTH_CLIENT.value}.1.1.1", + ), ], retry=retry, timeout=timeout, @@ -1003,11 +999,10 @@ def test_context_mgr_success(self): [ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), - # TODO(@odeke-em): enable with PR #1367. - # ( - # "x-goog-spanner-request-id", - # f"1.{REQ_RAND_PROCESS_ID}.{_Client.NTH_CLIENT.value}.1.2.1", - # ), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{_Client.NTH_CLIENT.value}.1.2.1", + ), ], ) From 4322bd6ec4451480b521f1a315231f2ed0aa61da Mon Sep 17 00:00:00 2001 From: Emmanuel T Odeke Date: Tue, 13 May 2025 15:46:07 -0700 Subject: [PATCH 2/4] Include batch* --- google/cloud/spanner_v1/batch.py | 21 +++++++++++++++------ tests/unit/test_batch.py | 27 ++++++++++++++++----------- tests/unit/test_database.py | 12 ++++++++++++ 3 files changed, 43 insertions(+), 17 deletions(-) diff --git a/google/cloud/spanner_v1/batch.py b/google/cloud/spanner_v1/batch.py index 3d632c7568..646c73a316 100644 --- a/google/cloud/spanner_v1/batch.py +++ b/google/cloud/spanner_v1/batch.py @@ -26,6 +26,7 @@ _metadata_with_prefix, _metadata_with_leader_aware_routing, _merge_Transaction_Options, + AtomicCounter, ) from google.cloud.spanner_v1._opentelemetry_tracing import trace_call from google.cloud.spanner_v1 import RequestOptions @@ -249,17 +250,25 @@ def commit( observability_options=observability_options, metadata=metadata, ), MetricsCapture(): - method = functools.partial( - api.commit, - request=request, - metadata=metadata, - ) + attempt = AtomicCounter() + nth_request = getattr(database, "_next_nth_request", 0) + + def wrapped_method(*args, **kwargs): + method = functools.partial( + api.commit, + request=request, + metadata=database.metadata_with_request_id( + nth_request, attempt.increment(), metadata + ), + ) + return method(*args, **kwargs) + deadline = time.time() + kwargs.get( "timeout_secs", DEFAULT_RETRY_TIMEOUT_SECS ) default_retry_delay = kwargs.get("default_retry_delay", None) response = _retry_on_aborted_exception( - method, + wrapped_method, deadline=deadline, default_retry_delay=default_retry_delay, ) diff --git a/tests/unit/test_batch.py b/tests/unit/test_batch.py index 33130dd98d..2014b60eb9 100644 --- a/tests/unit/test_batch.py +++ b/tests/unit/test_batch.py @@ -38,8 +38,10 @@ from google.rpc.status_pb2 import Status from google.cloud.spanner_v1._helpers import ( + AtomicCounter, _metadata_with_request_id, ) +from google.cloud.spanner_v1.request_id_header import REQ_RAND_PROCESS_ID TABLE_NAME = "citizens" COLUMNS = ["email", "first_name", "last_name", "age"] @@ -252,6 +254,10 @@ def test_commit_ok(self): [ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) self.assertEqual(request_options, RequestOptions()) @@ -346,6 +352,10 @@ def _test_commit_with_options( [ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) self.assertEqual(actual_request_options, expected_request_options) @@ -456,6 +466,10 @@ def test_context_mgr_success(self): [ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) self.assertEqual(request_options, RequestOptions()) @@ -582,12 +596,6 @@ def _test_batch_write_with_request_options( "traceparent is missing in metadata", ) - # expected_metadata.append( - # ( - # "x-goog-spanner-request-id", - # f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.1.1.1", - # ) - # ) # Remove traceparent from actual metadata for comparison filtered_metadata = [item for item in metadata if item[0] != "traceparent"] @@ -647,7 +655,7 @@ def session_id(self): class _Database(object): name = "testing" _route_to_leader_enabled = True - NTH_CLIENT = 1 + NTH_CLIENT_ID = AtomicCounter() def __init__(self, enable_end_to_end_tracing=False): self.name = "testing" @@ -656,16 +664,13 @@ def __init__(self, enable_end_to_end_tracing=False): self.observability_options = dict(enable_end_to_end_tracing=True) self.default_transaction_options = DefaultTransactionOptions() self._nth_request = 0 + self._nth_client_id = _Database.NTH_CLIENT_ID.increment() @property def _next_nth_request(self): self._nth_request += 1 return self._nth_request - @property - def _nth_client_id(self): - return 1 - def metadata_with_request_id(self, nth_request, nth_attempt, prior_metadata=[]): return _metadata_with_request_id( self._nth_client_id, diff --git a/tests/unit/test_database.py b/tests/unit/test_database.py index 17f9c78e6c..56ac22eab0 100644 --- a/tests/unit/test_database.py +++ b/tests/unit/test_database.py @@ -2063,6 +2063,10 @@ def test_context_mgr_success(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -2110,6 +2114,10 @@ def test_context_mgr_w_commit_stats_success(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) @@ -2155,6 +2163,10 @@ def test_context_mgr_w_aborted_commit_status(self): metadata=[ ("google-cloud-resource-prefix", database.name), ("x-goog-spanner-route-to-leader", "true"), + ( + "x-goog-spanner-request-id", + f"1.{REQ_RAND_PROCESS_ID}.{database._nth_client_id}.{database._channel_id}.1.1", + ), ], ) From 6f1ffff849b8543a28205299c1f85a949b57a2f9 Mon Sep 17 00:00:00 2001 From: Emmanuel T Odeke Date: Wed, 14 May 2025 05:12:07 -0700 Subject: [PATCH 3/4] Address review feedback --- google/cloud/spanner_v1/batch.py | 10 +++++---- google/cloud/spanner_v1/database.py | 5 +++-- google/cloud/spanner_v1/session.py | 8 +++---- google/cloud/spanner_v1/snapshot.py | 31 +++++++++----------------- google/cloud/spanner_v1/transaction.py | 3 +-- tests/unit/test_spanner.py | 1 - 6 files changed, 25 insertions(+), 33 deletions(-) diff --git a/google/cloud/spanner_v1/batch.py b/google/cloud/spanner_v1/batch.py index 646c73a316..4d35469d39 100644 --- a/google/cloud/spanner_v1/batch.py +++ b/google/cloud/spanner_v1/batch.py @@ -250,15 +250,17 @@ def commit( observability_options=observability_options, metadata=metadata, ), MetricsCapture(): - attempt = AtomicCounter() - nth_request = getattr(database, "_next_nth_request", 0) - def wrapped_method(*args, **kwargs): method = functools.partial( api.commit, request=request, metadata=database.metadata_with_request_id( - nth_request, attempt.increment(), metadata + # This code is retried due to ABORTED, hence nth_request + # should be increased. attempt can only be increased if + # we encounter UNAVAILABLE or INTERNAL. + getattr(database, "_next_nth_request", 0), + 1, + metadata, ), ) return method(*args, **kwargs) diff --git a/google/cloud/spanner_v1/database.py b/google/cloud/spanner_v1/database.py index c4bdfa7c07..f2d570feb9 100644 --- a/google/cloud/spanner_v1/database.py +++ b/google/cloud/spanner_v1/database.py @@ -806,7 +806,9 @@ def _next_nth_request(self): @property def _nth_client_id(self): - return self._instance._client._nth_client_id + if self._instance and self._instance._client: + return self._instance._client._nth_client_id + return 0 def session(self, labels=None, database_role=None): """Factory to create a session for this database. @@ -1028,7 +1030,6 @@ def restore(self, source): ) future = api.restore_database( request=request, - # TODO: Infer the channel_id being used. metadata=self.metadata_with_request_id(self._next_nth_request, 1, metadata), ) return future diff --git a/google/cloud/spanner_v1/session.py b/google/cloud/spanner_v1/session.py index 2e3492892b..e3ece505c6 100644 --- a/google/cloud/spanner_v1/session.py +++ b/google/cloud/spanner_v1/session.py @@ -252,9 +252,7 @@ def delete(self): database = self._database api = database.spanner_api - metadata = database.metadata_with_request_id( - database._next_nth_request, 1, _metadata_with_prefix(database.name) - ) + metadata = _metadata_with_prefix(database.name) observability_options = getattr(self._database, "observability_options", None) with trace_call( "CloudSpanner.DeleteSession", @@ -268,7 +266,9 @@ def delete(self): ), MetricsCapture(): api.delete_session( name=self.name, - metadata=metadata, + metadata=database.metadata_with_request_id( + database._next_nth_request, 1, metadata + ), ) def ping(self): diff --git a/google/cloud/spanner_v1/snapshot.py b/google/cloud/spanner_v1/snapshot.py index e00a253232..badc23026e 100644 --- a/google/cloud/spanner_v1/snapshot.py +++ b/google/cloud/spanner_v1/snapshot.py @@ -38,6 +38,7 @@ _retry, _check_rst_stream_error, _SessionWrapper, + AtomicCounter, ) from google.cloud.spanner_v1._opentelemetry_tracing import trace_call from google.cloud.spanner_v1.streamed import StreamedResultSet @@ -82,11 +83,6 @@ def _restart_on_unavailable( resume_token = b"" item_buffer = [] - def next_nth_request(): - return getattr(request_id_manager, "_next_nth_request", 0) - - nth_request = next_nth_request() - if transaction is not None: transaction_selector = transaction._make_txn_selector() elif transaction_selector is None: @@ -96,7 +92,8 @@ def next_nth_request(): request.transaction = transaction_selector iterator = None - attempt = 0 + attempt = 1 + nth_request = getattr(request_id_manager, "_next_nth_request", 0) while True: try: @@ -108,7 +105,6 @@ def next_nth_request(): observability_options=observability_options, metadata=metadata, ), MetricsCapture(): - attempt += 1 iterator = method( request=request, metadata=request_id_manager.metadata_with_request_id( @@ -142,8 +138,7 @@ def next_nth_request(): if transaction is not None: transaction_selector = transaction._make_txn_selector() request.transaction = transaction_selector - nth_request = next_nth_request() - attempt = 1 + attempt += 1 iterator = method( request=request, metadata=request_id_manager.metadata_with_request_id( @@ -169,8 +164,7 @@ def next_nth_request(): request.resume_token = resume_token if transaction is not None: transaction_selector = transaction._make_txn_selector() - nth_request = next_nth_request() - attempt = 1 + attempt += 1 request.transaction = transaction_selector iterator = method( request=request, @@ -753,12 +747,11 @@ def partition_read( metadata=metadata, ), MetricsCapture(): nth_request = getattr(database, "_next_nth_request", 0) - counters = dict(attempt=0) + attempt = AtomicCounter() def attempt_tracking_method(): - counters["attempt"] += 1 all_metadata = database.metadata_with_request_id( - nth_request, counters["attempt"], metadata + nth_request, attempt.increment(), metadata ) method = functools.partial( api.partition_read, @@ -867,12 +860,11 @@ def partition_query( metadata=metadata, ), MetricsCapture(): nth_request = getattr(database, "_next_nth_request", 0) - counters = dict(attempt=0) + attempt = AtomicCounter() def attempt_tracking_method(): - counters["attempt"] += 1 all_metadata = database.metadata_with_request_id( - nth_request, counters["attempt"], metadata + nth_request, attempt.increment(), metadata ) method = functools.partial( api.partition_query, @@ -1024,12 +1016,11 @@ def begin(self): metadata=metadata, ), MetricsCapture(): nth_request = getattr(database, "_next_nth_request", 0) - counters = dict(attempt=0) + attempt = AtomicCounter() def attempt_tracking_method(): - counters["attempt"] += 1 all_metadata = database.metadata_with_request_id( - nth_request, counters["attempt"], metadata + nth_request, attempt.increment(), metadata ) method = functools.partial( api.begin_transaction, diff --git a/google/cloud/spanner_v1/transaction.py b/google/cloud/spanner_v1/transaction.py index 8adbf8e108..e16912dcf1 100644 --- a/google/cloud/spanner_v1/transaction.py +++ b/google/cloud/spanner_v1/transaction.py @@ -186,13 +186,12 @@ def begin(self): nth_request = database._next_nth_request def wrapped_method(*args, **kwargs): - attempt.increment() method = functools.partial( api.begin_transaction, session=self._session.name, options=txn_options, metadata=database.metadata_with_request_id( - nth_request, attempt.value, metadata + nth_request, attempt.increment(), metadata ), ) return method(*args, **kwargs) diff --git a/tests/unit/test_spanner.py b/tests/unit/test_spanner.py index 46ab95b7d9..b3b24ad6c8 100644 --- a/tests/unit/test_spanner.py +++ b/tests/unit/test_spanner.py @@ -1008,7 +1008,6 @@ def test_transaction_for_concurrent_statement_should_begin_one_transaction_with_ ) self.assertEqual(api.execute_batch_dml.call_count, 2) - print("\033[34marg_list", api.execute_batch_dml.call_args_list, "\033[00m") self.assertEqual( api.execute_batch_dml.call_args_list, [ From cdc1d0e63014df05fece634ffb370861d0891ed7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Knut=20Olav=20L=C3=B8ite?= Date: Fri, 16 May 2025 18:19:14 +0200 Subject: [PATCH 4/4] chore: fix formatting --- google/cloud/spanner_v1/batch.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/google/cloud/spanner_v1/batch.py b/google/cloud/spanner_v1/batch.py index 4d35469d39..0cbf044672 100644 --- a/google/cloud/spanner_v1/batch.py +++ b/google/cloud/spanner_v1/batch.py @@ -26,7 +26,6 @@ _metadata_with_prefix, _metadata_with_leader_aware_routing, _merge_Transaction_Options, - AtomicCounter, ) from google.cloud.spanner_v1._opentelemetry_tracing import trace_call from google.cloud.spanner_v1 import RequestOptions @@ -250,6 +249,7 @@ def commit( observability_options=observability_options, metadata=metadata, ), MetricsCapture(): + def wrapped_method(*args, **kwargs): method = functools.partial( api.commit,