fix(spanner): align GFE/AFE metric names and instruments with backend specs

Author: sinhasubham

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.

packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/constants.py

@@ -59,18 +59,18 @@
 METRIC_NAME_OPERATION_COUNT = "operation_count"
 METRIC_NAME_ATTEMPT_COUNT = "attempt_count"
 METRIC_NAME_GFE_LATENCY = "gfe_latencies"
-METRIC_NAME_GFE_MISSING_HEADER_COUNT = "gfe_missing_header_count"
+METRIC_NAME_GFE_CONNECTIVITY_ERROR_COUNT = "gfe_connectivity_error_count"
 METRIC_NAME_AFE_LATENCY = "afe_latencies"
-METRIC_NAME_AFE_MISSING_HEADER_COUNT = "afe_missing_header_count"
+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_MISSING_HEADER_COUNT,
+    METRIC_NAME_GFE_CONNECTIVITY_ERROR_COUNT,
     METRIC_NAME_AFE_LATENCY,
-    METRIC_NAME_AFE_MISSING_HEADER_COUNT,
+    METRIC_NAME_AFE_CONNECTIVITY_ERROR_COUNT,
 ]
 
 METRIC_EXPORT_INTERVAL_MS = 60000  # 1 Minute

packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_interceptor.py

@@ -52,22 +52,27 @@ 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
@@ -147,8 +152,7 @@ def _wrap_response(response: Any, tracer: Any) -> Any:
                     metadata.extend(response.initial_metadata() or [])
                 except Exception as e:
                     logger.warning(f"Failed to retrieve initial metadata: {e}")
-            tracer.record_gfe_metrics(metadata)
-            tracer.record_afe_metrics(metadata)
+            tracer.record_front_end_metrics(metadata)
         except Exception as e:
             logger.warning(f"Failed to record metrics: {e}")
         return response
@@ -256,8 +260,7 @@ def _record_metrics(self):
                     metadata.extend(self._response.initial_metadata() or [])
                 except Exception as e:
                     logger.warning(f"Failed to retrieve initial metadata: {e}")
-            self._tracer.record_gfe_metrics(metadata)
-            self._tracer.record_afe_metrics(metadata)
+            self._tracer.record_front_end_metrics(metadata)
         except Exception as e:
             logger.warning(f"Failed to record metrics: {e}")
 
@@ -333,8 +336,7 @@ async def _record_metrics(self):
                     metadata.extend(res or [])
                 except Exception as e:
                     logger.warning(f"Failed to retrieve initial metadata: {e}")
-            self._tracer.record_gfe_metrics(metadata)
-            self._tracer.record_afe_metrics(metadata)
+            self._tracer.record_front_end_metrics(metadata)
         except Exception as e:
             logger.warning(f"Failed to record metrics: {e}")
 
@@ -439,8 +441,7 @@ async def _record_metrics(self):
                     metadata.extend(res or [])
                 except Exception as e:
                     logger.warning(f"Failed to retrieve initial metadata: {e}")
-            self._tracer.record_gfe_metrics(metadata)
-            self._tracer.record_afe_metrics(metadata)
+            self._tracer.record_front_end_metrics(metadata)
         except Exception as e:
             logger.warning(f"Failed to record metrics: {e}")
 

packages/google-cloud-spanner/google/cloud/spanner_v1/metrics/metrics_tracer.py

@@ -184,9 +184,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_missing_header_count: "Counter"
+    _instrument_afe_connectivity_error_count: "Counter"
     current_op: MetricOpTracer
     enabled: bool
     gfe_enabled: bool
@@ -200,11 +200,11 @@ 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,
-        instrument_gfe_latency: Optional["Histogram"] = None,
-        instrument_gfe_missing_header_count: Optional["Counter"] = None,
-        instrument_afe_latency: Optional["Histogram"] = None,
-        instrument_afe_missing_header_count: Optional["Counter"] = None,
     ):
         """
         Initialize a MetricsTracer instance with the given parameters.
@@ -221,10 +221,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, optional): Instrument for measuring GFE latency.
-            instrument_gfe_missing_header_count (Counter, optional): Instrument for counting missing GFE headers.
-            instrument_afe_latency (Histogram, optional): Instrument for measuring AFE latency.
-            instrument_afe_missing_header_count (Counter, optional): Instrument for counting missing AFE headers.
+            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
@@ -233,9 +233,13 @@ def __init__(
         self._instrument_operation_latency = instrument_operation_latency
         self._instrument_operation_counter = instrument_operation_counter
         self._instrument_gfe_latency = instrument_gfe_latency
-        self._instrument_gfe_missing_header_count = instrument_gfe_missing_header_count
+        self._instrument_gfe_connectivity_error_count = (
+            instrument_gfe_connectivity_error_count
+        )
         self._instrument_afe_latency = instrument_afe_latency
-        self._instrument_afe_missing_header_count = instrument_afe_missing_header_count
+        self._instrument_afe_connectivity_error_count = (
+            instrument_afe_connectivity_error_count
+        )
         self.enabled = enabled
         self.gfe_enabled = True
 
@@ -424,17 +428,17 @@ def record_gfe_latency(self, latency: int) -> None:
             amount=latency, attributes=self.client_attributes
         )
 
-    def record_gfe_missing_header_count(self) -> None:
+    def record_gfe_connectivity_error_count(self) -> None:
         """
