Feat: Added Operation and GFE Metrics

Author: lszinv

google/cloud/spanner_v1/_opentelemetry_tracing.py

@@ -33,6 +33,8 @@
 except ImportError:
     HAS_OPENTELEMETRY_INSTALLED = False
 
+from google.cloud.spanner_v1.metrics.metrics_capture import MetricsCapture
+
 TRACER_NAME = "cloud.google.com/python/spanner"
 TRACER_VERSION = gapic_version.__version__
 extended_tracing_globally_disabled = (
@@ -107,26 +109,27 @@ def trace_call(name, session=None, extra_attributes=None, observability_options=
     with tracer.start_as_current_span(
         name, kind=trace.SpanKind.CLIENT, attributes=attributes
     ) as span:
-        try:
-            yield span
-        except Exception as error:
-            span.set_status(Status(StatusCode.ERROR, str(error)))
-            # OpenTelemetry-Python imposes invoking span.record_exception on __exit__
-            # on any exception. We should file a bug later on with them to only
-            # invoke .record_exception if not already invoked, hence we should not
-            # invoke .record_exception on our own else we shall have 2 exceptions.
-            raise
-        else:
-            # All spans still have set_status available even if for example
-            # NonRecordingSpan doesn't have "_status".
-            absent_span_status = getattr(span, "_status", None) is None
-            if absent_span_status or span._status.status_code == StatusCode.UNSET:
-                # OpenTelemetry-Python only allows a status change
-                # if the current code is UNSET or ERROR. At the end
-                # of the generator's consumption, only set it to OK
-                # it wasn't previously set otherwise.
-                # https://github.com/googleapis/python-spanner/issues/1246
-                span.set_status(Status(StatusCode.OK))
+        with MetricsCapture():
+            try:
+                yield span
+            except Exception as error:
+                span.set_status(Status(StatusCode.ERROR, str(error)))
+                # OpenTelemetry-Python imposes invoking span.record_exception on __exit__
+                # on any exception. We should file a bug later on with them to only
+                # invoke .record_exception if not already invoked, hence we should not
+                # invoke .record_exception on our own else we shall have 2 exceptions.
+                raise
+            else:
+                # All spans still have set_status available even if for example
+                # NonRecordingSpan doesn't have "_status".
+                absent_span_status = getattr(span, "_status", None) is None
+                if absent_span_status or span._status.status_code == StatusCode.UNSET:
+                    # OpenTelemetry-Python only allows a status change
+                    # if the current code is UNSET or ERROR. At the end
+                    # of the generator's consumption, only set it to OK
+                    # it wasn't previously set otherwise.
+                    # https://github.com/googleapis/python-spanner/issues/1246
+                    span.set_status(Status(StatusCode.OK))
 
 
 def get_current_span():

google/cloud/spanner_v1/metrics/constants.py

@@ -1,4 +1,5 @@
-# Copyright 2025 Google LLC
+# -*- 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.

google/cloud/spanner_v1/metrics/metrics_capture.py

@@ -0,0 +1,67 @@
+# -*- coding: utf-8 -*-
+# 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.
+"""
+This module provides functionality for capturing metrics in Cloud Spanner operations.
+
+It includes a context manager class, MetricsCapture, which automatically handles the
+start and completion of metrics tracing for a given operation. This ensures that metrics
+are consistently recorded for Cloud Spanner operations, facilitating observability and
+performance monitoring.
+"""
+
+from .spanner_metrics_tracer_factory import SpannerMetricsTracerFactory
+
+
+class MetricsCapture:
+    """Context manager for capturing metrics in Cloud Spanner operations.
+
+    This class provides a context manager interface to automatically handle
+    the start and completion of metrics tracing for a given operation.
+    """
+
+    def __enter__(self):
+        """Enter the runtime context related to this object.
+
+        This method initializes a new metrics tracer for the operation and
+        records the start of the operation.
+
+        Returns:
+            MetricsCapture: The instance of the context manager.
+        """
+        factory = SpannerMetricsTracerFactory()
+
+        # Define a new metrics tracer for the new operation
+        SpannerMetricsTracerFactory.current_metrics_tracer = (
+            factory.create_metrics_tracer()
+        )
+        SpannerMetricsTracerFactory.current_metrics_tracer.record_operation_start()
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        """Exit the runtime context related to this object.
+
+        This method records the completion of the operation. If an exception
+        occurred, it will be propagated after the metrics are recorded.
+
+        Args:
+            exc_type (Type[BaseException]): The exception type.
+            exc_value (BaseException): The exception value.
+            traceback (TracebackType): The traceback object.
+
+        Returns:
+            bool: False to propagate the exception if any occurred.
+        """
+        SpannerMetricsTracerFactory.current_metrics_tracer.record_operation_completion()
+        return False  # Propagate the exception if any

google/cloud/spanner_v1/metrics/metrics_exporter.py

@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
 # Copyright 2025 Google LLC
 #
 # Licensed under the Apache License, Version 2.0 (the "License");

google/cloud/spanner_v1/metrics/metrics_interceptor.py

@@ -20,6 +20,7 @@
     GOOGLE_CLOUD_RESOURCE_KEY,
     SPANNER_METHOD_PREFIX,
 )
+
 from typing import Dict
 from .spanner_metrics_tracer_factory import SpannerMetricsTracerFactory
 import re
@@ -143,4 +144,11 @@ def intercept(self, invoked_method, request_or_iterator, call_details):
         response = invoked_method(request_or_iterator, call_details)
         SpannerMetricsTracerFactory.current_metrics_tracer.record_attempt_completion()
 
+        # Process and send GFE metrics if enabled
+        if SpannerMetricsTracerFactory.current_metrics_tracer.gfe_enabled:
+            metadata = response.initial_metadata()
+            SpannerMetricsTracerFactory.current_metrics_trace.record_gfe_metrics(
+                metadata
+            )
+
         return response

google/cloud/spanner_v1/metrics/metrics_tracer.py

@@ -56,7 +56,7 @@ class MetricAttemptTracer:
     direct_path_used: bool
     status: str
 
-    def __init__(self):
+    def __init__(self) -> None:
         """
         Initialize a MetricAttemptTracer instance with default values.
 
@@ -181,8 +181,11 @@ class should not have any knowledge about the observability framework used for m
     _instrument_attempt_latency: "Histogram"
     _instrument_operation_counter: "Counter"
     _instrument_operation_latency: "Histogram"
+    _instrument_gfe_latency: "Histogram"
+    _instrument_gfe_missing_header_count: "Counter"
     current_op: MetricOpTracer
     enabled: bool
+    gfe_enabled: bool
     method: str
 
     def __init__(
@@ -193,21 +196,23 @@ def __init__(
         instrument_operation_latency: "Histogram",
         instrument_operation_counter: "Counter",
         client_attributes: Dict[str, str],
+        gfe_enabled: bool = False,
     ):
         """
         Initialize a MetricsTracer instance with the given parameters.
 
-        This constructor initializes a MetricsTracer instance with the provided method name, enabled status, direct path enabled status,
-        instrumented metrics for attempt latency, attempt counter, operation latency, operation counter, and client attributes.
-        It sets up the necessary metrics tracing infrastructure for recording metrics related to RPC operations.
+        This constructor sets up a MetricsTracer instance with the specified parameters, including the enabled status,
+        instruments for measuring and counting attempt and operation metrics, and client attributes. It prepares the
+        infrastructure needed for recording metrics related to RPC operations.
 
         Args:
-            enabled (bool): A flag indicating if metrics tracing is enabled.
-            instrument_attempt_latency (Histogram): The instrument for measuring attempt latency.
-            instrument_attempt_counter (Counter): The instrument for counting attempts.
-            instrument_operation_latency (Histogram): The instrument for measuring operation latency.
-            instrument_operation_counter (Counter): The instrument for counting operations.
-            client_attributes (dict[str, str]): A dictionary of client attributes used for metrics tracing.
+            enabled (bool): Indicates if metrics tracing is enabled.
+            instrument_attempt_latency (Histogram): Instrument for measuring attempt latency.
+            instrument_attempt_counter (Counter): Instrument for counting attempts.
+            instrument_operation_latency (Histogram): Instrument for measuring operation latency.
+            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.
         """
         self.current_op = MetricOpTracer()
         self._client_attributes = client_attributes
@@ -216,6 +221,7 @@ def __init__(
         self._instrument_operation_latency = instrument_operation_latency
         self._instrument_operation_counter = instrument_operation_counter
         self.enabled = enabled
+        self.gfe_enabled = gfe_enabled
 
     @staticmethod
     def _get_ms_time_diff(start: datetime, end: datetime) -> float:
@@ -322,7 +328,7 @@ def record_attempt_completion(self, status: str = StatusCode.OK.name) -> None:
 
         If metrics tracing is not enabled, this method does not perform any operations.
         """
-        if not self.enabled:
+        if not self.enabled or not HAS_OPENTELEMETRY_INSTALLED:
             return
         self.current_op.current_attempt.status = status
 
@@ -347,7 +353,7 @@ def record_operation_start(self) -> None:
         It is used to track the start time of an operation, which is essential for calculating operation latency and other metrics.
         If metrics tracing is not enabled, this method does not perform any operations.
         """
-        if not self.enabled:
+        if not self.enabled or not HAS_OPENTELEMETRY_INSTALLED:
             return
         self.current_op.start()
 
@@ -360,7 +366,7 @@ def record_operation_completion(self) -> None:
         Additionally, it increments the operation count and records the attempt count for the operation.
         If metrics tracing is not enabled, this method does not perform any operations.
         """
-        if not self.enabled:
+        if not self.enabled or not HAS_OPENTELEMETRY_INSTALLED:
             return
         end_time = datetime.now()
         # Build Attributes
@@ -385,14 +391,37 @@ def record_operation_completion(self) -> None:
             self.current_op.attempt_count, attributes=attempt_attributes
         )
 
