From 71c0fdc7a195f5fa1d09ff39b092d7c3dbc0e5c4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Knut=20Olav=20L=C3=B8ite?= Date: Thu, 10 Sep 2026 10:29:28 +0200 Subject: [PATCH] perf(spanner): optimize built-in metrics hot path and harden concurrency Optimize hot-path execution in the Spanner client's built-in metrics subsystem and resolve subtle regex boundary and concurrency issues: - Correctness & Regex Parsing: - Add negative lookbehind `(? str: + """Safely decode bytes to str or return str representation without raising.""" + if value is None: + return "" + if isinstance(value, str): + return value + if isinstance(value, bytes): + return value.decode("utf-8", errors="replace") + return str(value) + + # Monitored resource labels MONITORED_RES_LABEL_KEY_PROJECT = "project_id" MONITORED_RES_LABEL_KEY_INSTANCE = "instance_id" diff --git a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_capture.py b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_capture.py index 77c1f86c4dee..640aa2b93f67 100644 --- a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_capture.py +++ b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_capture.py @@ -42,6 +42,7 @@ def __init__(self, resource_info: dict = None): resource_info (dict): Optional dictionary containing project, instance and database info. """ self._resource_info = resource_info + self._token = None def __enter__(self): """Enter the runtime context related to this object. @@ -88,15 +89,16 @@ def __exit__(self, exc_type, exc_value, traceback): Returns: bool: False to propagate the exception if any occurred. """ - # Short circuit out if metrics are disable - if not SpannerMetricsTracerFactory().enabled: + token = self._token + if token is None: return False - tracer = SpannerMetricsTracerFactory.get_current_tracer() - if tracer: - tracer.record_operation_completion() - - # Reset the context var using the token - if getattr(self, "_token", None): - SpannerMetricsTracerFactory.reset_current_tracer(self._token) + try: + tracer = SpannerMetricsTracerFactory.get_current_tracer() + if tracer: + tracer.record_operation_completion() + finally: + # Reset the context var using the token + SpannerMetricsTracerFactory.reset_current_tracer(token) + self._token = None return False # Propagate the exception if any diff --git a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_interceptor.py b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_interceptor.py index 1205b09c1840..6c2ca7855380 100644 --- a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_interceptor.py +++ b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_interceptor.py @@ -14,19 +14,55 @@ """Interceptor for collecting Cloud Spanner metrics.""" +import functools import inspect import logging import re +import threading from typing import Any, Dict import grpc from grpc_interceptor import ClientInterceptor -from .constants import GOOGLE_CLOUD_RESOURCE_KEY, SPANNER_METHOD_PREFIX +from .constants import ( + GOOGLE_CLOUD_RESOURCE_KEY, + SPANNER_METHOD_PREFIX, + _safe_decode_utf8, +) from .spanner_metrics_tracer_factory import SpannerMetricsTracerFactory logger = logging.getLogger(__name__) +_RESOURCE_PATH_PATTERN = re.compile( + r"^projects/(?P[^/]+)(/instances/(?P[^/]+))?(/databases/(?P[^/]+))?(/sessions/(?P[^/]+))?.*$" +) + +_RESOURCE_KEY_STR = GOOGLE_CLOUD_RESOURCE_KEY +_RESOURCE_KEY_BYTES = GOOGLE_CLOUD_RESOURCE_KEY.encode("utf-8") + + +@functools.lru_cache(maxsize=64) +def _format_method_name_str(method_str: str) -> str: + return method_str.removeprefix(SPANNER_METHOD_PREFIX).replace("/", ".") + + +def _format_method_name(method_name_input: Any) -> str: + """Format method name to be Spanner. with caching.""" + if isinstance(method_name_input, str): + return _format_method_name_str(method_name_input) + return _format_method_name_str(_safe_decode_utf8(method_name_input)) + + +@functools.lru_cache(maxsize=128) +def _parse_resource_path_cached(path: str) -> Dict[str, str]: + """Parse resource path using regex with LRU caching.""" + match = _RESOURCE_PATH_PATTERN.match(path) + if match: + return { + key: value for key, value in match.groupdict().items() if value is not None + } + return {} + class MetricsInterceptor(ClientInterceptor): """Interceptor that collects metrics for Cloud Spanner operations.""" @@ -41,15 +77,10 @@ def _parse_resource_path(path: str) -> dict: Returns: dict: Extracted resource components """ - # Match paths like: - # projects/{project}/instances/{instance}/databases/{database}/sessions/{session} - # projects/{project}/instances/{instance}/databases/{database} - # projects/{project}/instances/{instance} - pattern = r"^projects/(?P[^/]+)(/instances/(?P[^/]+))?(/databases/(?P[^/]+))?(/sessions/(?P[^/]+))?.*$" - match = re.match(pattern, path) - if match: - return {k: v for k, v in match.groupdict().items() if v is not None} - return {} + if not path or not isinstance(path, str): + return {} + + return _parse_resource_path_cached(path).copy() @staticmethod def _extract_resource_from_path(metadata: Any) -> Dict[str, str]: @@ -65,29 +96,43 @@ def _extract_resource_from_path(metadata: Any) -> Dict[str, str]: if not metadata: return {} - items = metadata.items() if isinstance(metadata, dict) else metadata path = "" - - for key, value in items: - key_str = key.decode("utf-8") if isinstance(key, bytes) else str(key) - if key_str == GOOGLE_CLOUD_RESOURCE_KEY: - path = value.decode("utf-8") if isinstance(value, bytes) else str(value) - break - - resources = MetricsInterceptor._parse_resource_path(path) - return resources + if isinstance(metadata, dict): + raw_path = metadata.get(_RESOURCE_KEY_STR) or metadata.get( + _RESOURCE_KEY_BYTES + ) + if raw_path is not None: + path = _safe_decode_utf8(raw_path) + else: + try: + metadata_iter = iter(metadata) + except TypeError: + return {} + for item in metadata_iter: + if not (isinstance(item, (list, tuple)) and len(item) == 2): + continue + key, value = item + if key == _RESOURCE_KEY_STR or key == _RESOURCE_KEY_BYTES: + path = _safe_decode_utf8(value) + break + + return MetricsInterceptor._parse_resource_path(path) @staticmethod - def _set_metrics_tracer_attributes(resources: Dict[str, str]) -> None: + def _set_metrics_tracer_attributes( + resources: Dict[str, str], tracer: Any = None + ) -> None: """ Sets the metric tracer attributes based on the provided resources. - This method updates the current metric tracer's attributes with the project, instance, and database information extracted from the resources dictionary. If the current metric tracer is not set, the method does nothing. + This method updates the metric tracer's attributes with the project, instance, and database information extracted from the resources dictionary. If the metric tracer is not set, the method does nothing. Args: resources (Dict[str, str]): A dictionary containing project, instance, and database information. + tracer (Any, optional): The metric tracer instance. If not provided, retrieves the current tracer. """ - tracer = SpannerMetricsTracerFactory.get_current_tracer() + if tracer is None: + tracer = SpannerMetricsTracerFactory.get_current_tracer() if tracer is None: return @@ -99,6 +144,23 @@ def _set_metrics_tracer_attributes(resources: Dict[str, str]) -> None: if "database" in resources: tracer.set_database(resources["database"]) + @staticmethod + def _prepare_attempt(tracer: Any, call_details: Any) -> None: + """Prepare tracer attributes and record attempt start from call details.""" + if not ( + tracer.client_attributes.get("project_id") + and tracer.client_attributes.get("instance_id") + and tracer.client_attributes.get("database") + ): + resources = MetricsInterceptor._extract_resource_from_path( + call_details.metadata + ) + MetricsInterceptor._set_metrics_tracer_attributes(resources, tracer=tracer) + + method_name = _format_method_name(call_details.method) + tracer.set_method(method_name) + tracer.record_attempt_start() + def intercept(self, invoked_method, request_or_iterator, call_details): """Intercept gRPC calls to collect metrics. @@ -115,25 +177,8 @@ def intercept(self, invoked_method, request_or_iterator, call_details): if tracer is None or not factory.enabled: return invoked_method(request_or_iterator, call_details) - # Setup Metric Tracer attributes from call details - ## Extract Project / Instance / Database from header information if not already set - if not ( - tracer.client_attributes.get("project_id") - and tracer.client_attributes.get("instance_id") - and tracer.client_attributes.get("database") - ): - resources = self._extract_resource_from_path(call_details.metadata) - self._set_metrics_tracer_attributes(resources) - - ## Format method to be be spanner. - method_str = call_details.method - method_name = method_str.removeprefix(SPANNER_METHOD_PREFIX).replace("/", ".") - - tracer.set_method(method_name) - tracer.record_attempt_start() - + self._prepare_attempt(tracer, call_details) response = invoked_method(request_or_iterator, call_details) - return _wrap_response(response, tracer) @@ -197,24 +242,7 @@ async def _async_intercept( if tracer is None or not factory.enabled: return await continuation(call_details, request_or_iterator) - if not ( - tracer.client_attributes.get("project_id") - and tracer.client_attributes.get("instance_id") - and tracer.client_attributes.get("database") - ): - resources = MetricsInterceptor._extract_resource_from_path( - call_details.metadata - ) - MetricsInterceptor._set_metrics_tracer_attributes(resources) - - method_str = call_details.method - if isinstance(method_str, bytes): - method_str = method_str.decode("utf-8") - method_name = method_str.removeprefix(SPANNER_METHOD_PREFIX).replace("/", ".") - - tracer.set_method(method_name) - tracer.record_attempt_start() - + MetricsInterceptor._prepare_attempt(tracer, call_details) response = await continuation(call_details, request_or_iterator) if hasattr(response, "__anext__"): return _AsyncStreamingResponseWrapper(response, tracer) @@ -230,6 +258,7 @@ def __init__(self, response, tracer): self._tracer = tracer self._metrics_recorded = False self._iterator = None + self._lock = threading.Lock() def __iter__(self): self._iterator = iter(self._response) @@ -248,9 +277,10 @@ def __next__(self): raise def _record_metrics(self): - if self._metrics_recorded: - return - self._metrics_recorded = True + with self._lock: + if self._metrics_recorded: + return + self._metrics_recorded = True try: self._tracer.record_attempt_completion() metadata = [] @@ -263,9 +293,40 @@ def _record_metrics(self): except Exception as e: logger.warning(f"Failed to record metrics: {e}") + def cancel(self, *args, **kwargs): + cancelled = None + if hasattr(self._response, "cancel"): + cancelled = self._response.cancel(*args, **kwargs) + if cancelled is not False: + with self._lock: + if self._metrics_recorded: + return cancelled + self._metrics_recorded = True + try: + self._tracer.record_attempt_completion( + status=grpc.StatusCode.CANCELLED.name + ) + metadata = [] + if hasattr(self._response, "initial_metadata"): + try: + metadata.extend(self._response.initial_metadata() or []) + except Exception: + pass + if metadata: + self._tracer.record_front_end_metrics(metadata) + except Exception as e: + logger.warning(f"Failed to record metrics on cancel: {e}") + return cancelled + def __del__(self): + with self._lock: + if self._metrics_recorded: + return + self._metrics_recorded = True try: - self._record_metrics() + self._tracer.record_attempt_completion( + status=grpc.StatusCode.CANCELLED.name + ) except Exception: pass @@ -273,19 +334,47 @@ def __getattr__(self, name): return getattr(self._response, name) -class _AsyncUnaryResponseWrapper(grpc.aio.UnaryUnaryCall): - """Wrapper for async unary RPC response to defer metrics recording until awaited.""" +class _BaseAsyncResponseWrapper: + """Base wrapper for async RPC responses to defer metrics recording.""" def __init__(self, response, tracer): self._response = response self._tracer = tracer self._metrics_recorded = False + self._lock = threading.Lock() def add_done_callback(self, *args, **kwargs): return getattr(self._response, "add_done_callback")(*args, **kwargs) def cancel(self, *args, **kwargs): - return getattr(self._response, "cancel")(*args, **kwargs) + cancel_fn = getattr(self._response, "cancel", None) + cancelled = cancel_fn(*args, **kwargs) if cancel_fn else True + if cancelled is not False: + with self._lock: + if self._metrics_recorded: + return cancelled + self._metrics_recorded = True + try: + self._tracer.record_attempt_completion( + status=grpc.StatusCode.CANCELLED.name + ) + metadata = [] + if hasattr(self._response, "initial_metadata"): + try: + metadata_result = self._response.initial_metadata() + if inspect.isawaitable(metadata_result): + getattr(metadata_result, "close", lambda: None)() + else: + metadata.extend(metadata_result or []) + except Exception as e: + logger.warning( + f"Failed to retrieve initial metadata on cancel: {e}" + ) + if metadata: + self._tracer.record_front_end_metrics(metadata) + except Exception as e: + logger.warning(f"Failed to record metrics on cancel: {e}") + return cancelled def cancelled(self, *args, **kwargs): return getattr(self._response, "cancelled")(*args, **kwargs) @@ -311,28 +400,26 @@ def trailing_metadata(self, *args, **kwargs): def wait_for_connection(self, *args, **kwargs): return getattr(self._response, "wait_for_connection")(*args, **kwargs) - def __await__(self): - async def _wait(): - try: - return await self._response - finally: - await self._record_metrics() + def write(self, *args, **kwargs): + return getattr(self._response, "write")(*args, **kwargs) - return _wait().__await__() + def done_writing(self, *args, **kwargs): + return getattr(self._response, "done_writing")(*args, **kwargs) async def _record_metrics(self): - if self._metrics_recorded: - return - self._metrics_recorded = True + with self._lock: + if self._metrics_recorded: + return + self._metrics_recorded = True try: self._tracer.record_attempt_completion() metadata = [] if hasattr(self._response, "initial_metadata"): try: - res = self._response.initial_metadata() - if inspect.isawaitable(res): - res = await res - metadata.extend(res or []) + metadata_result = self._response.initial_metadata() + if inspect.isawaitable(metadata_result): + metadata_result = await metadata_result + metadata.extend(metadata_result or []) except Exception as e: logger.warning(f"Failed to retrieve initial metadata: {e}") self._tracer.record_front_end_metrics(metadata) @@ -340,69 +427,52 @@ async def _record_metrics(self): logger.warning(f"Failed to record metrics: {e}") def __del__(self): - if not self._metrics_recorded: + with self._lock: + if self._metrics_recorded: + return self._metrics_recorded = True - try: - self._tracer.record_attempt_completion() - except Exception: - pass + try: + self._tracer.record_attempt_completion( + status=grpc.StatusCode.CANCELLED.name + ) + except Exception: + pass def __getattr__(self, name): return getattr(self._response, name) +class _AsyncUnaryResponseWrapper( + _BaseAsyncResponseWrapper, + grpc.aio.UnaryUnaryCall, + grpc.aio.StreamUnaryCall, +): + """Wrapper for async unary RPC response to defer metrics recording until awaited.""" + + def __await__(self): + async def _wait(): + try: + return await self._response + finally: + await self._record_metrics() + + return _wait().__await__() + + class _AsyncStreamingResponseWrapper( + _BaseAsyncResponseWrapper, grpc.aio.UnaryStreamCall, - grpc.aio.StreamUnaryCall, grpc.aio.StreamStreamCall, ): """Wrapper for async streaming RPC response iterators to defer metrics recording.""" def __init__(self, response, tracer): - self._response = response - self._tracer = tracer - self._metrics_recorded = False + super().__init__(response, tracer) self._iterator = None - def add_done_callback(self, *args, **kwargs): - return getattr(self._response, "add_done_callback")(*args, **kwargs) - - def cancel(self, *args, **kwargs): - return getattr(self._response, "cancel")(*args, **kwargs) - - def cancelled(self, *args, **kwargs): - return getattr(self._response, "cancelled")(*args, **kwargs) - - def code(self, *args, **kwargs): - return getattr(self._response, "code")(*args, **kwargs) - - def details(self, *args, **kwargs): - return getattr(self._response, "details")(*args, **kwargs) - - def done(self, *args, **kwargs): - return getattr(self._response, "done")(*args, **kwargs) - - def initial_metadata(self, *args, **kwargs): - return getattr(self._response, "initial_metadata")(*args, **kwargs) - - def time_remaining(self, *args, **kwargs): - return getattr(self._response, "time_remaining")(*args, **kwargs) - - def trailing_metadata(self, *args, **kwargs): - return getattr(self._response, "trailing_metadata")(*args, **kwargs) - - def wait_for_connection(self, *args, **kwargs): - return getattr(self._response, "wait_for_connection")(*args, **kwargs) - def read(self, *args, **kwargs): return getattr(self._response, "read")(*args, **kwargs) - def write(self, *args, **kwargs): - return getattr(self._response, "write")(*args, **kwargs) - - def done_writing(self, *args, **kwargs): - return getattr(self._response, "done_writing")(*args, **kwargs) - def __aiter__(self): if hasattr(self._response, "__aiter__"): self._iterator = self._response.__aiter__() @@ -424,33 +494,3 @@ async def __anext__(self): except Exception: await self._record_metrics() raise - - async def _record_metrics(self): - if self._metrics_recorded: - return - self._metrics_recorded = True - try: - self._tracer.record_attempt_completion() - metadata = [] - if hasattr(self._response, "initial_metadata"): - try: - res = self._response.initial_metadata() - if inspect.isawaitable(res): - res = await res - metadata.extend(res or []) - except Exception as e: - logger.warning(f"Failed to retrieve initial metadata: {e}") - self._tracer.record_front_end_metrics(metadata) - except Exception as e: - logger.warning(f"Failed to record metrics: {e}") - - def __del__(self): - if not self._metrics_recorded: - self._metrics_recorded = True - try: - self._tracer.record_attempt_completion() - except Exception: - pass - - def __getattr__(self, name): - return getattr(self._response, name) diff --git a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_tracer.py b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_tracer.py index 6106fa6e18b0..276d0a09d0c8 100644 --- a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_tracer.py +++ b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_tracer.py @@ -22,7 +22,7 @@ import os import re from datetime import datetime -from typing import Any, Dict, Optional +from typing import Any, Dict, Optional, Tuple from grpc import StatusCode @@ -38,6 +38,7 @@ MONITORED_RES_LABEL_KEY_INSTANCE_CONFIG, MONITORED_RES_LABEL_KEY_LOCATION, MONITORED_RES_LABEL_KEY_PROJECT, + _safe_decode_utf8, ) try: @@ -47,6 +48,73 @@ except ImportError: # pragma: NO COVER HAS_OPENTELEMETRY_INSTALLED = False +_GFE_TIMING_PATTERN = re.compile(r"(? Optional[int]: + """Search for the pattern in text and parse the captured latency to int.""" + match = pattern.search(text) + if match: + try: + return int(float(match.group(1))) + except ValueError: + pass + return None + + +class _ObservableDict(dict): + """A dictionary that invokes an invalidation callback upon modification.""" + + def __init__(self, *args, on_change=None, **kwargs): + super().__init__(*args, **kwargs) + self._on_change = on_change + + def __setitem__(self, key, value): + super().__setitem__(key, value) + if self._on_change is not None: + self._on_change() + + def __delitem__(self, key): + super().__delitem__(key) + if self._on_change is not None: + self._on_change() + + def update(self, *args, **kwargs): + super().update(*args, **kwargs) + if self._on_change is not None: + self._on_change() + + def clear(self): + super().clear() + if self._on_change is not None: + self._on_change() + + def pop(self, *args, **kwargs): + result = super().pop(*args, **kwargs) + if self._on_change is not None: + self._on_change() + return result + + def popitem(self): + result = super().popitem() + if self._on_change is not None: + self._on_change() + return result + + def setdefault(self, key, default=None): + if key not in self: + result = super().setdefault(key, default) + if self._on_change is not None: + self._on_change() + return result + return super().setdefault(key, default) + + def copy(self): + return dict(self) + class MetricAttemptTracer: """ @@ -225,7 +293,10 @@ def __init__( instrument_afe_connectivity_error_count (Counter): Instrument for counting AFE connectivity errors. """ self.current_op = MetricOpTracer() - self._client_attributes = client_attributes + self._client_attributes = _ObservableDict( + client_attributes or {}, + on_change=self._invalidate_attribute_cache, + ) self._instrument_attempt_latency = instrument_attempt_latency self._instrument_attempt_counter = instrument_attempt_counter self._instrument_operation_latency = instrument_operation_latency @@ -242,6 +313,19 @@ def __init__( self.afe_server_timing_enabled = ( os.environ.get("SPANNER_DISABLE_AFE_SERVER_TIMING", "").lower() != "true" ) + self._cached_attempt_attributes: Optional[dict] = None + self._cached_attempt_status: Optional[str] = None + self._cached_attempt: Optional[MetricAttemptTracer] = None + self._cached_operation_attributes: Optional[dict] = None + self._cached_operation_status: Optional[str] = None + + def _invalidate_attribute_cache(self) -> None: + """Invalidates cached attribute dictionaries when client attributes are modified.""" + self._cached_attempt_attributes = None + self._cached_attempt_status = None + self._cached_attempt = None + self._cached_operation_attributes = None + self._cached_operation_status = None @staticmethod def _get_ms_time_diff(start: datetime, end: datetime) -> float: @@ -272,7 +356,7 @@ def client_attributes(self) -> Dict[str, str]: These attributes are used to provide context to the metrics being traced. Returns: - dict[str, str]: A dictionary of client attributes. + Dict[str, str]: A dictionary of client attributes. """ return self._client_attributes @@ -478,69 +562,58 @@ def record_afe_connectivity_error_count(self) -> None: @staticmethod def extract_front_end_latencies( metadata: Any, - ) -> tuple[Optional[int], Optional[int]]: - """ - Extracts both GFE and AFE latency values (in milliseconds) from response metadata. + ) -> Tuple[Optional[int], Optional[int]]: + """Extracts both GFE and AFE latency values (in milliseconds) from response metadata. + + :type metadata: Any + :param metadata: The metadata sequence or dict from the RPC response. + + :rtype: Tuple[Optional[int], Optional[int]] + :return: A tuple containing (gfe_latency, afe_latency) in milliseconds, or None if not found. """ if not metadata: return None, None if isinstance(metadata, dict): items = metadata.items() - elif isinstance(metadata, (list, tuple)): - items = [ - item - for item in metadata - if isinstance(item, (list, tuple)) and len(item) == 2 - ] else: - items = [] - - header_vals = [] - for key, val in items: - key_str = key.decode("utf-8") if isinstance(key, bytes) else str(key) - if key_str and key_str.lower() == "server-timing": - if isinstance(val, (list, tuple)): - header_vals.extend(val) - else: - header_vals.append(val) + try: + items = iter(metadata) + except TypeError: + return None, None gfe_latency = None afe_latency = None - for header_val in header_vals: - if not header_val: + for item in items: + if not (isinstance(item, (list, tuple)) and len(item) == 2): continue - if isinstance(header_val, bytes): - try: - header_val = header_val.decode("utf-8") - except Exception: - header_val = str(header_val) - elif not isinstance(header_val, str): - header_val = str(header_val) - - if gfe_latency is None: - match = re.search(r"gfet4t7;\s*dur=([0-9.]+)", header_val) - if match: - try: - gfe_latency = int(float(match.group(1))) - except ValueError: - pass - - if afe_latency is None: - match = re.search(r"afe;\s*dur=([0-9.]+)", header_val) - if match: - try: - afe_latency = int(float(match.group(1))) - except ValueError: - pass + key, value = item + is_server_timing = ( + isinstance(key, str) and key.lower() == _SERVER_TIMING_HEADER_STR + ) or (isinstance(key, bytes) and key.lower() == _SERVER_TIMING_HEADER_BYTES) + if not is_server_timing: + continue + + timing_values = value if isinstance(value, (list, tuple)) else (value,) + for timing_value in timing_values: + if not timing_value: + continue + text = _safe_decode_utf8(timing_value) + + if gfe_latency is None and "gfet4t7" in text: + gfe_latency = _extract_metric_latency(_GFE_TIMING_PATTERN, text) + + if afe_latency is None and "afe" in text: + afe_latency = _extract_metric_latency(_AFE_TIMING_PATTERN, text) + + if gfe_latency is not None and afe_latency is not None: + return gfe_latency, afe_latency return gfe_latency, afe_latency def record_front_end_metrics(self, metadata: Any) -> None: - """ - Extracts and records both GFE and AFE metrics from the RPC response metadata. - """ + """Extracts and records both GFE and AFE metrics from the RPC response metadata.""" if not self.enabled or not HAS_OPENTELEMETRY_INSTALLED: return gfe_latency, afe_latency = self.extract_front_end_latencies(metadata) @@ -556,35 +629,55 @@ def record_front_end_metrics(self, metadata: Any) -> None: self.record_afe_connectivity_error_count() def _create_operation_otel_attributes(self) -> dict: - """ - Create additional attributes for operation metrics tracing. + """Create additional attributes for operation metrics tracing. This method populates the client attributes dictionary with the operation status if metrics tracing is enabled. - It returns the updated client attributes dictionary. + It returns the updated client attributes dictionary (returned by reference from internal cache for performance; + should be treated as read-only by callers). """ if not self.enabled or not HAS_OPENTELEMETRY_INSTALLED: return {} + status = self.current_op.status + if ( + self._cached_operation_attributes is not None + and self._cached_operation_status == status + ): + return self._cached_operation_attributes + attributes = self._client_attributes.copy() - attributes[METRIC_LABEL_KEY_STATUS] = self.current_op.status + attributes[METRIC_LABEL_KEY_STATUS] = status + self._cached_operation_attributes = attributes + self._cached_operation_status = status return attributes def _create_attempt_otel_attributes(self) -> dict: - """ - Create additional attributes for attempt metrics tracing. + """Create additional attributes for attempt metrics tracing. This method populates the attributes dictionary with the attempt status if metrics tracing is enabled and an attempt exists. - It returns the updated attributes dictionary. + It returns the updated attributes dictionary (returned by reference from internal cache for performance; + should be treated as read-only by callers). """ if not self.enabled or not HAS_OPENTELEMETRY_INSTALLED: return {} - attributes = self._client_attributes.copy() - + current_attempt = self.current_op.current_attempt # Short circuit out if we don't have an attempt - if self.current_op.current_attempt is None: - return attributes + if current_attempt is None: + return self._client_attributes.copy() + + status = current_attempt.status + if ( + self._cached_attempt_attributes is not None + and self._cached_attempt_status == status + and self._cached_attempt is current_attempt + ): + return self._cached_attempt_attributes - attributes[METRIC_LABEL_KEY_STATUS] = self.current_op.current_attempt.status + attributes = self._client_attributes.copy() + attributes[METRIC_LABEL_KEY_STATUS] = status + self._cached_attempt_attributes = attributes + self._cached_attempt_status = status + self._cached_attempt = current_attempt return attributes def set_project(self, project: str) -> "MetricsTracer": @@ -712,7 +805,7 @@ def set_method(self, method: str) -> "MetricsTracer": :return: This instance of MetricsTracer for method chaining. """ if METRIC_LABEL_KEY_METHOD not in self._client_attributes: - self.client_attributes[METRIC_LABEL_KEY_METHOD] = method + self._client_attributes[METRIC_LABEL_KEY_METHOD] = method return self def enable_direct_path(self, enable: bool = False) -> "MetricsTracer": diff --git a/packages/google-cloud-spanner/tests/unit/test_metrics_capture.py b/packages/google-cloud-spanner/tests/unit/test_metrics_capture.py index 1bd1c19f9bab..b8f3ccdf5828 100644 --- a/packages/google-cloud-spanner/tests/unit/test_metrics_capture.py +++ b/packages/google-cloud-spanner/tests/unit/test_metrics_capture.py @@ -50,3 +50,95 @@ def test_metrics_capture_exit(mock_tracer_factory): pass mock_tracer.record_operation_completion.assert_called_once() + + +def test_metrics_capture_reuse(mock_tracer_factory): + mock_tracer = mock.Mock() + mock_tracer_factory.return_value = mock_tracer + + capture = MetricsCapture() + with capture: + assert SpannerMetricsTracerFactory.get_current_tracer() is mock_tracer + + assert SpannerMetricsTracerFactory.get_current_tracer() is None + assert capture._token is None + + # Reusing the same context manager instance must not raise an error + with capture: + assert SpannerMetricsTracerFactory.get_current_tracer() is mock_tracer + + assert SpannerMetricsTracerFactory.get_current_tracer() is None + assert capture._token is None + + +def test_metrics_capture_disabled(): + SpannerMetricsTracerFactory(enabled=False) + try: + with MetricsCapture() as capture: + assert capture is not None + assert SpannerMetricsTracerFactory.get_current_tracer() is None + finally: + SpannerMetricsTracerFactory(enabled=True) + + +def test_metrics_capture_with_resource_info(mock_tracer_factory): + mock_tracer = mock.Mock() + mock_tracer_factory.return_value = mock_tracer + + resource_info = { + "project": "test_p", + "instance": "test_i", + "database": "test_d", + } + with MetricsCapture(resource_info=resource_info): + pass + + mock_tracer.set_project.assert_called_once_with("test_p") + mock_tracer.set_instance.assert_called_once_with("test_i") + mock_tracer.set_database.assert_called_once_with("test_d") + + +def test_metrics_capture_exit_without_token(): + capture = MetricsCapture() + assert capture.__exit__(None, None, None) is False + + +def test_metrics_capture_with_partial_resource_info(mock_tracer_factory): + mock_tracer = mock.Mock() + mock_tracer_factory.return_value = mock_tracer + with MetricsCapture(resource_info={"database": "only_db"}): + pass + mock_tracer.set_database.assert_called_once_with("only_db") + mock_tracer.set_project.assert_not_called() + mock_tracer.set_instance.assert_not_called() + + +def test_metrics_capture_factory_returns_none(mock_tracer_factory): + mock_tracer_factory.return_value = None + with MetricsCapture(resource_info={"project": "p"}): + pass + + +def test_metrics_capture_with_project_and_instance_only(mock_tracer_factory): + mock_tracer = mock.Mock() + mock_tracer_factory.return_value = mock_tracer + with MetricsCapture(resource_info={"project": "p", "instance": "i"}): + pass + mock_tracer.set_project.assert_called_once_with("p") + mock_tracer.set_instance.assert_called_once_with("i") + mock_tracer.set_database.assert_not_called() + + +def test_metrics_capture_exit_error_resets_token(mock_tracer_factory): + mock_tracer = mock.Mock() + mock_tracer.record_operation_completion.side_effect = RuntimeError( + "Completion failure" + ) + mock_tracer_factory.return_value = mock_tracer + + with pytest.raises(RuntimeError, match="Completion failure"): + with MetricsCapture(): + assert SpannerMetricsTracerFactory.get_current_tracer() is mock_tracer + + # Verified: Token is cleanly reset even on exception + assert SpannerMetricsTracerFactory.get_current_tracer() is None diff --git a/packages/google-cloud-spanner/tests/unit/test_metrics_interceptor.py b/packages/google-cloud-spanner/tests/unit/test_metrics_interceptor.py index aa31dc5f1210..a2f30a8b2b5e 100644 --- a/packages/google-cloud-spanner/tests/unit/test_metrics_interceptor.py +++ b/packages/google-cloud-spanner/tests/unit/test_metrics_interceptor.py @@ -16,6 +16,7 @@ import pytest +from google.cloud.spanner_v1.metrics.constants import _safe_decode_utf8 from google.cloud.spanner_v1.metrics.metrics_interceptor import MetricsInterceptor from google.cloud.spanner_v1.metrics.spanner_metrics_tracer_factory import ( SpannerMetricsTracerFactory, @@ -118,3 +119,1043 @@ def test_intercept_with_tracer(interceptor, mock_tracer_ctx): mock_tracer_ctx.record_attempt_completion.assert_called_once() mock_tracer_ctx.record_front_end_metrics.assert_called_once() mock_invoked_method.assert_called_once_with("request", call_details) + + +def test_format_method_name(): + from google.cloud.spanner_v1.metrics.metrics_interceptor import _format_method_name + + method_str = "/google.spanner.v1.Spanner/ExecuteStreamingSql" + expected = "Spanner.ExecuteStreamingSql" + assert _format_method_name(method_str) == expected + + # Test bytes input + method_bytes = b"/google.spanner.v1.Spanner/ExecuteStreamingSql" + assert _format_method_name(method_bytes) == expected + + # Verify cached result + assert _format_method_name(method_bytes) == expected + + # Unhashable type input + assert ( + _format_method_name(["/google.spanner.v1.Spanner/ExecuteSql"]) + == "['.google.spanner.v1.Spanner.ExecuteSql']" + ) + # Non-decodable bytes + assert "Spanner" in _format_method_name(b"/google.spanner.v1.Spanner/\xff\xfe") + + +def test_extract_resource_from_path_bytes_and_dict(interceptor): + path = "projects/p/instances/i/databases/d" + expected = {"project": "p", "instance": "i", "database": "d"} + + # Bytes key in list of tuples + metadata_bytes = [(b"google-cloud-resource-prefix", path.encode("utf-8"))] + assert interceptor._extract_resource_from_path(metadata_bytes) == expected + + # Dict metadata + metadata_dict = {"google-cloud-resource-prefix": path} + assert interceptor._extract_resource_from_path(metadata_dict) == expected + + # Dict with bytes key and value + metadata_dict_bytes = {b"google-cloud-resource-prefix": path.encode("utf-8")} + assert interceptor._extract_resource_from_path(metadata_dict_bytes) == expected + + # Empty metadata + assert interceptor._extract_resource_from_path([]) == {} + assert interceptor._extract_resource_from_path({}) == {} + + +def test_parse_resource_path_edge_cases(interceptor): + path = "projects/p1/instances/i1/databases/d1" + first = interceptor._parse_resource_path(path) + second = interceptor._parse_resource_path(path) + assert first == {"project": "p1", "instance": "i1", "database": "d1"} + assert first == second + + # Mutating returned dict should not corrupt future calls + first["mutated"] = True + third = interceptor._parse_resource_path(path) + assert "mutated" not in third + + assert interceptor._parse_resource_path("") == {} + assert interceptor._parse_resource_path(None) == {} + assert interceptor._parse_resource_path(12345) == {} + + # Database named "sessions" + db_named_sessions = "projects/p1/instances/i1/databases/sessions/sessions/s123" + assert interceptor._parse_resource_path(db_named_sessions) == { + "project": "p1", + "instance": "i1", + "database": "sessions", + "session": "s123", + } + + # Instance named "sessions" + instance_named_sessions = ( + "projects/p1/instances/sessions/databases/d1/sessions/s123" + ) + assert interceptor._parse_resource_path(instance_named_sessions) == { + "project": "p1", + "instance": "sessions", + "database": "d1", + "session": "s123", + } + + # Session paths + session_path = "projects/p1/instances/i1/databases/d1/sessions/s1" + session_result = interceptor._parse_resource_path(session_path) + assert session_result == { + "project": "p1", + "instance": "i1", + "database": "d1", + "session": "s1", + } + session_path_empty = "projects/p1/instances/i1/databases/d1/sessions/" + session_empty_result = interceptor._parse_resource_path(session_path_empty) + assert session_empty_result == { + "project": "p1", + "instance": "i1", + "database": "d1", + } + # Session part starting with slash + session_slash = "projects/p1/instances/i1/databases/d1/sessions//extra" + assert interceptor._parse_resource_path(session_slash) == { + "project": "p1", + "instance": "i1", + "database": "d1", + } + # Invalid paths with sessions must not return session + assert interceptor._parse_resource_path("invalid/sessions/s123") == {} + assert interceptor._parse_resource_path("/sessions/s123") == {} + + +def test_async_streaming_response_wrapper_not_awaitable(): + from google.cloud.spanner_v1.metrics.metrics_interceptor import ( + _AsyncStreamingResponseWrapper, + ) + + mock_stream = MagicMock() + mock_tracer = MagicMock() + wrapper = _AsyncStreamingResponseWrapper(mock_stream, mock_tracer) + assert not hasattr(wrapper, "__await__") + + +@pytest.mark.asyncio +async def test_async_unary_response_wrapper_stream_unary(): + from google.cloud.spanner_v1.metrics.metrics_interceptor import ( + _AsyncUnaryResponseWrapper, + ) + + class StreamUnaryMock: + def __init__(self): + self.written = [] + self.done_writing_called = False + + def write(self, data): + self.written.append(data) + + def done_writing(self): + self.done_writing_called = True + + def initial_metadata(self): + return [] + + def __await__(self): + async def _coro(): + return "done" + + return _coro().__await__() + + mock_stream_unary = StreamUnaryMock() + mock_tracer = MagicMock() + wrapper = _AsyncUnaryResponseWrapper(mock_stream_unary, mock_tracer) + wrapper.write("item1") + wrapper.done_writing() + assert mock_stream_unary.written == ["item1"] + assert mock_stream_unary.done_writing_called is True + result = await wrapper + assert result == "done" + mock_tracer.record_attempt_completion.assert_called_once() + + +def test_extract_resource_from_path_edge_cases(interceptor): + # Non-iterable metadata + assert interceptor._extract_resource_from_path(12345) == {} + assert interceptor._extract_resource_from_path(None) == {} + + # Dict without resource prefix + assert interceptor._extract_resource_from_path({"unrelated": "header"}) == {} + + # Non-decodable bytes in dict and list metadata + assert ( + interceptor._extract_resource_from_path( + {"google-cloud-resource-prefix": b"\xff\xfe\xfd"} + ) + == {} + ) + assert ( + interceptor._extract_resource_from_path( + [("google-cloud-resource-prefix", b"\xff\xfe\xfd")] + ) + == {} + ) + + # Malformed tuple entries (not length 2) + malformed = [("single_element",), ("a", "b", "c")] + assert interceptor._extract_resource_from_path(malformed) == {} + + # Generator metadata + path = "projects/p/instances/i/databases/d" + + def metadata_generator(): + yield ("unrelated", "value") + yield ("google-cloud-resource-prefix", path) + + assert interceptor._extract_resource_from_path(metadata_generator()) == { + "project": "p", + "instance": "i", + "database": "d", + } + + +@pytest.mark.asyncio +async def test_async_metrics_interceptor(mock_tracer_ctx): + from google.cloud.spanner_v1.metrics.metrics_interceptor import ( + AsyncMetricsInterceptor, + ) + + interceptor = AsyncMetricsInterceptor() + + # 1. Async unary call + class AwaitableCallMock: + def initial_metadata(self): + return [("server-timing", "gfet4t7; dur=55")] + + def __await__(self): + async def _coro(): + return "unary_result" + + return _coro().__await__() + + async def mock_unary_continuation(details, request): + return AwaitableCallMock() + + call_details = MagicMock( + method="/google.spanner.v1.Spanner/ExecuteSql", + metadata=[ + ( + "google-cloud-resource-prefix", + "projects/p_async/instances/i_async/databases/d_async", + ) + ], + ) + + wrapped_call = await interceptor.intercept_unary_unary( + mock_unary_continuation, call_details, "req" + ) + result = await wrapped_call + assert result == "unary_result" + mock_tracer_ctx.record_attempt_start.assert_called_once() + mock_tracer_ctx.record_attempt_completion.assert_called_once() + mock_tracer_ctx.record_front_end_metrics.assert_called_once() + mock_tracer_ctx.set_method.assert_called_with("Spanner.ExecuteSql") + mock_tracer_ctx.set_project.assert_called_with("p_async") + mock_tracer_ctx.set_instance.assert_called_with("i_async") + mock_tracer_ctx.set_database.assert_called_with("d_async") + + # 2. Async streaming call + mock_tracer_ctx.record_attempt_start.reset_mock() + mock_tracer_ctx.record_attempt_completion.reset_mock() + + class AsyncIteratorMock: + def __init__(self, items): + self._items = list(items) + + def __aiter__(self): + return self + + async def __anext__(self): + if not self._items: + raise StopAsyncIteration + return self._items.pop(0) + + def initial_metadata(self): + return [("server-timing", "afe; dur=20")] + + async def mock_stream_continuation(details, request): + return AsyncIteratorMock(["chunk1", "chunk2"]) + + stream_details = MagicMock( + method="/google.spanner.v1.Spanner/ExecuteStreamingSql", + metadata=[], + ) + + wrapped_stream = await interceptor.intercept_unary_stream( + mock_stream_continuation, stream_details, "req" + ) + items = [] + async for item in wrapped_stream: + items.append(item) + assert items == ["chunk1", "chunk2"] + mock_tracer_ctx.record_attempt_start.assert_called_once() + mock_tracer_ctx.record_attempt_completion.assert_called_once() + + +def test_streaming_response_wrapper_lifecycle(): + from google.cloud.spanner_v1.metrics.metrics_interceptor import ( + _StreamingResponseWrapper, + ) + + # 1. Normal iteration + mock_response = MagicMock() + mock_response.__iter__.return_value = iter(["chunk1", "chunk2"]) + mock_response.initial_metadata.return_value = [("server-timing", "gfet4t7; dur=10")] + mock_tracer = MagicMock() + + wrapper = _StreamingResponseWrapper(mock_response, mock_tracer) + items = list(wrapper) + assert items == ["chunk1", "chunk2"] + mock_tracer.record_attempt_completion.assert_called_once() + mock_tracer.record_front_end_metrics.assert_called_once_with( + [("server-timing", "gfet4t7; dur=10")] + ) + + # 2. Exception during iteration + mock_response_err = MagicMock() + mock_response_err.__iter__.return_value = iter(["ok"]) + + class FaultyIterator: + def __iter__(self): + return self + + def __next__(self): + raise RuntimeError("Stream broken") + + mock_tracer_err = MagicMock() + wrapper_err = _StreamingResponseWrapper(FaultyIterator(), mock_tracer_err) + with pytest.raises(RuntimeError, match="Stream broken"): + next(wrapper_err) + mock_tracer_err.record_attempt_completion.assert_called_once() + + # 3. Explicit cancellation + mock_response_cancel = MagicMock() + mock_tracer_cancel = MagicMock() + wrapper_cancel = _StreamingResponseWrapper(mock_response_cancel, mock_tracer_cancel) + wrapper_cancel.cancel() + mock_tracer_cancel.record_attempt_completion.assert_called_once_with( + status="CANCELLED" + ) + mock_response_cancel.cancel.assert_called_once() + + # 4. Finalizer (__del__) when not completed + mock_response_del = MagicMock() + mock_tracer_del = MagicMock() + wrapper_del = _StreamingResponseWrapper(mock_response_del, mock_tracer_del) + wrapper_del.__del__() + mock_tracer_del.record_attempt_completion.assert_called_once_with( + status="CANCELLED" + ) + + # 5. __getattr__ delegation and error handling + mock_response_attr = MagicMock() + mock_response_attr.custom_field = "custom_value" + mock_response_attr.initial_metadata.side_effect = RuntimeError("Metadata failed") + mock_tracer_attr = MagicMock() + mock_tracer_attr.record_attempt_completion.side_effect = RuntimeError( + "Tracer failed" + ) + wrapper_attr = _StreamingResponseWrapper(mock_response_attr, mock_tracer_attr) + assert wrapper_attr.custom_field == "custom_value" + # _record_metrics should swallow exceptions gracefully + wrapper_attr._record_metrics() + # Calling it a second time hits early return + wrapper_attr._record_metrics() + + # Cancel and del error handling + mock_tracer_cancel_err = MagicMock() + mock_tracer_cancel_err.record_attempt_completion.side_effect = RuntimeError( + "Cancel error" + ) + wrapper_cancel_err = _StreamingResponseWrapper(MagicMock(), mock_tracer_cancel_err) + wrapper_cancel_err.cancel() + + mock_tracer_del_err = MagicMock() + mock_tracer_del_err.record_attempt_completion.side_effect = RuntimeError( + "Del error" + ) + wrapper_del_err = _StreamingResponseWrapper(MagicMock(), mock_tracer_del_err) + wrapper_del_err.__del__() + + +@pytest.mark.asyncio +async def test_async_unary_response_wrapper_lifecycle(): + from google.cloud.spanner_v1.metrics.metrics_interceptor import ( + _AsyncUnaryResponseWrapper, + ) + + # 1. Exception during await + class FaultyAwaitable: + def __await__(self): + async def _coro(): + raise ValueError("RPC failed") + + return _coro().__await__() + + mock_tracer_err = MagicMock() + wrapper_err = _AsyncUnaryResponseWrapper(FaultyAwaitable(), mock_tracer_err) + with pytest.raises(ValueError, match="RPC failed"): + await wrapper_err + mock_tracer_err.record_attempt_completion.assert_called_once() + + # 2. Explicit cancellation + mock_response_cancel = MagicMock() + mock_tracer_cancel = MagicMock() + wrapper_cancel = _AsyncUnaryResponseWrapper( + mock_response_cancel, mock_tracer_cancel + ) + wrapper_cancel.cancel() + mock_tracer_cancel.record_attempt_completion.assert_called_once_with( + status="CANCELLED" + ) + mock_response_cancel.cancel.assert_called_once() + + # 3. Finalizer (__del__) when unawaited + mock_response_del = MagicMock() + mock_tracer_del = MagicMock() + wrapper_del = _AsyncUnaryResponseWrapper(mock_response_del, mock_tracer_del) + wrapper_del.__del__() + mock_tracer_del.record_attempt_completion.assert_called_once_with( + status="CANCELLED" + ) + + # 4. Proxy methods + mock_delegate = MagicMock() + mock_tracer = MagicMock() + wrapper = _AsyncUnaryResponseWrapper(mock_delegate, mock_tracer) + + wrapper.add_done_callback(MagicMock()) + mock_delegate.add_done_callback.assert_called_once() + wrapper.cancelled() + mock_delegate.cancelled.assert_called_once() + wrapper.code() + mock_delegate.code.assert_called_once() + wrapper.details() + mock_delegate.details.assert_called_once() + wrapper.done() + mock_delegate.done.assert_called_once() + wrapper.initial_metadata() + mock_delegate.initial_metadata.assert_called_once() + wrapper.time_remaining() + mock_delegate.time_remaining.assert_called_once() + wrapper.trailing_metadata() + mock_delegate.trailing_metadata.assert_called_once() + wrapper.wait_for_connection() + mock_delegate.wait_for_connection.assert_called_once() + assert wrapper.some_custom_attr == mock_delegate.some_custom_attr + + # 5. Async initial metadata and error handling + class AsyncMetadataCall: + async def initial_metadata(self): + return [("server-timing", "gfet4t7; dur=40")] + + def __await__(self): + async def _coro(): + return "ok" + + return _coro().__await__() + + mock_tracer_meta = MagicMock() + wrapper_meta = _AsyncUnaryResponseWrapper(AsyncMetadataCall(), mock_tracer_meta) + result = await wrapper_meta + assert result == "ok" + mock_tracer_meta.record_front_end_metrics.assert_called_once_with( + [("server-timing", "gfet4t7; dur=40")] + ) + # Calling it a second time hits early return + await wrapper_meta._record_metrics() + + # Cancel and del error handling + mock_tracer_cancel_err = MagicMock() + mock_tracer_cancel_err.record_attempt_completion.side_effect = RuntimeError( + "Cancel error" + ) + wrapper_cancel_err = _AsyncUnaryResponseWrapper(MagicMock(), mock_tracer_cancel_err) + wrapper_cancel_err.cancel() + + mock_tracer_del_err = MagicMock() + mock_tracer_del_err.record_attempt_completion.side_effect = RuntimeError( + "Del error" + ) + wrapper_del_err = _AsyncUnaryResponseWrapper(MagicMock(), mock_tracer_del_err) + wrapper_del_err.__del__() + + # Metadata error in _record_metrics + class UnaryMetadataErrorCall: + def initial_metadata(self): + raise RuntimeError("Metadata failed") + + def __await__(self): + async def _coro(): + return "ok" + + return _coro().__await__() + + mock_tracer_err2 = MagicMock() + mock_tracer_err2.record_attempt_completion.side_effect = RuntimeError( + "Tracer failed" + ) + wrapper_meta_err = _AsyncUnaryResponseWrapper( + UnaryMetadataErrorCall(), mock_tracer_err2 + ) + await wrapper_meta_err + + +@pytest.mark.asyncio +async def test_async_streaming_response_wrapper_lifecycle(): + from google.cloud.spanner_v1.metrics.metrics_interceptor import ( + _AsyncStreamingResponseWrapper, + ) + + # 1. Exception during async iteration + class FaultyAsyncIterator: + def __aiter__(self): + return self + + async def __anext__(self): + raise RuntimeError("Stream error") + + mock_tracer_err = MagicMock() + wrapper_err = _AsyncStreamingResponseWrapper(FaultyAsyncIterator(), mock_tracer_err) + with pytest.raises(RuntimeError, match="Stream error"): + async for _ in wrapper_err: + pass + mock_tracer_err.record_attempt_completion.assert_called_once() + + # 2. Cancellation + mock_response_cancel = MagicMock() + mock_tracer_cancel = MagicMock() + wrapper_cancel = _AsyncStreamingResponseWrapper( + mock_response_cancel, mock_tracer_cancel + ) + wrapper_cancel.cancel() + mock_tracer_cancel.record_attempt_completion.assert_called_once_with( + status="CANCELLED" + ) + mock_response_cancel.cancel.assert_called_once() + + # 3. Finalizer (__del__) when not completed + mock_response_del = MagicMock() + mock_tracer_del = MagicMock() + wrapper_del = _AsyncStreamingResponseWrapper(mock_response_del, mock_tracer_del) + wrapper_del.__del__() + mock_tracer_del.record_attempt_completion.assert_called_once_with( + status="CANCELLED" + ) + + # 4. Proxy methods + mock_delegate = MagicMock() + mock_tracer = MagicMock() + wrapper = _AsyncStreamingResponseWrapper(mock_delegate, mock_tracer) + + wrapper.add_done_callback(MagicMock()) + mock_delegate.add_done_callback.assert_called_once() + wrapper.cancelled() + mock_delegate.cancelled.assert_called_once() + wrapper.code() + mock_delegate.code.assert_called_once() + wrapper.details() + mock_delegate.details.assert_called_once() + wrapper.done() + mock_delegate.done.assert_called_once() + wrapper.initial_metadata() + mock_delegate.initial_metadata.assert_called_once() + wrapper.time_remaining() + mock_delegate.time_remaining.assert_called_once() + wrapper.trailing_metadata() + mock_delegate.trailing_metadata.assert_called_once() + wrapper.wait_for_connection() + mock_delegate.wait_for_connection.assert_called_once() + wrapper.read() + mock_delegate.read.assert_called_once() + wrapper.write("data") + mock_delegate.write.assert_called_once_with("data") + wrapper.done_writing() + mock_delegate.done_writing.assert_called_once() + assert wrapper.custom_attr == mock_delegate.custom_attr + + # 5. Async initial metadata in stream + class StreamAsyncMetadata: + def __init__(self): + self.yielded = False + + def __aiter__(self): + return self + + async def __anext__(self): + if not self.yielded: + self.yielded = True + return "item" + raise StopAsyncIteration + + async def initial_metadata(self): + return [("server-timing", "afe; dur=15")] + + mock_tracer_stream_meta = MagicMock() + wrapper_stream_meta = _AsyncStreamingResponseWrapper( + StreamAsyncMetadata(), mock_tracer_stream_meta + ) + items = [] + async for item in wrapper_stream_meta: + items.append(item) + assert items == ["item"] + mock_tracer_stream_meta.record_front_end_metrics.assert_called_once_with( + [("server-timing", "afe; dur=15")] + ) + # Calling it a second time hits early return + await wrapper_stream_meta._record_metrics() + + # Cancel and del error handling + mock_tracer_cancel_err = MagicMock() + mock_tracer_cancel_err.record_attempt_completion.side_effect = RuntimeError( + "Cancel error" + ) + wrapper_cancel_err = _AsyncStreamingResponseWrapper( + MagicMock(), mock_tracer_cancel_err + ) + wrapper_cancel_err.cancel() + + mock_tracer_del_err = MagicMock() + mock_tracer_del_err.record_attempt_completion.side_effect = RuntimeError( + "Del error" + ) + wrapper_del_err = _AsyncStreamingResponseWrapper(MagicMock(), mock_tracer_del_err) + wrapper_del_err.__del__() + + # Metadata error in stream + class StreamMetadataError: + def __init__(self): + self.yielded = False + + def __aiter__(self): + return self + + async def __anext__(self): + if not self.yielded: + self.yielded = True + return "chunk" + raise StopAsyncIteration + + def initial_metadata(self): + raise RuntimeError("Stream metadata failed") + + mock_tracer_stream_err = MagicMock() + mock_tracer_stream_err.record_attempt_completion.side_effect = RuntimeError( + "Tracer stream error" + ) + wrapper_stream_err = _AsyncStreamingResponseWrapper( + StreamMetadataError(), mock_tracer_stream_err + ) + async for _ in wrapper_stream_err: + pass + + +def test_metrics_interceptor_sync_methods_and_disabled(mock_tracer_ctx): + from google.cloud.spanner_v1.metrics.metrics_interceptor import ( + MetricsInterceptor, + _StreamingResponseWrapper, + _wrap_response, + ) + + interceptor = MetricsInterceptor() + + # 0. _set_metrics_tracer_attributes when tracer is None + token = SpannerMetricsTracerFactory._current_metrics_tracer_ctx.set(None) + try: + interceptor._set_metrics_tracer_attributes({"project": "p"}) + finally: + SpannerMetricsTracerFactory._current_metrics_tracer_ctx.reset(token) + + # 1. Intercept when disabled + SpannerMetricsTracerFactory(enabled=False) + mock_continuation = MagicMock(return_value="raw_response") + call_details = MagicMock( + method="/google.spanner.v1.Spanner/ExecuteSql", metadata=[] + ) + result = interceptor.intercept(mock_continuation, "request", call_details) + assert result == "raw_response" + mock_continuation.assert_called_once_with("request", call_details) + SpannerMetricsTracerFactory(enabled=True) + + # 2. _wrap_response with streaming response + class StreamingCallMock: + def __next__(self): + raise StopIteration + + mock_stream_call = StreamingCallMock() + wrapped_stream = _wrap_response(mock_stream_call, mock_tracer_ctx) + assert isinstance(wrapped_stream, _StreamingResponseWrapper) + + # 3. _wrap_response with unary response handling errors gracefully + class SimpleUnaryResponse: + def initial_metadata(self): + raise RuntimeError("Metadata error") + + unary_response = SimpleUnaryResponse() + mock_faulty_tracer = MagicMock() + mock_faulty_tracer.record_attempt_completion.side_effect = RuntimeError( + "Tracer error" + ) + result_unary = _wrap_response(unary_response, mock_faulty_tracer) + assert result_unary is unary_response + + # 4. _StreamingResponseWrapper with initial_metadata error + mock_resp_meta_err = MagicMock() + mock_resp_meta_err.__iter__.return_value = iter(["chunk"]) + mock_resp_meta_err.initial_metadata.side_effect = RuntimeError("Metadata failed") + wrapper_stream_meta_err = _StreamingResponseWrapper(mock_resp_meta_err, MagicMock()) + assert list(wrapper_stream_meta_err) == ["chunk"] + + +@pytest.mark.asyncio +async def test_async_wrapper_additional_error_branches(): + from google.cloud.spanner_v1.metrics.metrics_interceptor import ( + _AsyncStreamingResponseWrapper, + _AsyncUnaryResponseWrapper, + ) + + # Unary wrapper with initial_metadata error + class UnaryInitialMetaErr: + def initial_metadata(self): + raise RuntimeError("Initial meta err") + + def __await__(self): + async def _coro(): + return "ok" + + return _coro().__await__() + + wrapper_unary_meta_err = _AsyncUnaryResponseWrapper( + UnaryInitialMetaErr(), MagicMock() + ) + assert await wrapper_unary_meta_err == "ok" + + # Streaming wrapper without __aiter__ on response directly calling __anext__ + class DirectAsyncIterator: + def __init__(self): + self.done = False + + async def __anext__(self): + if not self.done: + self.done = True + return "first" + raise StopAsyncIteration + + def initial_metadata(self): + raise RuntimeError("Stream meta err") + + wrapper_direct = _AsyncStreamingResponseWrapper(DirectAsyncIterator(), MagicMock()) + item = await wrapper_direct.__anext__() + assert item == "first" + with pytest.raises(StopAsyncIteration): + await wrapper_direct.__anext__() + + +@pytest.mark.asyncio +async def test_async_metrics_interceptor_all_methods_and_disabled(mock_tracer_ctx): + from google.cloud.spanner_v1.metrics.metrics_interceptor import ( + AsyncMetricsInterceptor, + ) + + interceptor = AsyncMetricsInterceptor() + + # 1. Intercept when disabled + SpannerMetricsTracerFactory(enabled=False) + + async def mock_continuation(details, request): + return "async_raw" + + call_details = MagicMock( + method="/google.spanner.v1.Spanner/ExecuteSql", metadata=[] + ) + result = await interceptor.intercept_unary_unary( + mock_continuation, call_details, "req" + ) + assert result == "async_raw" + SpannerMetricsTracerFactory(enabled=True) + + # 2. intercept_stream_unary + async def mock_stream_unary_continuation(details, request_iterator): + class AwaitableResult: + def __await__(self): + async def _coro(): + return "stream_unary_done" + + return _coro().__await__() + + return AwaitableResult() + + wrapped_stream_unary = await interceptor.intercept_stream_unary( + mock_stream_unary_continuation, call_details, ["req1"] + ) + assert await wrapped_stream_unary == "stream_unary_done" + + # 3. intercept_stream_stream + async def mock_stream_stream_continuation(details, request_iterator): + class AsyncStreamResult: + def __aiter__(self): + return self + + async def __anext__(self): + raise StopAsyncIteration + + return AsyncStreamResult() + + wrapped_stream_stream = await interceptor.intercept_stream_stream( + mock_stream_stream_continuation, call_details, ["req1"] + ) + items = [] + async for item in wrapped_stream_stream: + items.append(item) + assert items == [] + + +@pytest.mark.asyncio +async def test_interceptor_wrapper_and_branch_edge_cases(mock_tracer_ctx): + from google.cloud.spanner_v1.metrics.metrics_interceptor import ( + AsyncMetricsInterceptor, + MetricsInterceptor, + _AsyncStreamingResponseWrapper, + _AsyncUnaryResponseWrapper, + _StreamingResponseWrapper, + _wrap_response, + ) + + interceptor = MetricsInterceptor() + async_interceptor = AsyncMetricsInterceptor() + + # 1. _set_metrics_tracer_attributes with partial and empty dicts + interceptor._set_metrics_tracer_attributes({"project": "p"}) + interceptor._set_metrics_tracer_attributes({"instance": "i"}) + interceptor._set_metrics_tracer_attributes({}) + + # 2. Interceptor call when tracer already has all resource attributes set + mock_tracer_ctx.client_attributes["project_id"] = "proj" + mock_tracer_ctx.client_attributes["instance_id"] = "inst" + mock_tracer_ctx.client_attributes["database"] = "db" + + mock_continuation = MagicMock(return_value="response") + call_details = MagicMock( + method="/google.spanner.v1.Spanner/ExecuteSql", metadata=[] + ) + result = interceptor.intercept(mock_continuation, "req", call_details) + assert result == "response" + + async def mock_async_continuation(details, request): + class AsyncCall: + def __await__(self): + async def _coro(): + return "async_response" + + return _coro().__await__() + + return AsyncCall() + + async_wrapped = await async_interceptor.intercept_unary_unary( + mock_async_continuation, call_details, "req" + ) + assert await async_wrapped == "async_response" + + # 3. _wrap_response unary branch when initial_metadata raises or is missing + class UnaryWithFailingMetadata: + def initial_metadata(self): + raise RuntimeError("Metadata failed") + + mock_tracer = MagicMock() + _wrap_response(UnaryWithFailingMetadata(), mock_tracer) + mock_tracer.record_attempt_completion.assert_called_once() + mock_tracer.record_front_end_metrics.assert_called_once_with([]) + + _wrap_response("no_initial_metadata", mock_tracer) + + # 4. _StreamingResponseWrapper: double cancel, missing cancel, del after recorded + # 4. _StreamingResponseWrapper: + # 4a. cancel returns False + stream_delegate_refused = MagicMock() + stream_delegate_refused.cancel.return_value = False + mock_tracer_unrecorded = MagicMock() + stream_wrapper_refused = _StreamingResponseWrapper( + stream_delegate_refused, mock_tracer_unrecorded + ) + assert stream_wrapper_refused.cancel() is False + assert stream_wrapper_refused._metrics_recorded is False + mock_tracer_unrecorded.record_attempt_completion.assert_not_called() + + # 4b. cancel with metadata error and successful metadata + stream_delegate_meta_err = MagicMock() + stream_delegate_meta_err.initial_metadata.side_effect = RuntimeError("meta failed") + stream_wrapper_meta_err = _StreamingResponseWrapper( + stream_delegate_meta_err, mock_tracer + ) + stream_wrapper_meta_err.cancel() + + stream_delegate_with_meta = MagicMock() + stream_delegate_with_meta.initial_metadata.return_value = [ + ("server-timing", "gfet4t7; dur=50") + ] + stream_wrapper_with_meta = _StreamingResponseWrapper( + stream_delegate_with_meta, mock_tracer + ) + stream_wrapper_with_meta.cancel() + mock_tracer.record_front_end_metrics.assert_called_with( + [("server-timing", "gfet4t7; dur=50")] + ) + + stream_delegate = MagicMock(spec=["__next__"]) + stream_wrapper = _StreamingResponseWrapper(stream_delegate, mock_tracer) + stream_wrapper.cancel() + stream_wrapper.cancel() + stream_wrapper.__del__() + + # 5. _AsyncUnaryResponseWrapper: + # 5a. cancel returns False + async_unary_refused = MagicMock() + async_unary_refused.cancel.return_value = False + async_unary_wrapper_refused = _AsyncUnaryResponseWrapper( + async_unary_refused, mock_tracer_unrecorded + ) + assert async_unary_wrapper_refused.cancel() is False + assert async_unary_wrapper_refused._metrics_recorded is False + + # 5b. cancel with awaitable metadata, failing metadata, and normal metadata + async_unary_async_meta = MagicMock() + + async def async_meta(): + return [("server-timing", "afe; dur=25")] + + async_unary_async_meta.initial_metadata.return_value = async_meta() + async_unary_wrapper_meta = _AsyncUnaryResponseWrapper( + async_unary_async_meta, mock_tracer + ) + async_unary_wrapper_meta.cancel() + + async_unary_err_meta = MagicMock() + async_unary_err_meta.initial_metadata.side_effect = RuntimeError("async meta error") + async_unary_wrapper_err = _AsyncUnaryResponseWrapper( + async_unary_err_meta, mock_tracer + ) + async_unary_wrapper_err.cancel() + + async_unary_normal_meta = MagicMock() + async_unary_normal_meta.initial_metadata.return_value = [ + ("server-timing", "afe; dur=25") + ] + async_unary_wrapper_normal = _AsyncUnaryResponseWrapper( + async_unary_normal_meta, mock_tracer + ) + async_unary_wrapper_normal.cancel() + mock_tracer.record_front_end_metrics.assert_called_with( + [("server-timing", "afe; dur=25")] + ) + + async_unary_delegate = MagicMock(spec=["cancel"]) + async_unary_wrapper = _AsyncUnaryResponseWrapper(async_unary_delegate, mock_tracer) + async_unary_wrapper.cancel() + async_unary_wrapper.cancel() + + # 6. _AsyncStreamingResponseWrapper: + # 6a. cancel returns False + async_stream_refused = MagicMock() + async_stream_refused.cancel.return_value = False + async_stream_wrapper_refused = _AsyncStreamingResponseWrapper( + async_stream_refused, mock_tracer_unrecorded + ) + assert async_stream_wrapper_refused.cancel() is False + assert async_stream_wrapper_refused._metrics_recorded is False + + # 6b. cancel with awaitable metadata, failing metadata, and normal metadata + async_stream_async_meta = MagicMock() + async_stream_async_meta.initial_metadata.return_value = async_meta() + async_stream_wrapper_meta = _AsyncStreamingResponseWrapper( + async_stream_async_meta, mock_tracer + ) + async_stream_wrapper_meta.cancel() + + async_stream_err_meta = MagicMock() + async_stream_err_meta.initial_metadata.side_effect = RuntimeError( + "async meta error" + ) + async_stream_wrapper_err = _AsyncStreamingResponseWrapper( + async_stream_err_meta, mock_tracer + ) + async_stream_wrapper_err.cancel() + + async_stream_normal_meta = MagicMock() + async_stream_normal_meta.initial_metadata.return_value = [ + ("server-timing", "gfet4t7; dur=30") + ] + async_stream_wrapper_normal = _AsyncStreamingResponseWrapper( + async_stream_normal_meta, mock_tracer + ) + async_stream_wrapper_normal.cancel() + mock_tracer.record_front_end_metrics.assert_called_with( + [("server-timing", "gfet4t7; dur=30")] + ) + + async_stream_delegate = MagicMock(spec=["cancel"]) + async_stream_wrapper = _AsyncStreamingResponseWrapper( + async_stream_delegate, mock_tracer + ) + async_stream_wrapper.cancel() + async_stream_wrapper.cancel() + + # Async response without __aiter__ (custom async iterator) + class CustomAsyncIterator: + async def __anext__(self): + raise StopAsyncIteration + + custom_async_wrapper = _AsyncStreamingResponseWrapper( + CustomAsyncIterator(), mock_tracer + ) + assert custom_async_wrapper.__aiter__() is custom_async_wrapper + async for _ in custom_async_wrapper: + pass + + # Async response where __anext__ is called before __aiter__ + class AsyncStreamWithAiter: + def __aiter__(self): + async def _gen(): + if False: + yield 1 + + return _gen() + + anext_first_wrapper = _AsyncStreamingResponseWrapper( + AsyncStreamWithAiter(), mock_tracer + ) + with pytest.raises(StopAsyncIteration): + await anext_first_wrapper.__anext__() + + +def test_safe_decode_utf8(): + assert _safe_decode_utf8(None) == "" + assert _safe_decode_utf8("hello") == "hello" + assert _safe_decode_utf8(b"world") == "world" + assert _safe_decode_utf8(123) == "123" + + +def test_prepare_attempt(): + mock_tracer = MagicMock() + mock_tracer.client_attributes = {} + call_details = MagicMock() + call_details.metadata = [ + ("google-cloud-resource-prefix", "projects/p/instances/i/databases/d") + ] + call_details.method = "/google.spanner.v1.Spanner/ExecuteSql" + + MetricsInterceptor._prepare_attempt(mock_tracer, call_details) + + mock_tracer.set_project.assert_called_with("p") + mock_tracer.set_instance.assert_called_with("i") + mock_tracer.set_database.assert_called_with("d") + mock_tracer.set_method.assert_called_with("Spanner.ExecuteSql") + mock_tracer.record_attempt_start.assert_called_once() diff --git a/packages/google-cloud-spanner/tests/unit/test_metrics_tracer.py b/packages/google-cloud-spanner/tests/unit/test_metrics_tracer.py index c645905c40ee..243859e529f4 100644 --- a/packages/google-cloud-spanner/tests/unit/test_metrics_tracer.py +++ b/packages/google-cloud-spanner/tests/unit/test_metrics_tracer.py @@ -284,10 +284,59 @@ def test_extract_front_end_latencies(): ] assert MetricsTracer.extract_front_end_latencies(metadata_list) == (123, 100) + # Combined header in single value (standard Spanner wire response) + combined = [("server-timing", "gfet4t7; dur=55, afe; dur=23")] + assert MetricsTracer.extract_front_end_latencies(combined) == (55, 23) + + # Bytes header key in list of tuples + bytes_list = [(b"server-timing", "gfet4t7; dur=55, afe; dur=23")] + assert MetricsTracer.extract_front_end_latencies(bytes_list) == (55, 23) + # Valid metadata dict metadata_dict = {"server-timing": "gfet4t7; dur=456"} assert MetricsTracer.extract_front_end_latencies(metadata_dict) == (456, None) + # Metadata dict with bytes key and combined value + metadata_dict_bytes = {b"server-timing": "gfet4t7; dur=55, afe; dur=23"} + assert MetricsTracer.extract_front_end_latencies(metadata_dict_bytes) == (55, 23) + + # Metadata dict with mixed case key + metadata_dict_case = {"Server-Timing": "gfet4t7; dur=55, afe; dur=23"} + assert MetricsTracer.extract_front_end_latencies(metadata_dict_case) == (55, 23) + + # Metadata dict with list of values + metadata_dict_list = {"server-timing": ["gfet4t7; dur=12", "afe; dur=34"]} + assert MetricsTracer.extract_front_end_latencies(metadata_dict_list) == (12, 34) + + # Floating point latencies truncated to int + float_headers = [("server-timing", "gfet4t7; dur=55.8, afe; dur=23.2")] + assert MetricsTracer.extract_front_end_latencies(float_headers) == (55, 23) + + # Bytes header value + bytes_val = [("server-timing", b"gfet4t7; dur=55, afe; dur=23")] + assert MetricsTracer.extract_front_end_latencies(bytes_val) == (55, 23) + + # Generator / custom iterable (simulating grpc.aio.Metadata) + def timing_generator(): + yield ("unrelated", "1") + yield ("server-timing", "gfet4t7; dur=77, afe; dur=88") + + assert MetricsTracer.extract_front_end_latencies(timing_generator()) == (77, 88) + + # Dict with multiple case variants (ensuring no premature loop termination) + metadata_dict_multiple_cases = { + "Server-Timing": "gfet4t7; dur=55", + "server-timing": "afe; dur=23", + } + assert MetricsTracer.extract_front_end_latencies(metadata_dict_multiple_cases) == ( + 55, + 23, + ) + + # Sequence with list-of-values + metadata_seq_list = [("server-timing", ["gfet4t7; dur=12", "afe; dur=34"])] + assert MetricsTracer.extract_front_end_latencies(metadata_seq_list) == (12, 34) + # Missing header assert MetricsTracer.extract_front_end_latencies([("other-header", "val")]) == ( None, @@ -295,6 +344,45 @@ def test_extract_front_end_latencies(): ) assert MetricsTracer.extract_front_end_latencies(None) == (None, None) + # Non-iterable or malformed metadata + assert MetricsTracer.extract_front_end_latencies(12345) == (None, None) + assert MetricsTracer.extract_front_end_latencies([("single_item",)]) == (None, None) + assert MetricsTracer.extract_front_end_latencies( + [("server-timing", "gfet4t7; dur=invalid")] + ) == (None, None) + # Trigger ValueError in float conversion (dur=.) + assert MetricsTracer.extract_front_end_latencies( + [("server-timing", "gfet4t7; dur=.")] + ) == (None, None) + assert MetricsTracer.extract_front_end_latencies( + [("server-timing", "afe; dur=.")] + ) == (None, None) + # Non-string, non-bytes header value + assert MetricsTracer.extract_front_end_latencies([("server-timing", 12345)]) == ( + None, + None, + ) + # Non-decodable bytes fallback + assert MetricsTracer.extract_front_end_latencies( + [("server-timing", b"\xff\xfe\xfd")] + ) == (None, None) + + # Non-matching bytes key and non-str non-bytes key + assert MetricsTracer.extract_front_end_latencies( + [(b"unrelated", "val"), (12345, "val")] + ) == (None, None) + + # Empty header value + assert MetricsTracer.extract_front_end_latencies([("server-timing", "")]) == ( + None, + None, + ) + + # Separate headers where first sets AFE, second sets GFE + assert MetricsTracer.extract_front_end_latencies( + [("server-timing", "afe; dur=20"), ("server-timing", "gfet4t7; dur=30")] + ) == (30, 20) + def test_record_front_end_metrics(metrics_tracer): mock_gfe_latency = mock.create_autospec(Histogram, instance=True) @@ -324,6 +412,15 @@ def test_record_front_end_metrics(metrics_tracer): assert mock_afe_latency.record.call_count == 1 assert mock_afe_missing.add.call_count == 1 + # When disabled, record_front_end_metrics does nothing + metrics_tracer.enabled = False + metrics_tracer.record_front_end_metrics( + [("server-timing", "gfet4t7; dur=88"), ("server-timing", "afe; dur=90")] + ) + assert mock_gfe_latency.record.call_count == 1 + assert mock_afe_latency.record.call_count == 1 + metrics_tracer.enabled = True + def test_record_afe_latency(metrics_tracer): mock_afe_latency = mock.create_autospec(Histogram, instance=True) @@ -367,3 +464,126 @@ def test_record_afe_connectivity_error_count(metrics_tracer): metrics_tracer.record_afe_connectivity_error_count() assert mock_afe_missing.add.call_count == 1 metrics_tracer.enabled = True + + +def test_attribute_caching_and_invalidation(metrics_tracer): + # Test attempt attribute caching + metrics_tracer.current_op.new_attempt() + metrics_tracer.current_op.current_attempt.status = "OK" + + first_attempt_attrs = metrics_tracer._create_attempt_otel_attributes() + second_attempt_attrs = metrics_tracer._create_attempt_otel_attributes() + assert first_attempt_attrs == second_attempt_attrs + # Same cached object returned + assert first_attempt_attrs is second_attempt_attrs + + # Changing attempt status returns new object + metrics_tracer.current_op.current_attempt.status = "UNAVAILABLE" + third_attempt_attrs = metrics_tracer._create_attempt_otel_attributes() + assert third_attempt_attrs["status"] == "UNAVAILABLE" + assert third_attempt_attrs is not first_attempt_attrs + + # Test operation attribute caching + first_op_attrs = metrics_tracer._create_operation_otel_attributes() + second_op_attrs = metrics_tracer._create_operation_otel_attributes() + assert first_op_attrs is second_op_attrs + + # Non-matching AFE and GFE patterns when substring is present (lookbehind boundary check) + assert MetricsTracer.extract_front_end_latencies( + [("server-timing", "safe; dur=55")] + ) == (None, None) + assert MetricsTracer.extract_front_end_latencies( + [("server-timing", "custom-afe; dur=20")] + ) == (None, None) + assert MetricsTracer.extract_front_end_latencies( + [("server-timing", "x-gfet4t7; dur=15")] + ) == (None, None) + assert MetricsTracer.extract_front_end_latencies( + [ + ( + "server-timing", + "safe; dur=55, afe; dur=25, x-gfet4t7; dur=10, gfet4t7; dur=35", + ) + ] + ) == (35, 25) + + # Setter method invalidates cache + empty_tracer = MetricsTracer( + enabled=True, + instrument_attempt_latency=mock.MagicMock(), + instrument_attempt_counter=mock.MagicMock(), + instrument_operation_latency=mock.MagicMock(), + instrument_operation_counter=mock.MagicMock(), + client_attributes={}, + instrument_gfe_latency=mock.MagicMock(), + instrument_gfe_connectivity_error_count=mock.MagicMock(), + instrument_afe_latency=mock.MagicMock(), + instrument_afe_connectivity_error_count=mock.MagicMock(), + ) + empty_tracer.set_project("new-project") + assert empty_tracer.client_attributes["project_id"] == "new-project" + + metrics_tracer.set_instance("updated-instance") + invalidated_attrs = metrics_tracer._create_attempt_otel_attributes() + assert invalidated_attrs["instance_id"] == "updated-instance" + assert invalidated_attrs is not third_attempt_attrs + + # Direct mutation of client_attributes invalidates cache via _ObservableDict + before_direct = metrics_tracer._create_attempt_otel_attributes() + metrics_tracer.client_attributes["database"] = "mutated_db" + after_direct = metrics_tracer._create_attempt_otel_attributes() + assert after_direct["database"] == "mutated_db" + assert after_direct is not before_direct + + # _ObservableDict copy returns standard dict + attrs_copy = metrics_tracer.client_attributes.copy() + assert type(attrs_copy) is dict + assert attrs_copy["database"] == "mutated_db" + + # Other observable dict operations trigger invalidation + metrics_tracer.client_attributes.update({"instance_id": "updated_via_update"}) + assert ( + metrics_tracer._create_attempt_otel_attributes()["instance_id"] + == "updated_via_update" + ) + # setdefault when key is new + metrics_tracer.client_attributes.setdefault("new_key", "default_val") + assert metrics_tracer._create_attempt_otel_attributes()["new_key"] == "default_val" + # setdefault when key already exists + assert ( + metrics_tracer.client_attributes.setdefault("new_key", "other_val") + == "default_val" + ) + # pop + metrics_tracer.client_attributes.pop("new_key") + assert "new_key" not in metrics_tracer._create_attempt_otel_attributes() + + # delitem + metrics_tracer.client_attributes["to_delete"] = "val" + del metrics_tracer.client_attributes["to_delete"] + assert "to_delete" not in metrics_tracer._create_attempt_otel_attributes() + + # popitem + metrics_tracer.client_attributes["to_pop"] = "val" + metrics_tracer.client_attributes.popitem() + + # ObservableDict with no on_change callback + from google.cloud.spanner_v1.metrics.metrics_tracer import _ObservableDict + + no_callback_dict = _ObservableDict({"a": 1}) + no_callback_dict["b"] = 2 + del no_callback_dict["a"] + no_callback_dict.update({"c": 3}) + no_callback_dict.setdefault("d", 4) + no_callback_dict.pop("c") + no_callback_dict.popitem() + no_callback_dict.clear() + + # clear on client_attributes + metrics_tracer.client_attributes.clear() + assert metrics_tracer._create_attempt_otel_attributes() == {"status": "UNAVAILABLE"} + + # Ensure client_attributes and cached attributes are dict instances for backward compatibility + assert isinstance(metrics_tracer.client_attributes, dict) + assert isinstance(first_attempt_attrs, dict) + assert isinstance(first_op_attrs, dict)