feat(storage): add OpenTelemetry metrics gating and configuration

packages/google-cloud-storage/google/cloud/storage/_opentelemetry_metrics.py

@@ -0,0 +1,140 @@
+# Copyright 2026 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.
+
+"""Manages OpenTelemetry metrics instruments and gating for GCS client."""
+
+import logging
+import os
+from typing import Any, Dict, Optional
+
+from google.cloud.storage.version import __version__
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# 1. Hidden Development Gate
+# ---------------------------------------------------------------------------
+# Must remain False in production branches.
+# Only enabled during active development and test runs.
+_ENABLE_METRICS_DEV_GATE = False
+
+# ---------------------------------------------------------------------------
+# 2. Standardized Configuration and Environment Variable Names
+# ---------------------------------------------------------------------------
+ENABLE_OTEL_METRICS_ENV_VAR = "GCP_STORAGE_PYTHON_ENABLE_OTEL_METRICS"
+ENABLE_DEBUG_METRICS_ENV_VAR = "GCP_STORAGE_PYTHON_ENABLE_DEBUG_METRICS"
+
+_DEFAULT_ENABLE_METRICS = False
+_DEFAULT_ENABLE_DEBUG_METRICS = False
+
+# ---------------------------------------------------------------------------
+# 3. Optional OpenTelemetry Dependency Check
+# ---------------------------------------------------------------------------
+try:
+    from opentelemetry import metrics
+
+    HAS_OPENTELEMETRY_METRICS = True
+except ImportError:
+    HAS_OPENTELEMETRY_METRICS = False
+    logger.debug(
+        "OpenTelemetry metrics package (opentelemetry-api >= 1.12.0) is not "
+        "installed. GCS client metrics are disabled."
+    )
+
+
+def _parse_bool_env(name: str, default: bool = False) -> bool:
+    """Parses a boolean from an environment variable."""
+    val = os.environ.get(name)
+    if val is None:
+        return default
+    return str(val).strip().lower() in {"1", "true", "yes", "on"}
+
+
+def is_metrics_enabled(client_setting: Optional[bool] = None) -> bool:
+    """Evaluates whether standard GCS metrics should be recorded.
+
+    Args:
+        client_setting: Optional boolean configured on the client instance.
+            Takes precedence over the environment variable if specified.
+
+    Returns:
+        bool: True if metrics recording is enabled, False otherwise.
+    """
+    if client_setting is not None and not isinstance(client_setting, bool):
+        raise TypeError("enable_metrics must be a boolean or None.")
+
+    if not HAS_OPENTELEMETRY_METRICS:
+        return False
+
+    if not _ENABLE_METRICS_DEV_GATE:
+        return False
+
+    if client_setting is not None:
+        return client_setting
+
+    return _parse_bool_env(ENABLE_OTEL_METRICS_ENV_VAR, _DEFAULT_ENABLE_METRICS)
+
+
+def is_advanced_metrics_enabled(
+    client_setting: Optional[bool] = None,
+    base_setting: Optional[bool] = None,
+) -> bool:
+    """Evaluates whether high-frequency debug metrics should be recorded.
+
+    Args:
+        client_setting: Optional boolean configured on the client instance.
+            Takes precedence over the environment variable if specified.
+        base_setting: Optional boolean configured on the client instance for
+            base metrics.
+
+    Returns:
+        bool: True if advanced metrics recording is enabled, False otherwise.
+    """
+    if client_setting is not None and not isinstance(client_setting, bool):
+        raise TypeError("enable_advanced_metrics must be a boolean or None.")
+
+    if not is_metrics_enabled(base_setting):
+        return False
+
+    if client_setting is not None:
+        return client_setting
+
+    return _parse_bool_env(ENABLE_DEBUG_METRICS_ENV_VAR, _DEFAULT_ENABLE_DEBUG_METRICS)
+
+
+# ---------------------------------------------------------------------------
+# 4. Standard Common Attributes & Meter Provider
+# ---------------------------------------------------------------------------
+_COMMON_ATTRIBUTES: Dict[str, Any] = {
+    "gcp.client.service": "storage",
+    "gcp.client.version": __version__,
+    "gcp.client.repo": "googleapis/google-cloud-python",
+    "gcp.client.artifact": "google-cloud-storage",
+}
+
+
+def get_common_attributes() -> Dict[str, Any]:
+    """Returns a copy of standard GCS client attributes for metrics."""
+    return _COMMON_ATTRIBUTES.copy()
+
+
+def get_meter(meter_provider: Optional[Any] = None) -> Optional[Any]:
+    """Returns the OpenTelemetry Meter for Google Cloud Storage."""
+    if not HAS_OPENTELEMETRY_METRICS or not _ENABLE_METRICS_DEV_GATE:
+        return None
+    return metrics.get_meter(
+        "google.cloud.storage",
+        __version__,
+        meter_provider=meter_provider,
+    )

packages/google-cloud-storage/google/cloud/storage/asyncio/async_grpc_client.py

@@ -21,7 +21,7 @@
 from google.cloud._storage_v2.services.storage.transports.base import (
     DEFAULT_CLIENT_INFO,
 )