+    def record_gfe_latency(self, latency: int) -> None:
+        """
+        Records the GFE 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 self.gfe_enabled:
+            return
+        self._instrument_gfe_latency.record(
+            amount=latency, attributes=self.client_attributes
+        )
+
+    def record_gfe_missing_header_count(self) -> None:
+        """
+        Increments the counter for missing GFE headers.
+        """
+        if not self.enabled or not HAS_OPENTELEMETRY_INSTALLED or not self.gfe_enabled:
+            return
+        self._instrument_gfe_missing_header_count.add(
+            amount=1, attributes=self.client_attributes
+        )
+
     def _create_operation_otel_attributes(self) -> dict:
         """
         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.
         """
-        if not self.enabled:
+        if not self.enabled or not HAS_OPENTELEMETRY_INSTALLED:
             return {}
         attributes = self._client_attributes.copy()
         attributes[METRIC_LABEL_KEY_STATUS] = self.current_op.status
@@ -405,7 +434,7 @@ def _create_attempt_otel_attributes(self) -> dict:
         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.
         """
-        if not self.enabled:
+        if not self.enabled or not HAS_OPENTELEMETRY_INSTALLED:
             return {}
 
         attributes = self._client_attributes.copy()

google/cloud/spanner_v1/metrics/metrics_tracer_factory.py

