Feat: Added Metric Interceptor integration with Attempt metrics

Author: lszinv

google/cloud/spanner_v1/client.py

@@ -48,9 +48,27 @@
 from google.cloud.spanner_v1._helpers import _merge_query_options
 from google.cloud.spanner_v1._helpers import _metadata_with_prefix
 from google.cloud.spanner_v1.instance import Instance
+from google.cloud.spanner_v1.metrics.constants import ENABLE_SPANNER_METRICS_ENV_VAR
+from google.cloud.spanner_v1.metrics.spanner_metrics_tracer_factory import (
+    SpannerMetricsTracerFactory,
+)
+from google.cloud.spanner_v1.metrics.metrics_exporter import (
+    CloudMonitoringMetricsExporter,
+)
+
+try:
+    from opentelemetry import metrics
+    from opentelemetry.sdk.metrics import MeterProvider
+    from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+
+    HAS_GOOGLE_CLOUD_MONITORING_INSTALLED = True
+except ImportError:  # pragma: NO COVER
+    HAS_GOOGLE_CLOUD_MONITORING_INSTALLED = False
+
 
 _CLIENT_INFO = client_info.ClientInfo(client_library_version=__version__)
 EMULATOR_ENV_VAR = "SPANNER_EMULATOR_HOST"
+ENABLE_BUILTIN_METRICS_ENV_VAR = "SPANNER_ENABLE_BUILTIN_METRICS"
 _EMULATOR_HOST_HTTP_SCHEME = (
     "%s contains a http scheme. When used with a scheme it may cause gRPC's "
     "DNS resolver to endlessly attempt to resolve. %s is intended to be used "
@@ -73,6 +91,10 @@ def _get_spanner_optimizer_statistics_package():
     return os.getenv(OPTIMIZER_STATISITCS_PACKAGE_ENV_VAR, "")
 
 
+def _get_spanner_enable_builtin_metrics():
+    return os.getenv(ENABLE_SPANNER_METRICS_ENV_VAR) == "true"
+
+
 class Client(ClientWithProject):
     """Client for interacting with Cloud Spanner API.
 
@@ -196,6 +218,18 @@ def __init__(
         ):
             warnings.warn(_EMULATOR_HOST_HTTP_SCHEME)
 
+        # Check flag to enable Spanner builtin metrics
+        if _get_spanner_enable_builtin_metrics():
+            meter_provider = metrics.NoOpMeterProvider()
+            if not _get_spanner_emulator_host():
+                meter_provider = MeterProvider(
+                    metric_readers=[
+                        PeriodicExportingMetricReader(CloudMonitoringMetricsExporter())
+                    ]
+                )
+            metrics.set_meter_provider(meter_provider)
+            SpannerMetricsTracerFactory()
+
         self._route_to_leader_enabled = route_to_leader_enabled
         self._directed_read_options = directed_read_options
         self._observability_options = observability_options

google/cloud/spanner_v1/metrics/constants.py

@@ -15,6 +15,12 @@
 BUILT_IN_METRICS_METER_NAME = "gax-python"
 NATIVE_METRICS_PREFIX = "spanner.googleapis.com/internal/client"
 SPANNER_RESOURCE_TYPE = "spanner_instance_client"
+SPANNER_SERVICE_NAME = "spanner-python"
+GOOGLE_CLOUD_RESOURCE_KEY = "google-cloud-resource-prefix"
+GOOGLE_CLOUD_REGION_KEY = "cloud.region"
+GOOGLE_CLOUD_REGION_GLOBAL = "global"
+SPANNER_METHOD_PREFIX = "/google.spanner.v1."
+ENABLE_SPANNER_METRICS_ENV_VAR = "SPANNER_ENABLE_BUILTIN_METRICS"
 
 # Monitored resource labels
 MONITORED_RES_LABEL_KEY_PROJECT = "project_id"

google/cloud/spanner_v1/metrics/metrics_interceptor.py

@@ -0,0 +1,125 @@
+"""Interceptor for collecting Cloud Spanner metrics."""
+
+from grpc_interceptor import ClientInterceptor
+from .constants import (
+    GOOGLE_CLOUD_RESOURCE_KEY,
+    SPANNER_METHOD_PREFIX,
+)
+from .metrics_tracer import MetricsTracer
+from typing import Dict
+from .spanner_metrics_tracer_factory import SpannerMetricsTracerFactory
+import re
+
+
+class MetricsInterceptor(ClientInterceptor):
+    """Interceptor that collects metrics for Cloud Spanner operations."""
+
+    @staticmethod
+    def _parse_resource_path(path: str) -> dict:
+        """Parse the resource path to extract project, instance and database.
+
+        Args:
+            path (str): The resource path from the request
+
+        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<project>[^/]+)(/instances/(?P<instance>[^/]+))?(/databases/(?P<database>[^/]+))?(/sessions/(?P<session>[^/]+))?.*$"
+        match = re.match(pattern, path)
+        if match:
+            return {k: v for k, v in match.groupdict().items() if v is not None}
+        return {}
+
+    @staticmethod
+    def _extract_resource_from_path(metadata: Dict[str, str]) -> 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.
+
+        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), ""
+        )
+
+        resources = MetricsInterceptor._parse_resource_path(path)
+        return resources
+
+    @staticmethod
+    def _remove_prefix(s: str, prefix: str) -> str:
+        """
+        This function removes the prefix from the given string.
+
+        Args:
+            s (str): The string from which the prefix is to be removed.
+            prefix (str): The prefix to be removed from the string.
+
+        Returns:
+            str: The string with the prefix removed.
+
+        Note:
+            This function is used because the `removeprefix` method does not exist in Python 3.8.
+        """
+        if s.startswith(prefix):
+            return s[len(prefix) :]
+        return s
+
+    def _set_metrics_tracer_attributes(self, resources: Dict[str, str]) -> 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.
+
+        Args:
+            resources (Dict[str, str]): A dictionary containing project, instance, and database information.
+        """
+        if SpannerMetricsTracerFactory.current_metrics_tracer is None:
+            return
+
+        if resources:
+            if "project" in resources:
+                SpannerMetricsTracerFactory.current_metrics_tracer.set_project(resources["project"])
+            if "instance" in resources:
+                SpannerMetricsTracerFactory.current_metrics_tracer.set_instance(resources["instance"])
+            if "database" in resources:
+                SpannerMetricsTracerFactory.current_metrics_tracer.set_database(resources["database"])
+
+    def intercept(self, invoked_method, request_or_iterator, call_details):
+        """Intercept gRPC calls to collect metrics.
+
+        Args:
+            invoked_method: The RPC method
+            request_or_iterator: The RPC request
+            call_details: Details about the RPC call
+
+        Returns:
+            The RPC response
+        """
+        if SpannerMetricsTracerFactory.current_metrics_tracer is None:
+            return invoked_method(request_or_iterator, call_details)
+
+        # Setup Metric Tracer attributes from call details
+        ## Extract  Project / Instance / Databse from header information
+        resources = self._extract_resource_from_path(call_details.metadata)
+        self._set_metrics_tracer_attributes(resources)
+        ## Format method to be be spanner.<method name>
+        method_name = self._remove_prefix(
+            call_details.method, SPANNER_METHOD_PREFIX
+        ).replace("/", ".")
+        SpannerMetricsTracerFactory.current_metrics_tracer.set_method(method_name)
+
+        SpannerMetricsTracerFactory.current_metrics_tracer.record_attempt_start()
+        response = invoked_method(request_or_iterator, call_details)
+        SpannerMetricsTracerFactory.current_metrics_tracer.record_attempt_completion()
+
+        return response

google/cloud/spanner_v1/metrics/metrics_tracer.py

@@ -28,7 +28,6 @@
     METRIC_LABEL_KEY_CLIENT_UID,
     METRIC_LABEL_KEY_DATABASE,
     METRIC_LABEL_KEY_DIRECT_PATH_ENABLED,
-    METRIC_LABEL_KEY_DIRECT_PATH_USED,
     METRIC_LABEL_KEY_METHOD,
     METRIC_LABEL_KEY_STATUS,
     MONITORED_RES_LABEL_KEY_CLIENT_HASH,
@@ -399,7 +398,6 @@ def _create_operation_otel_attributes(self) -> dict:
         self._client_attributes[METRIC_LABEL_KEY_STATUS] = self.current_op.status
         return self._client_attributes
 
-
     def _create_attempt_otel_attributes(self) -> dict:
         """
         Create additional attributes for attempt metrics tracing.

