diff --git a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/README.md b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/README.md index 9619715c8531..179e9b2d9464 100644 --- a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/README.md +++ b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/README.md @@ -1,4 +1,4 @@ -# Custom Metric Exporter +# Custom Metric Exporter The custom metric exporter, as defined in [metrics_exporter.py](./metrics_exporter.py), is designed to work in conjunction with OpenTelemetry and the Spanner client. It converts data into its protobuf equivalent and sends it to Google Cloud Monitoring. ## Filtering Criteria @@ -10,8 +10,10 @@ The exporter filters metrics based on the following conditions, utilizing values * `attempt_count` * `operation_latencies` * `operation_count` - * `gfe_latency` - * `gfe_missing_header_count` + * `gfe_latencies` + * `gfe_connectivity_error_count` + * `afe_latencies` + * `afe_connectivity_error_count` ## Service Endpoint The exporter sends metrics to the Google Cloud Monitoring [service endpoint](https://cloud.google.com/python/docs/reference/monitoring/latest/google.cloud.monitoring_v3.services.metric_service.MetricServiceClient#google_cloud_monitoring_v3_services_metric_service_MetricServiceClient_create_service_time_series), distinct from the regular client endpoint. This service endpoint operates under a different quota limit than the user endpoint and features an additional server-side filter that only permits a predefined set of metrics to pass through. diff --git a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/constants.py b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/constants.py index a5f709881b12..fa5f5ca4d98d 100644 --- a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/constants.py +++ b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/constants.py @@ -58,13 +58,19 @@ METRIC_NAME_ATTEMPT_LATENCIES = "attempt_latencies" METRIC_NAME_OPERATION_COUNT = "operation_count" METRIC_NAME_ATTEMPT_COUNT = "attempt_count" -METRIC_NAME_GFE_LATENCY = "gfe_latency" -METRIC_NAME_GFE_MISSING_HEADER_COUNT = "gfe_missing_header_count" +METRIC_NAME_GFE_LATENCY = "gfe_latencies" +METRIC_NAME_GFE_CONNECTIVITY_ERROR_COUNT = "gfe_connectivity_error_count" +METRIC_NAME_AFE_LATENCY = "afe_latencies" +METRIC_NAME_AFE_CONNECTIVITY_ERROR_COUNT = "afe_connectivity_error_count" METRIC_NAMES = [ METRIC_NAME_OPERATION_LATENCIES, METRIC_NAME_ATTEMPT_LATENCIES, METRIC_NAME_OPERATION_COUNT, METRIC_NAME_ATTEMPT_COUNT, + METRIC_NAME_GFE_LATENCY, + METRIC_NAME_GFE_CONNECTIVITY_ERROR_COUNT, + METRIC_NAME_AFE_LATENCY, + METRIC_NAME_AFE_CONNECTIVITY_ERROR_COUNT, ] METRIC_EXPORT_INTERVAL_MS = 60000 # 1 Minute 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 3e38c4e0191d..1a4e1863ea4b 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,14 +14,20 @@ """Interceptor for collecting Cloud Spanner metrics.""" +import inspect +import logging +import os import re -from typing import Dict +from typing import Any, Dict +import grpc from grpc_interceptor import ClientInterceptor from .constants import GOOGLE_CLOUD_RESOURCE_KEY, SPANNER_METHOD_PREFIX from .spanner_metrics_tracer_factory import SpannerMetricsTracerFactory +logger = logging.getLogger(__name__) + class MetricsInterceptor(ClientInterceptor): """Interceptor that collects metrics for Cloud Spanner operations.""" @@ -47,27 +53,33 @@ def _parse_resource_path(path: str) -> dict: return {} @staticmethod - def _extract_resource_from_path(metadata: Dict[str, str]) -> Dict[str, str]: + def _extract_resource_from_path(metadata: Any) -> Dict[str, str]: """ Extracts resource information from the metadata based on the path. - This method iterates through the metadata dictionary to find the first tuple containing the key 'google-cloud-resource-prefix'. It then extracts the path from this tuple and parses it to extract project, instance, and database information using the _parse_resource_path method. - Args: - metadata (Dict[str, str]): A dictionary containing metadata information. + metadata (Any): A sequence or dictionary containing metadata information. Returns: Dict[str, str]: A dictionary containing extracted project, instance, and database information. """ - # Extract resource info from the first metadata tuple containing :path - path = next( - (value for key, value in metadata if key == GOOGLE_CLOUD_RESOURCE_KEY), "" - ) + 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 - def _set_metrics_tracer_attributes(self, resources: Dict[str, str]) -> None: + @staticmethod + def _set_metrics_tracer_attributes(resources: Dict[str, str]) -> None: """ Sets the metric tracer attributes based on the provided resources. @@ -115,17 +127,343 @@ def intercept(self, invoked_method, request_or_iterator, call_details): self._set_metrics_tracer_attributes(resources) ## Format method to be be spanner. - method_name = call_details.method.removeprefix(SPANNER_METHOD_PREFIX).replace( - "/", "." - ) + 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() + + if os.environ.get("SPANNER_DISABLE_AFE_SERVER_TIMING", "").lower() != "true": + metadata = list(call_details.metadata or []) + metadata.append(("x-goog-spanner-enable-afe-server-timing", "true")) + call_details = call_details._replace(metadata=metadata) + response = invoked_method(request_or_iterator, call_details) - tracer.record_attempt_completion() - # Process and send GFE metrics if enabled - if tracer.gfe_enabled: - metadata = response.initial_metadata() - tracer.record_gfe_metrics(metadata) + return _wrap_response(response, tracer) + + +def _wrap_response(response: Any, tracer: Any) -> Any: + """Wraps the response if it is streaming, or records metrics immediately if unary.""" + if hasattr(response, "__next__"): + return _StreamingResponseWrapper(response, tracer) + else: + # Unary call: execute completion and record metrics immediately + try: + tracer.record_attempt_completion() + metadata = [] + if hasattr(response, "initial_metadata"): + try: + metadata.extend(response.initial_metadata() or []) + except Exception as e: + logger.warning(f"Failed to retrieve initial metadata: {e}") + tracer.record_front_end_metrics(metadata) + except Exception as e: + logger.warning(f"Failed to record metrics: {e}") return response + + +class AsyncMetricsInterceptor( + grpc.aio.UnaryUnaryClientInterceptor, + grpc.aio.UnaryStreamClientInterceptor, + grpc.aio.StreamUnaryClientInterceptor, + grpc.aio.StreamStreamClientInterceptor, +): + """Async Interceptor that collects metrics for Cloud Spanner operations.""" + + async def intercept_unary_unary(self, continuation, client_call_details, request): + return await self._async_intercept(continuation, client_call_details, request) + + async def intercept_unary_stream(self, continuation, client_call_details, request): + return await self._async_intercept(continuation, client_call_details, request) + + async def intercept_stream_unary( + self, continuation, client_call_details, request_iterator + ): + return await self._async_intercept( + continuation, client_call_details, request_iterator + ) + + async def intercept_stream_stream( + self, continuation, client_call_details, request_iterator + ): + return await self._async_intercept( + continuation, client_call_details, request_iterator + ) + + async def _async_intercept( + self, + continuation: Any, + call_details: grpc.ClientCallDetails, + request_or_iterator: Any, + ) -> Any: + # Implementation for async interceptor + factory = SpannerMetricsTracerFactory() + tracer = SpannerMetricsTracerFactory.get_current_tracer() + 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() + + if os.environ.get("SPANNER_DISABLE_AFE_SERVER_TIMING", "").lower() != "true": + metadata = list(call_details.metadata or []) + metadata.append(("x-goog-spanner-enable-afe-server-timing", "true")) + call_details = call_details._replace(metadata=metadata) + + response = await continuation(call_details, request_or_iterator) + if hasattr(response, "__anext__"): + return _AsyncStreamingResponseWrapper(response, tracer) + else: + return _AsyncUnaryResponseWrapper(response, tracer) + + +class _StreamingResponseWrapper: + """Wrapper for streaming RPC response iterators to defer metrics recording.""" + + def __init__(self, response, tracer): + self._response = response + self._tracer = tracer + self._metrics_recorded = False + self._iterator = None + + def __iter__(self): + self._iterator = iter(self._response) + return self + + def __next__(self): + if self._iterator is None: + self._iterator = iter(self._response) + try: + return next(self._iterator) + except StopIteration: + self._record_metrics() + raise + except Exception: + self._record_metrics() + raise + + 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: + metadata.extend(self._response.initial_metadata() 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): + try: + self._record_metrics() + except Exception: + pass + + 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.""" + + def __init__(self, response, tracer): + self._response = response + self._tracer = tracer + self._metrics_recorded = False + + 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 __await__(self): + async def _wait(): + try: + return await self._response + finally: + await self._record_metrics() + + return _wait().__await__() + + 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) + + +class _AsyncStreamingResponseWrapper( + 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 + 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__() + else: + self._iterator = self._response + return self + + async def __anext__(self): + if self._iterator is None: + if hasattr(self._response, "__aiter__"): + self._iterator = self._response.__aiter__() + else: + self._iterator = self._response + try: + return await self._iterator.__anext__() + except StopAsyncIteration: + await self._record_metrics() + raise + 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 f79869948f99..f24a09c594c0 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 @@ -19,8 +19,10 @@ while the helper classes provide additional functionality and context for the metrics being traced. """ +import os +import re from datetime import datetime -from typing import Dict +from typing import Any, Dict, Optional from grpc import StatusCode @@ -183,7 +185,9 @@ class should not have any knowledge about the observability framework used for m _instrument_operation_counter: "Counter" _instrument_operation_latency: "Histogram" _instrument_gfe_latency: "Histogram" - _instrument_gfe_missing_header_count: "Counter" + _instrument_gfe_connectivity_error_count: "Counter" + _instrument_afe_latency: "Histogram" + _instrument_afe_connectivity_error_count: "Counter" current_op: MetricOpTracer enabled: bool gfe_enabled: bool @@ -197,6 +201,10 @@ def __init__( instrument_operation_latency: "Histogram", instrument_operation_counter: "Counter", client_attributes: Dict[str, str], + instrument_gfe_latency: "Histogram", + instrument_gfe_connectivity_error_count: "Counter", + instrument_afe_latency: "Histogram", + instrument_afe_connectivity_error_count: "Counter", gfe_enabled: bool = False, ): """ @@ -214,6 +222,10 @@ def __init__( instrument_operation_counter (Counter): Instrument for counting operations. client_attributes (Dict[str, str]): Dictionary of client attributes used for metrics tracing. gfe_enabled (bool, optional): Indicates if GFE metrics are enabled. Defaults to False. + instrument_gfe_latency (Histogram): Instrument for measuring GFE latency. + instrument_gfe_connectivity_error_count (Counter): Instrument for counting GFE connectivity errors. + instrument_afe_latency (Histogram): Instrument for measuring AFE latency. + instrument_afe_connectivity_error_count (Counter): Instrument for counting AFE connectivity errors. """ self.current_op = MetricOpTracer() self._client_attributes = client_attributes @@ -221,8 +233,16 @@ def __init__( self._instrument_attempt_counter = instrument_attempt_counter self._instrument_operation_latency = instrument_operation_latency self._instrument_operation_counter = instrument_operation_counter + self._instrument_gfe_latency = instrument_gfe_latency + self._instrument_gfe_connectivity_error_count = ( + instrument_gfe_connectivity_error_count + ) + self._instrument_afe_latency = instrument_afe_latency + self._instrument_afe_connectivity_error_count = ( + instrument_afe_connectivity_error_count + ) self.enabled = enabled - self.gfe_enabled = gfe_enabled + self.gfe_enabled = True @staticmethod def _get_ms_time_diff(start: datetime, end: datetime) -> float: @@ -399,22 +419,143 @@ def record_gfe_latency(self, latency: int) -> None: Args: latency (int): The latency duration to be recorded. """ - if not self.enabled or not HAS_OPENTELEMETRY_INSTALLED or not self.gfe_enabled: + if ( + not self.enabled + or not HAS_OPENTELEMETRY_INSTALLED + or not getattr(self, "_instrument_gfe_latency", None) + ): return self._instrument_gfe_latency.record( - amount=latency, attributes=self.client_attributes + amount=latency, attributes=self._create_attempt_otel_attributes() + ) + + def record_gfe_connectivity_error_count(self) -> None: + """ + Increments the counter for GFE connectivity errors. + """ + if ( + not self.enabled + or not HAS_OPENTELEMETRY_INSTALLED + or not getattr(self, "_instrument_gfe_connectivity_error_count", None) + ): + return + self._instrument_gfe_connectivity_error_count.add( + amount=1, attributes=self._create_attempt_otel_attributes() + ) + + def record_afe_latency(self, latency: int) -> None: + """ + Records the AFE latency using the Histogram instrument. + + Args: + latency (int): The latency duration to be recorded. + """ + if ( + not self.enabled + or not HAS_OPENTELEMETRY_INSTALLED + or not getattr(self, "_instrument_afe_latency", None) + or os.environ.get("SPANNER_DISABLE_AFE_SERVER_TIMING", "").lower() == "true" + ): + return + self._instrument_afe_latency.record( + amount=latency, attributes=self._create_attempt_otel_attributes() ) - def record_gfe_missing_header_count(self) -> None: + def record_afe_connectivity_error_count(self) -> None: """ - Increments the counter for missing GFE headers. + Increments the counter for AFE connectivity errors. """ - if not self.enabled or not HAS_OPENTELEMETRY_INSTALLED or not self.gfe_enabled: + if ( + not self.enabled + or not HAS_OPENTELEMETRY_INSTALLED + or not getattr(self, "_instrument_afe_connectivity_error_count", None) + or os.environ.get("SPANNER_DISABLE_AFE_SERVER_TIMING", "").lower() == "true" + ): return - self._instrument_gfe_missing_header_count.add( - amount=1, attributes=self.client_attributes + self._instrument_afe_connectivity_error_count.add( + amount=1, attributes=self._create_attempt_otel_attributes() ) + @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. + """ + 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() in ("server-timing", "server_timing"): + if isinstance(val, (list, tuple)): + header_vals.extend(val) + else: + header_vals.append(val) + + gfe_latency = None + afe_latency = None + + for header_val in header_vals: + if not header_val: + 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(?:t4t7)?;\s*dur=([0-9.]+)", header_val) + if match: + try: + afe_latency = int(float(match.group(1))) + except ValueError: + pass + + 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. + """ + if not self.enabled or not HAS_OPENTELEMETRY_INSTALLED: + return + gfe_latency, afe_latency = self.extract_front_end_latencies(metadata) + + if gfe_latency is not None: + self.record_gfe_latency(gfe_latency) + else: + self.record_gfe_connectivity_error_count() + + if afe_latency is not None: + self.record_afe_latency(afe_latency) + else: + self.record_afe_connectivity_error_count() + def _create_operation_otel_attributes(self) -> dict: """ Create additional attributes for operation metrics tracing. diff --git a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_tracer_factory.py b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_tracer_factory.py index f22d285c9750..804fb46f1faf 100644 --- a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_tracer_factory.py +++ b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_tracer_factory.py @@ -22,10 +22,12 @@ METRIC_LABEL_KEY_CLIENT_UID, METRIC_LABEL_KEY_DATABASE, METRIC_LABEL_KEY_DIRECT_PATH_ENABLED, + METRIC_NAME_AFE_CONNECTIVITY_ERROR_COUNT, + METRIC_NAME_AFE_LATENCY, METRIC_NAME_ATTEMPT_COUNT, METRIC_NAME_ATTEMPT_LATENCIES, + METRIC_NAME_GFE_CONNECTIVITY_ERROR_COUNT, METRIC_NAME_GFE_LATENCY, - METRIC_NAME_GFE_MISSING_HEADER_COUNT, METRIC_NAME_OPERATION_COUNT, METRIC_NAME_OPERATION_LATENCIES, MONITORED_RES_LABEL_KEY_CLIENT_HASH, @@ -56,7 +58,9 @@ class MetricsTracerFactory: _instrument_operation_latency: "Histogram" _instrument_operation_counter: "Counter" _instrument_gfe_latency: "Histogram" - _instrument_gfe_missing_header_count: "Counter" + _instrument_gfe_connectivity_error_count: "Counter" + _instrument_afe_latency: "Histogram" + _instrument_afe_connectivity_error_count: "Counter" _client_attributes: Dict[str, str] @property @@ -85,6 +89,7 @@ def __init__(self, enabled: bool, service_name: str): project (str): The project ID for the monitored resource. """ self.enabled = enabled + self.gfe_enabled = True self._create_metric_instruments(service_name) self._client_attributes = {} @@ -268,6 +273,11 @@ def create_metrics_tracer(self) -> MetricsTracer: instrument_operation_latency=self._instrument_operation_latency, instrument_operation_counter=self._instrument_operation_counter, client_attributes=self._client_attributes.copy(), + gfe_enabled=True, + instrument_gfe_latency=self._instrument_gfe_latency, + instrument_gfe_connectivity_error_count=self._instrument_gfe_connectivity_error_count, + instrument_afe_latency=self._instrument_afe_latency, + instrument_afe_connectivity_error_count=self._instrument_afe_connectivity_error_count, ) return metrics_tracer @@ -320,8 +330,20 @@ def _create_metric_instruments(self, service_name: str) -> None: description="GFE Latency.", ) - self._instrument_gfe_missing_header_count = meter.create_counter( - name=METRIC_NAME_GFE_MISSING_HEADER_COUNT, + self._instrument_gfe_connectivity_error_count = meter.create_counter( + name=METRIC_NAME_GFE_CONNECTIVITY_ERROR_COUNT, unit="1", - description="GFE missing header count.", + description="GFE connectivity error count.", + ) + + self._instrument_afe_latency = meter.create_histogram( + name=METRIC_NAME_AFE_LATENCY, + unit="ms", + description="AFE Latency.", + ) + + self._instrument_afe_connectivity_error_count = meter.create_counter( + name=METRIC_NAME_AFE_CONNECTIVITY_ERROR_COUNT, + unit="1", + description="AFE connectivity error count.", ) diff --git a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/spanner_metrics_tracer_factory.py b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/spanner_metrics_tracer_factory.py index 6fc5956582c1..7886e555f120 100644 --- a/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/spanner_metrics_tracer_factory.py +++ b/packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/spanner_metrics_tracer_factory.py @@ -51,9 +51,7 @@ class SpannerMetricsTracerFactory(MetricsTracerFactory): "current_metrics_tracer", default=None ) - def __new__( - cls, enabled: bool = True, gfe_enabled: bool = False - ) -> "SpannerMetricsTracerFactory": + def __new__(cls, enabled: bool = True) -> "SpannerMetricsTracerFactory": """ Create a new instance of SpannerMetricsTracerFactory if it doesn't already exist. @@ -63,7 +61,6 @@ def __new__( Args: enabled (bool): A flag indicating whether metrics tracing is enabled. Defaults to True. - gfe_enabled (bool): A flag indicating whether GFE metrics are enabled. Defaults to False. Returns: SpannerMetricsTracerFactory: The singleton instance of SpannerMetricsTracerFactory. @@ -83,7 +80,7 @@ def __new__( cls._generate_client_hash(client_uid) ) cls._metrics_tracer_factory.set_location(_get_cloud_region()) - cls._metrics_tracer_factory.gfe_enabled = gfe_enabled + cls._metrics_tracer_factory.gfe_enabled = True if cls._metrics_tracer_factory.enabled != enabled: cls._metrics_tracer_factory.enabled = enabled diff --git a/packages/google-cloud-spanner/google/cloud/spanner_v1/services/spanner/transports/grpc_asyncio.py b/packages/google-cloud-spanner/google/cloud/spanner_v1/services/spanner/transports/grpc_asyncio.py index c688b31eefc4..c56ab0112d23 100644 --- a/packages/google-cloud-spanner/google/cloud/spanner_v1/services/spanner/transports/grpc_asyncio.py +++ b/packages/google-cloud-spanner/google/cloud/spanner_v1/services/spanner/transports/grpc_asyncio.py @@ -32,7 +32,10 @@ from google.protobuf.json_format import MessageToJson from grpc.experimental import aio # type: ignore -from google.cloud.spanner_v1.metrics.metrics_interceptor import MetricsInterceptor +from google.cloud.spanner_v1.metrics.metrics_interceptor import ( + AsyncMetricsInterceptor, + MetricsInterceptor, +) from google.cloud.spanner_v1.types import ( commit_response, location, @@ -327,6 +330,26 @@ def __init__( ], ) + if metrics_interceptor is not None: + self._metrics_interceptor = AsyncMetricsInterceptor() + # Attach interceptor directly since grpc.aio does not provide intercept_channel. + if hasattr(self._grpc_channel, "_unary_unary_interceptors"): + self._grpc_channel._unary_unary_interceptors.append( + self._metrics_interceptor + ) + if hasattr(self._grpc_channel, "_unary_stream_interceptors"): + self._grpc_channel._unary_stream_interceptors.append( + self._metrics_interceptor + ) + if hasattr(self._grpc_channel, "_stream_unary_interceptors"): + self._grpc_channel._stream_unary_interceptors.append( + self._metrics_interceptor + ) + if hasattr(self._grpc_channel, "_stream_stream_interceptors"): + self._grpc_channel._stream_stream_interceptors.append( + self._metrics_interceptor + ) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/google-cloud-spanner/tests/_helpers.py b/packages/google-cloud-spanner/tests/_helpers.py index 83aecfd8b2f6..519a79f5a447 100644 --- a/packages/google-cloud-spanner/tests/_helpers.py +++ b/packages/google-cloud-spanner/tests/_helpers.py @@ -13,6 +13,9 @@ try: from opentelemetry import trace + from opentelemetry.resourcedetector.gcp_resource_detector import ( # noqa: F401 + GoogleCloudResourceDetector, + ) from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( diff --git a/packages/google-cloud-spanner/tests/mockserver_tests/test_frontend_metrics.py b/packages/google-cloud-spanner/tests/mockserver_tests/test_frontend_metrics.py new file mode 100644 index 000000000000..367104260c4d --- /dev/null +++ b/packages/google-cloud-spanner/tests/mockserver_tests/test_frontend_metrics.py @@ -0,0 +1,222 @@ +# Copyright 2025 Google LLC All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import os +from unittest import mock + +import grpc +from google.api_core.client_options import ClientOptions +from google.auth.credentials import AnonymousCredentials +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics.export import InMemoryMetricReader + +import google.cloud.spanner_v1.client as client_mod +from google.cloud.spanner_v1 import Client +from google.cloud.spanner_v1.metrics.metrics_interceptor import MetricsInterceptor +from google.cloud.spanner_v1.metrics.spanner_metrics_tracer_factory import ( + SpannerMetricsTracerFactory, +) +from google.cloud.spanner_v1.pool import FixedSizePool +from tests.mockserver_tests.mock_server_test_base import ( + MockServerTestBase, + add_select1_result, +) + + +class TestFrontendMetricsIntegration(MockServerTestBase): + def setUp(self): + super().setUp() + os.environ["SPANNER_DISABLE_BUILTIN_METRICS"] = "false" + SpannerMetricsTracerFactory._metrics_tracer_factory = None + client_mod._metrics_monitor_initialized = False + + def tearDown(self): + super().tearDown() + os.environ["SPANNER_DISABLE_BUILTIN_METRICS"] = "true" + SpannerMetricsTracerFactory._metrics_tracer_factory = None + client_mod._metrics_monitor_initialized = False + + def test_gfe_metrics_exported(self): + add_select1_result() + reader = InMemoryMetricReader() + meter_provider = MeterProvider(metric_readers=[reader]) + + orig_call = grpc._channel._UnaryStreamMultiCallable.__call__ + orig_initial_metadata = grpc._channel._MultiThreadedRendezvous.initial_metadata + orig_trailing_metadata = ( + grpc._channel._MultiThreadedRendezvous.trailing_metadata + ) + + def custom_initial_metadata(self): + mocked = getattr(self, "_is_execute_streaming_sql_mock", False) + if mocked: + return (("server-timing", "gfet4t7; dur=55, afe; dur=23"),) + return orig_initial_metadata(self) + + def custom_trailing_metadata(self): + mocked = getattr(self, "_is_execute_streaming_sql_mock", False) + if mocked: + return (("server-timing", "gfet4t7; dur=55, afe; dur=23"),) + return orig_trailing_metadata(self) + + def custom_call(self_callable, request, *args, **kwargs): + method = getattr(self_callable, "_method", b"") + method_str = method.decode("utf-8") if isinstance(method, bytes) else method + response = orig_call(self_callable, request, *args, **kwargs) + if "ExecuteStreamingSql" in method_str: + response._is_execute_streaming_sql_mock = True + return response + + try: + with ( + mock.patch( + "google.cloud.spanner_v1.metrics.metrics_tracer_factory.get_meter_provider", + return_value=meter_provider, + ), + mock.patch( + "google.cloud.spanner_v1.client.MeterProvider", + return_value=meter_provider, + ), + mock.patch( + "google.cloud.spanner_v1.client._get_spanner_emulator_host", + return_value=None, + ), + mock.patch( + "grpc._channel._UnaryStreamMultiCallable.__call__", + custom_call, + ), + mock.patch( + "grpc._channel._MultiThreadedRendezvous.initial_metadata", + custom_initial_metadata, + ), + mock.patch( + "grpc._channel._MultiThreadedRendezvous.trailing_metadata", + custom_trailing_metadata, + ), + ): + client = Client( + project="p", + credentials=AnonymousCredentials(), + client_options=ClientOptions( + api_endpoint="localhost:" + str(MockServerTestBase.port), + ), + ) + instance = client.instance("test-instance") + database = instance.database( + "test-database", + pool=FixedSizePool(size=10), + enable_interceptors_in_tests=True, + ) + database._interceptors.append(MetricsInterceptor()) + database._spanner_api = ( + None # Force recreation with the new interceptor + ) + + with database.snapshot() as snapshot: + results = snapshot.execute_sql("select 1") + # Consume the streaming results to complete the stream + list(results) + + metric_data = reader.get_metrics_data() + self.assertIsNotNone(metric_data) + metrics = { + metric.name: metric + for rm in metric_data.resource_metrics + for sm in rm.scope_metrics + for metric in sm.metrics + } + + self.assertIn("gfe_latencies", metrics, f"Metrics: {list(metrics.keys())}") + gfe_metric = metrics["gfe_latencies"] + point = next(iter(gfe_metric.data.data_points)) + self.assertEqual(point.sum, 55) + + self.assertIn("afe_latencies", metrics, f"Metrics: {list(metrics.keys())}") + afe_metric = metrics["afe_latencies"] + point = next(iter(afe_metric.data.data_points)) + self.assertEqual(point.sum, 23) + + finally: + pass + + def test_gfe_missing_header_count_exported(self): + add_select1_result() + reader = InMemoryMetricReader() + meter_provider = MeterProvider(metric_readers=[reader]) + + try: + with ( + mock.patch( + "google.cloud.spanner_v1.metrics.metrics_tracer_factory.get_meter_provider", + return_value=meter_provider, + ), + mock.patch( + "google.cloud.spanner_v1.client.MeterProvider", + return_value=meter_provider, + ), + mock.patch( + "google.cloud.spanner_v1.client._get_spanner_emulator_host", + return_value=None, + ), + ): + client = Client( + project="p", + credentials=AnonymousCredentials(), + client_options=ClientOptions( + api_endpoint="localhost:" + str(MockServerTestBase.port), + ), + ) + instance = client.instance("test-instance") + database = instance.database( + "test-database", + pool=FixedSizePool(size=10), + enable_interceptors_in_tests=True, + ) + database._interceptors.append(MetricsInterceptor()) + database._spanner_api = ( + None # Force recreation with the new interceptor + ) + + with database.snapshot() as snapshot: + results = snapshot.execute_sql("select 1") + list(results) + + metric_data = reader.get_metrics_data() + self.assertIsNotNone(metric_data) + metrics = { + metric.name: metric + for rm in metric_data.resource_metrics + for sm in rm.scope_metrics + for metric in sm.metrics + } + + self.assertIn( + "gfe_connectivity_error_count", + metrics, + f"Metrics: {list(metrics.keys())}", + ) + missing_metric = metrics["gfe_connectivity_error_count"] + point = next(iter(missing_metric.data.data_points)) + self.assertGreaterEqual(point.value, 1) + + self.assertIn( + "afe_connectivity_error_count", + metrics, + f"Metrics: {list(metrics.keys())}", + ) + afe_missing_metric = metrics["afe_connectivity_error_count"] + afe_point = next(iter(afe_missing_metric.data.data_points)) + self.assertGreaterEqual(afe_point.value, 1) + finally: + pass 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 6e091860b425..2d4c1bbcbe20 100644 --- a/packages/google-cloud-spanner/tests/unit/test_metrics_interceptor.py +++ b/packages/google-cloud-spanner/tests/unit/test_metrics_interceptor.py @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -from unittest.mock import MagicMock +from unittest.mock import MagicMock, Mock import pytest @@ -41,11 +41,11 @@ def __init__(self): self.project = None self.instance = None self.database = None - self.gfe_enabled = False + self.gfe_enabled = True self.record_attempt_start = MagicMock() self.record_attempt_completion = MagicMock() self.set_method = MagicMock() - self.record_gfe_metrics = MagicMock() + self.record_front_end_metrics = MagicMock() self.set_project = MagicMock() self.set_instance = MagicMock() self.set_database = MagicMock() @@ -99,10 +99,8 @@ def test_set_metrics_tracer_attributes(interceptor, mock_tracer_ctx): def test_intercept_with_tracer(interceptor, mock_tracer_ctx): # mock_tracer_ctx fixture sets the ContextVar - mock_tracer_ctx.gfe_enabled = False - - invoked_response = MagicMock() - invoked_response.initial_metadata.return_value = {} + invoked_response = Mock() + invoked_response.initial_metadata.return_value = [] mock_invoked_method = MagicMock(return_value=invoked_response) call_details = MagicMock( @@ -115,8 +113,10 @@ def test_intercept_with_tracer(interceptor, mock_tracer_ctx): ], ) + replaced_call_details = call_details._replace.return_value response = interceptor.intercept(mock_invoked_method, "request", call_details) assert response == invoked_response mock_tracer_ctx.record_attempt_start.assert_called() mock_tracer_ctx.record_attempt_completion.assert_called_once() - mock_invoked_method.assert_called_once_with("request", call_details) + mock_tracer_ctx.record_front_end_metrics.assert_called_once() + mock_invoked_method.assert_called_once_with("request", replaced_call_details) 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 90b2f2f511f9..a35284703d87 100644 --- a/packages/google-cloud-spanner/tests/unit/test_metrics_tracer.py +++ b/packages/google-cloud-spanner/tests/unit/test_metrics_tracer.py @@ -29,6 +29,10 @@ def metrics_tracer(): mock_attempt_latency = mock.create_autospec(Histogram, instance=True) mock_operation_counter = mock.create_autospec(Counter, instance=True) mock_operation_latency = mock.create_autospec(Histogram, instance=True) + mock_gfe_latency = mock.create_autospec(Histogram, instance=True) + mock_gfe_missing = mock.create_autospec(Counter, instance=True) + mock_afe_latency = mock.create_autospec(Histogram, instance=True) + mock_afe_missing = mock.create_autospec(Counter, instance=True) return MetricsTracer( enabled=True, instrument_attempt_latency=mock_attempt_latency, @@ -36,6 +40,10 @@ def metrics_tracer(): instrument_operation_latency=mock_operation_latency, instrument_operation_counter=mock_operation_counter, client_attributes={"project_id": "test_project"}, + instrument_gfe_latency=mock_gfe_latency, + instrument_gfe_connectivity_error_count=mock_gfe_missing, + instrument_afe_latency=mock_afe_latency, + instrument_afe_connectivity_error_count=mock_afe_missing, ) @@ -235,7 +243,7 @@ def test_record_gfe_latency(metrics_tracer): assert mock_gfe_latency.record.call_args[1]["amount"] == 100 assert ( mock_gfe_latency.record.call_args[1]["attributes"] - == metrics_tracer.client_attributes + == metrics_tracer._create_attempt_otel_attributes() ) # Test when tracing is disabled @@ -245,22 +253,120 @@ def test_record_gfe_latency(metrics_tracer): metrics_tracer.enabled = True # Reset for next test -def test_record_gfe_missing_header_count(metrics_tracer): - mock_gfe_missing_header_count = mock.create_autospec(Counter, instance=True) - metrics_tracer._instrument_gfe_missing_header_count = mock_gfe_missing_header_count +def test_record_gfe_connectivity_error_count(metrics_tracer): + mock_gfe_connectivity_error_count = mock.create_autospec(Counter, instance=True) + metrics_tracer._instrument_gfe_connectivity_error_count = ( + mock_gfe_connectivity_error_count + ) metrics_tracer.gfe_enabled = True # Ensure GFE is enabled # Test when tracing is enabled - metrics_tracer.record_gfe_missing_header_count() - assert mock_gfe_missing_header_count.add.call_count == 1 - assert mock_gfe_missing_header_count.add.call_args[1]["amount"] == 1 + metrics_tracer.record_gfe_connectivity_error_count() + assert mock_gfe_connectivity_error_count.add.call_count == 1 + assert mock_gfe_connectivity_error_count.add.call_args[1]["amount"] == 1 assert ( - mock_gfe_missing_header_count.add.call_args[1]["attributes"] - == metrics_tracer.client_attributes + mock_gfe_connectivity_error_count.add.call_args[1]["attributes"] + == metrics_tracer._create_attempt_otel_attributes() ) # Test when tracing is disabled metrics_tracer.enabled = False - metrics_tracer.record_gfe_missing_header_count() - assert mock_gfe_missing_header_count.add.call_count == 1 # Should not increment + metrics_tracer.record_gfe_connectivity_error_count() + assert mock_gfe_connectivity_error_count.add.call_count == 1 # Should not increment metrics_tracer.enabled = True # Reset for next test + + +def test_extract_front_end_latencies(): + # Valid trailing metadata list of tuples + metadata_list = [ + ("server-timing", "gfet4t7; dur=123"), + ("server-timing", "afe; dur=100"), + ] + assert MetricsTracer.extract_front_end_latencies(metadata_list) == (123, 100) + + # Valid metadata dict + metadata_dict = {"server-timing": "gfet4t7; dur=456"} + assert MetricsTracer.extract_front_end_latencies(metadata_dict) == (456, None) + + # Missing header + assert MetricsTracer.extract_front_end_latencies([("other-header", "val")]) == ( + None, + None, + ) + assert MetricsTracer.extract_front_end_latencies(None) == (None, None) + + +def test_record_front_end_metrics(metrics_tracer): + mock_gfe_latency = mock.create_autospec(Histogram, instance=True) + mock_gfe_missing = mock.create_autospec(Counter, instance=True) + mock_afe_latency = mock.create_autospec(Histogram, instance=True) + mock_afe_missing = mock.create_autospec(Counter, instance=True) + metrics_tracer._instrument_gfe_latency = mock_gfe_latency + metrics_tracer._instrument_gfe_connectivity_error_count = mock_gfe_missing + metrics_tracer._instrument_afe_latency = mock_afe_latency + metrics_tracer._instrument_afe_connectivity_error_count = mock_afe_missing + metrics_tracer.gfe_enabled = True + + # With header + 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_gfe_latency.record.call_args[1]["amount"] == 88 + assert mock_gfe_missing.add.call_count == 0 + assert mock_afe_latency.record.call_count == 1 + assert mock_afe_latency.record.call_args[1]["amount"] == 90 + assert mock_afe_missing.add.call_count == 0 + + # Without header + metrics_tracer.record_front_end_metrics([("other", "1")]) + assert mock_gfe_latency.record.call_count == 1 + assert mock_gfe_missing.add.call_count == 1 + assert mock_afe_latency.record.call_count == 1 + assert mock_afe_missing.add.call_count == 1 + + +def test_record_afe_latency(metrics_tracer): + mock_afe_latency = mock.create_autospec(Histogram, instance=True) + metrics_tracer._instrument_afe_latency = mock_afe_latency + metrics_tracer.gfe_enabled = True + + metrics_tracer.record_afe_latency(100) + assert mock_afe_latency.record.call_count == 1 + assert mock_afe_latency.record.call_args[1]["amount"] == 100 + assert ( + mock_afe_latency.record.call_args[1]["attributes"] + == metrics_tracer._create_attempt_otel_attributes() + ) + + with mock.patch.dict("os.environ", {"SPANNER_DISABLE_AFE_SERVER_TIMING": "true"}): + metrics_tracer.record_afe_latency(300) + assert mock_afe_latency.record.call_count == 1 + + metrics_tracer.enabled = False + metrics_tracer.record_afe_latency(200) + assert mock_afe_latency.record.call_count == 1 + metrics_tracer.enabled = True + + +def test_record_afe_connectivity_error_count(metrics_tracer): + mock_afe_missing = mock.create_autospec(Counter, instance=True) + metrics_tracer._instrument_afe_connectivity_error_count = mock_afe_missing + metrics_tracer.gfe_enabled = True + + metrics_tracer.record_afe_connectivity_error_count() + assert mock_afe_missing.add.call_count == 1 + assert mock_afe_missing.add.call_args[1]["amount"] == 1 + assert ( + mock_afe_missing.add.call_args[1]["attributes"] + == metrics_tracer._create_attempt_otel_attributes() + ) + + with mock.patch.dict("os.environ", {"SPANNER_DISABLE_AFE_SERVER_TIMING": "true"}): + metrics_tracer.record_afe_connectivity_error_count() + assert mock_afe_missing.add.call_count == 1 + + metrics_tracer.enabled = False + metrics_tracer.record_afe_connectivity_error_count() + assert mock_afe_missing.add.call_count == 1 + metrics_tracer.enabled = True