@@ -32,6 +32,8 @@
     METRIC_LABEL_KEY_DATABASE,
     METRIC_LABEL_KEY_DIRECT_PATH_ENABLED,
     BUILT_IN_METRICS_METER_NAME,
+    METRIC_NAME_GFE_LATENCY,
+    METRIC_NAME_GFE_MISSING_HEADER_COUNT,
 )
 
 from typing import Dict
@@ -50,10 +52,13 @@ class MetricsTracerFactory:
     """Factory class for creating MetricTracer instances. This class facilitates the creation of MetricTracer objects, which are responsible for collecting and tracing metrics."""
 
     enabled: bool
+    gfe_enabled: bool
     _instrument_attempt_latency: "Histogram"
     _instrument_attempt_counter: "Counter"
     _instrument_operation_latency: "Histogram"
     _instrument_operation_counter: "Counter"
+    _instrument_gfe_latency: "Histogram"
+    _instrument_gfe_missing_header_count: "Counter"
     _client_attributes: Dict[str, str]
 
     @property
@@ -307,3 +312,15 @@ def _create_metric_instruments(self, service_name: str) -> None:
             unit="1",
             description="Number of operations.",
         )
+
+        self._instrument_gfe_latency = meter.create_histogram(
+            name=METRIC_NAME_GFE_LATENCY,
+            unit="ms",
+            description="GFE Latency.",
+        )
+
+        self._instrument_gfe_missing_header_count = meter.create_counter(
+            name=METRIC_NAME_GFE_MISSING_HEADER_COUNT,
+            unit="1",
+            description="GFE missing header count.",
+        )