-from google.cloud.storage import __version__
+from google.cloud.storage import __version__, _opentelemetry_metrics
 
 _DEFAULT_HOST = "storage.googleapis.com"
 
@@ -63,6 +63,19 @@ class AsyncGrpcClient:
     :param attempt_direct_path:
         (Optional) Whether to attempt to use DirectPath for gRPC connections.
         Defaults to ``True``.
+
+    :type enable_metrics: bool or None
+    :param enable_metrics:
+        (Optional, Experimental) Whether to enable OpenTelemetry metrics. If
+        None, falls back to the GCP_STORAGE_PYTHON_ENABLE_OTEL_METRICS
+        environment variable, or False if unset.
+
+    :type enable_advanced_metrics: bool or None
+    :param enable_advanced_metrics:
+        (Optional, Experimental) Whether to enable advanced/debug OpenTelemetry
+        metrics. If None, falls back to the
+        GCP_STORAGE_PYTHON_ENABLE_DEBUG_METRICS environment variable, or False
+        if unset. Requires base OpenTelemetry metrics to be enabled.
     """
 
     def __init__(
@@ -72,7 +85,12 @@ def __init__(
         client_options=None,
         *,
         attempt_direct_path=True,
+        enable_metrics=None,
+        enable_advanced_metrics=None,
     ):
+        self._enable_metrics = enable_metrics
+        self._enable_advanced_metrics = enable_advanced_metrics
+
         if isinstance(credentials, auth_credentials.AnonymousCredentials):
             if client_options is None or client_options.api_endpoint is None:
                 raise ValueError(
@@ -99,6 +117,19 @@ def __init__(
             attempt_direct_path=attempt_direct_path,
         )
 
+    @property
+    def metrics_enabled(self) -> bool:
+        """Returns True if metrics recording is active for this client."""
+        return _opentelemetry_metrics.is_metrics_enabled(self._enable_metrics)
+
+    @property
+    def advanced_metrics_enabled(self) -> bool:
+        """Returns True if advanced metrics recording is active for this client."""
+        return _opentelemetry_metrics.is_advanced_metrics_enabled(
+            self._enable_advanced_metrics,
+            self._enable_metrics,
+        )
+
     def _create_anonymous_client(self, client_options, credentials):
         channel = grpc.aio.insecure_channel(client_options.api_endpoint)
         transport = storage_v2.services.storage.transports.StorageGrpcAsyncIOTransport(

packages/google-cloud-storage/google/cloud/storage/client.py

@@ -32,6 +32,7 @@
 from google.cloud.client import ClientWithProject
 from google.cloud.exceptions import NotFound
 
+from google.cloud.storage import _opentelemetry_metrics
 from google.cloud.storage._bucket_metadata_cache import BucketMetadataCache
 from google.cloud.storage._helpers import (
     _DEFAULT_SCHEME,
@@ -126,6 +127,19 @@ class Client(ClientWithProject):
         (Optional) An API key. Mutually exclusive with any other credentials.
         This parameter is an alias for setting `client_options.api_key` and
         will supercede any api key set in the `client_options` parameter.
+
+    :type enable_metrics: bool or None
+    :param enable_metrics:
+        (Optional, Experimental) Whether to enable OpenTelemetry metrics. If
+        None, falls back to the GCP_STORAGE_PYTHON_ENABLE_OTEL_METRICS
+        environment variable, or False if unset.
+
+    :type enable_advanced_metrics: bool or None
+    :param enable_advanced_metrics:
+        (Optional, Experimental) Whether to enable advanced/debug OpenTelemetry
+        metrics. If None, falls back to the
+        GCP_STORAGE_PYTHON_ENABLE_DEBUG_METRICS environment variable, or False
+        if unset. Requires base OpenTelemetry metrics to be enabled.
     """
 
     SCOPE = (
@@ -146,6 +160,8 @@ def __init__(
         extra_headers={},
         *,
         api_key=None,
+        enable_metrics=None,
+        enable_advanced_metrics=None,
     ):
         self._base_connection = None
 
@@ -293,6 +309,21 @@ def __init__(
         self._connection = connection
         self._batch_stack = _LocalStack()
         self._bucket_metadata_cache = BucketMetadataCache(self)
+        self._enable_metrics = enable_metrics
+        self._enable_advanced_metrics = enable_advanced_metrics
+
+    @property
+    def metrics_enabled(self) -> bool:
+        """Returns True if metrics recording is active for this client."""
+        return _opentelemetry_metrics.is_metrics_enabled(self._enable_metrics)
+
+    @property
+    def advanced_metrics_enabled(self) -> bool:
+        """Returns True if advanced metrics recording is active for this client."""
+        return _opentelemetry_metrics.is_advanced_metrics_enabled(
+            self._enable_advanced_metrics,
+            self._enable_metrics,
+        )
 
     def close(self):
         """Close the client and clear any cached metadata or active connections."""

packages/google-cloud-storage/google/cloud/storage/grpc_client.py

@@ -17,6 +17,7 @@
 from google.cloud.client import ClientWithProject
 
 from google.cloud import _storage_v2 as storage_v2
+from google.cloud.storage import _opentelemetry_metrics
 
 _marker = object()
 
@@ -58,6 +59,19 @@ class GrpcClient(ClientWithProject):
         This provides a direct, unproxied connection to GCS for lower latency
         and higher throughput, and is highly recommended when running on Google
         Cloud infrastructure. Defaults to ``True``.
+
+    :type enable_metrics: bool or None
+    :param enable_metrics:
+        (Optional, Experimental) Whether to enable OpenTelemetry metrics. If
+        None, falls back to the GCP_STORAGE_PYTHON_ENABLE_OTEL_METRICS
+        environment variable, or False if unset.
+
+    :type enable_advanced_metrics: bool or None
+    :param enable_advanced_metrics:
+        (Optional, Experimental) Whether to enable advanced/debug OpenTelemetry
+        metrics. If None, falls back to the
+        GCP_STORAGE_PYTHON_ENABLE_DEBUG_METRICS environment variable, or False
+        if unset. Requires base OpenTelemetry metrics to be enabled.
     """
 
     def __init__(
@@ -69,9 +83,14 @@ def __init__(
         *,
         api_key=None,
         attempt_direct_path=True,
+        enable_metrics=None,
+        enable_advanced_metrics=None,
     ):
         super(GrpcClient, self).__init__(project=project, credentials=credentials)
 
+        self._enable_metrics = enable_metrics
+        self._enable_advanced_metrics = enable_advanced_metrics
+
         if isinstance(client_options, dict):
             if api_key:
                 client_options["api_key"] = api_key
@@ -87,6 +106,19 @@ def __init__(
             attempt_direct_path=attempt_direct_path,
         )
 
+    @property
+    def metrics_enabled(self) -> bool:
+        """Returns True if metrics recording is active for this client."""
+        return _opentelemetry_metrics.is_metrics_enabled(self._enable_metrics)
+
+    @property
+    def advanced_metrics_enabled(self) -> bool:
+        """Returns True if advanced metrics recording is active for this client."""
+        return _opentelemetry_metrics.is_advanced_metrics_enabled(
+            self._enable_advanced_metrics,
+            self._enable_metrics,
+        )
+
     def _create_gapic_client(
         self,
         credentials=None,

packages/google-cloud-storage/google/cloud/storage/transfer_manager.py

@@ -1383,6 +1383,8 @@ def _reduce_client(cl):
     client_info = cl._initial_client_info
     client_options = cl._initial_client_options
     extra_headers = getattr(cl, "_extra_headers", {})
+    enable_metrics = getattr(cl, "_enable_metrics", None)
+    enable_advanced_metrics = getattr(cl, "_enable_advanced_metrics", None)
 
     return _LazyClient, (
         client_object_id,
@@ -1392,6 +1394,8 @@ def _reduce_client(cl):
         client_info,
         client_options,
         extra_headers,
+        enable_metrics,
+        enable_advanced_metrics,
     )
 
 
@@ -1461,6 +1465,10 @@ def __new__(cls, id, *args, **kwargs):
         if cached_client:
             return cached_client
         else:
+            if len(args) >= 8:
+                kwargs.setdefault("enable_metrics", args[6])
+                kwargs.setdefault("enable_advanced_metrics", args[7])
+                args = args[:6]
             cached_client = Client(*args, **kwargs)
             _cached_clients[id] = cached_client
             return cached_client

packages/google-cloud-storage/tests/unit/asyncio/test_async_grpc_client.py

@@ -149,6 +149,36 @@ def test_constructor_disables_directpath(self, mock_async_storage_client):
         mock_channel = mock_transport_cls.create_channel.return_value
         mock_transport_cls.assert_called_once_with(channel=mock_channel)
 
+    @mock.patch("google.cloud._storage_v2.StorageAsyncClient")
+    def test_metrics_properties(self, mock_async_storage_client):
+        from google.cloud.storage import _opentelemetry_metrics
+
+        mock_transport_cls = mock.MagicMock()
+        mock_async_storage_client.get_transport_class.return_value = mock_transport_cls
+        mock_creds = _make_credentials()
+
+        client = async_grpc_client.AsyncGrpcClient(
+            credentials=mock_creds,
+            enable_metrics=True,
+            enable_advanced_metrics=True,
+        )
+
+        with mock.patch.multiple(
+            _opentelemetry_metrics,
+            HAS_OPENTELEMETRY_METRICS=True,
+            _ENABLE_METRICS_DEV_GATE=True,
+        ):
+            assert client.metrics_enabled is True
+            assert client.advanced_metrics_enabled is True
+
+        with mock.patch.multiple(
+            _opentelemetry_metrics,
+            HAS_OPENTELEMETRY_METRICS=True,
+            _ENABLE_METRICS_DEV_GATE=False,
+        ):
+            assert client.metrics_enabled is False
+            assert client.advanced_metrics_enabled is False
+
     @mock.patch("google.cloud._storage_v2.StorageAsyncClient")
     def test_grpc_client_property(self, mock_grpc_gapic_client):
         # Arrange