google/cloud/spanner_v1/metrics/spanner_metrics_tracer_factory.py

@@ -0,0 +1,113 @@
+# -*- coding: utf-8 -*-
+# Copyright 2025 Google LLC
+#
+# 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.
+
+
+"""This module provides a singleton factory for creating SpannerMetricsTracer instances."""
+
+from .metrics_tracer_factory import MetricsTracerFactory
+import os
+from .constants import (
+    SPANNER_SERVICE_NAME,
+    GOOGLE_CLOUD_REGION_KEY,
+    GOOGLE_CLOUD_REGION_GLOBAL,
+)
+from opentelemetry.resourcedetector.gcp_resource_detector import (
+    GoogleCloudResourceDetector,
+)
+from .metrics_tracer import MetricsTracer
+from google.cloud.spanner_v1 import __version__
+import mmh3
+from uuid import uuid4
+
+
+class SpannerMetricsTracerFactory(MetricsTracerFactory):
+    """A factory for creating SpannerMetricsTracer instances."""
+
+    _metrics_tracer_factory: "SpannerMetricsTracerFactory" = None
+    current_metrics_tracer: MetricsTracer = None
+
+    def __new__(cls, enabled: bool = True) -> "SpannerMetricsTracerFactory":
+        """Create a new instance of SpannerMetricsTracerFactory if it doesn't already exist."""
+        if cls._metrics_tracer_factory is None:
+            cls._metrics_tracer_factory = MetricsTracerFactory(
+                enabled, SPANNER_SERVICE_NAME
+            )
+
+            client_uid = cls._generate_client_uid()
+            cls._metrics_tracer_factory.set_client_uid(client_uid)
+            cls._metrics_tracer_factory.set_instance_config(cls._get_instance_config())
+            cls._metrics_tracer_factory.set_client_name(cls._get_client_name)
+            cls._metrics_tracer_factory.set_client_hash(
+                cls._generate_client_hash(client_uid)
+            )
+            cls._metrics_tracer_factory.set_location(cls._get_location())
+        return cls._metrics_tracer_factory
+
+    @staticmethod
+    def _generate_client_uid() -> str:
+        """Generate a client UID in the form of uuidv4@pid@hostname."""
+        try:
+            hostname = os.uname()[1]
+            pid = str(os.getpid())[0:10]  # Limit PID to 10 characters
+            uuid = uuid4()
+            return f"{uuid}@{pid}@{hostname}"
+        except Exception:
+            return ""
+
+    @staticmethod
+    def _get_instance_config() -> str:
+        """Get the instance configuration."""
+        # TODO: unknown until there's a good way to get it.
+        return "unknown"
+
+    @staticmethod
+    def _get_client_name() -> str:
+        """Get the client name."""
+        return f"{SPANNER_SERVICE_NAME}/{__version__}"
+
+    @staticmethod
+    def _generate_client_hash(client_uid: str) -> str:
+        """
+        Generate a 6-digit zero-padded lowercase hexadecimal hash using the 10 most significant bits of a 64-bit hash value.
+
+            The primary purpose of this function is to generate a hash value for the `client_hash`
+            resource label using `client_uid` metric field. The range of values is chosen to be small
+            enough to keep the cardinality of the Resource targets under control. Note: If at later time
+            the range needs to be increased, it can be done by increasing the value of `kPrefixLength` to
+            up to 24 bits without changing the format of the returned value.
+        """
+        if not client_uid:
+            return "000000"
+        hashed_client = mmh3.hash64(client_uid)
+
+        # Join the hashes back together since mmh3 splits into high and low 32bits
+        full_hash = (hashed_client[0] << 32) | (hashed_client[1] & 0xFFFFFFFF)
+        unsigned_hash = full_hash & 0xFFFFFFFFFFFFFFFF
+
+        k_prefix_length = 10
+        sig_figs = unsigned_hash >> (64 - k_prefix_length)
+
+        # Return as 6 digit zero padded hex string
+        return f"{sig_figs:06x}"
+
+    @staticmethod
+    def _get_location() -> str:
+        """Get the location of the resource."""
+        detector = GoogleCloudResourceDetector()
+        resources = detector.detect()
+        if GOOGLE_CLOUD_REGION_KEY not in resources.attributes:
+            return GOOGLE_CLOUD_REGION_GLOBAL
+        else:
+            return resources[GOOGLE_CLOUD_REGION_KEY]