-        Increments the counter for missing GFE headers.
+        Increments the counter for GFE connectivity errors.
         """
         if (
             not self.enabled
             or not HAS_OPENTELEMETRY_INSTALLED
-            or not getattr(self, "_instrument_gfe_missing_header_count", None)
+            or not getattr(self, "_instrument_gfe_connectivity_error_count", None)
         ):
             return
-        self._instrument_gfe_missing_header_count.add(
+        self._instrument_gfe_connectivity_error_count.add(
             amount=1, attributes=self.client_attributes
         )
 
@@ -455,101 +459,52 @@ def record_afe_latency(self, latency: int) -> None:
             amount=latency, attributes=self.client_attributes
         )
 
-    def record_afe_missing_header_count(self) -> None:
+    def record_afe_connectivity_error_count(self) -> None:
         """
-        Increments the counter for missing AFE headers.
+        Increments the counter for AFE connectivity errors.
         """
         if (
             not self.enabled
             or not HAS_OPENTELEMETRY_INSTALLED
-            or not getattr(self, "_instrument_afe_missing_header_count", None)
+            or not getattr(self, "_instrument_afe_connectivity_error_count", None)
         ):
             return
-        self._instrument_afe_missing_header_count.add(
+        self._instrument_afe_connectivity_error_count.add(
             amount=1, attributes=self.client_attributes
         )
 
     @staticmethod
-    def extract_gfe_latency(metadata: Any) -> Optional[int]:
+    def extract_front_end_latencies(
+        metadata: Any,
+    ) -> tuple[Optional[int], Optional[int]]:
         """
-        Extracts the GFE latency value (in milliseconds) from response metadata.
+        Extracts both GFE and AFE latency values (in milliseconds) from response metadata.
         """
         if not metadata:
-            return None
+            return None, None
 
-        header_vals = []
         if isinstance(metadata, dict):
-            for key, val in metadata.items():
-                if key and str(key).lower() in ("server-timing", "server_timing"):
-                    if isinstance(val, (list, tuple)):
-                        header_vals.extend(val)
-                    else:
-                        header_vals.append(val)
+            items = metadata.items()
         elif isinstance(metadata, (list, tuple)):
-            for item in metadata:
-                if isinstance(item, (list, tuple)) and len(item) == 2:
-                    key, val = item
-                    if key and str(key).lower() in ("server-timing", "server_timing"):
-                        if isinstance(val, (list, tuple)):
-                            header_vals.extend(val)
-                        else:
-                            header_vals.append(val)
-
-        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)
-            match = re.search(r"gfet4t7;\s*dur=([0-9.]+)", header_val)
-            if match:
-                try:
-                    return int(float(match.group(1)))
-                except ValueError:
-                    pass
-        return None
-
-    def record_gfe_metrics(self, metadata: Any) -> None:
-        """
-        Extracts and records GFE metrics from the RPC response metadata.
-        """
-        if not self.enabled or not HAS_OPENTELEMETRY_INSTALLED:
-            return
-        latency = self.extract_gfe_latency(metadata)
-        if latency is not None:
-            self.record_gfe_latency(latency)
+            items = [
+                item
+                for item in metadata
+                if isinstance(item, (list, tuple)) and len(item) == 2
+            ]
         else:
-            self.record_gfe_missing_header_count()
-
-    @staticmethod
-    def extract_afe_latency(metadata: Any) -> Optional[int]:
-        """
-        Extracts the AFE latency value (in milliseconds) from response metadata.
-        """
-        if not metadata:
-            return None
+            items = []
 
         header_vals = []
-        if isinstance(metadata, dict):
-            for key, val in metadata.items():
-                if key and str(key).lower() in ("server-timing", "server_timing"):
-                    if isinstance(val, (list, tuple)):
-                        header_vals.extend(val)
-                    else:
-                        header_vals.append(val)
-        elif isinstance(metadata, (list, tuple)):
-            for item in metadata:
-                if isinstance(item, (list, tuple)) and len(item) == 2:
-                    key, val = item
-                    if key and str(key).lower() in ("server-timing", "server_timing"):
-                        if isinstance(val, (list, tuple)):
-                            header_vals.extend(val)
-                        else:
-                            header_vals.append(val)
+        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:
@@ -561,25 +516,42 @@ def extract_afe_latency(metadata: Any) -> Optional[int]:
                     header_val = str(header_val)
             elif not isinstance(header_val, str):
                 header_val = str(header_val)
-            match = re.search(r"afe(?:t4t7)?;\s*dur=([0-9.]+)", header_val)
-            if match:
-                try:
-                    return int(float(match.group(1)))
-                except ValueError:
-                    pass
-        return None
 
-    def record_afe_metrics(self, metadata: Any) -> None:
+            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 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
-        latency = self.extract_afe_latency(metadata)
-        if latency is not None:
-            self.record_afe_latency(latency)
+        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_missing_header_count()
+            self.record_afe_connectivity_error_count()
 
     def _create_operation_otel_attributes(self) -> dict:
         """