google/cloud/spanner_v1/services/spanner/client.py

@@ -62,6 +62,7 @@
 from .transports.grpc import SpannerGrpcTransport
 from .transports.grpc_asyncio import SpannerGrpcAsyncIOTransport
 from .transports.rest import SpannerRestTransport
+from google.cloud.spanner_v1.metrics.metrics_interceptor import MetricsInterceptor
 
 
 class SpannerClientMeta(type):
@@ -705,6 +706,7 @@ def __init__(
                 client_info=client_info,
                 always_use_jwt_access=True,
                 api_audience=self._client_options.api_audience,
+                metrics_interceptor=MetricsInterceptor(),
             )
 
     def create_session(

google/cloud/spanner_v1/services/spanner/transports/base.py

@@ -31,6 +31,7 @@
 from google.cloud.spanner_v1.types import spanner
 from google.cloud.spanner_v1.types import transaction
 from google.protobuf import empty_pb2  # type: ignore
+from google.cloud.spanner_v1.metrics.metrics_interceptor import MetricsInterceptor
 
 DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
     gapic_version=package_version.__version__
@@ -58,6 +59,7 @@ def __init__(
         client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
         always_use_jwt_access: Optional[bool] = False,
         api_audience: Optional[str] = None,
+        metrics_interceptor: Optional[MetricsInterceptor] = None,
         **kwargs,
     ) -> None:
         """Instantiate the transport.

google/cloud/spanner_v1/services/spanner/transports/grpc.py

@@ -28,6 +28,9 @@
 from google.cloud.spanner_v1.types import result_set
 from google.cloud.spanner_v1.types import spanner
 from google.cloud.spanner_v1.types import transaction
+
+from google.cloud.spanner_v1.metrics.metrics_interceptor import MetricsInterceptor
+from google.cloud.spanner_v1.metrics.constants import ENABLE_SPANNER_METRICS_ENV_VAR
 from google.protobuf import empty_pb2  # type: ignore
 from .base import SpannerTransport, DEFAULT_CLIENT_INFO
 
@@ -66,6 +69,7 @@ def __init__(
         client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
         always_use_jwt_access: Optional[bool] = False,
         api_audience: Optional[str] = None,
+        metrics_interceptor: Optional[MetricsInterceptor] = None,
     ) -> None:
         """Instantiate the transport.
 
@@ -187,6 +191,12 @@ def __init__(
                 ],
             )
 
+        # Wrap the gRPC channel with the metric interceptor
+        if metrics_interceptor is not None:
+            self._grpc_channel = grpc.intercept_channel(
+                self._grpc_channel, metrics_interceptor
+            )
+
         # Wrap messages. This must be done after self._grpc_channel exists
         self._prep_wrapped_messages(client_info)