feat(storage): integrate OpenTelemetry metrics gating across clients and transfer_manager

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_debug_metrics: bool or None
+    :param enable_debug_metrics:
+        (Optional, Experimental) Whether to enable debug OpenTelemetry
+        metrics. If ``None``, falls back to the
+        ``GCP_STORAGE_PYTHON_ENABLE_OTEL_DEBUG_METRICS`` environment variable,
+        or ``False`` if unset.
     """
 
     def __init__(
@@ -72,7 +85,12 @@ def __init__(
         client_options=None,
         *,
         attempt_direct_path=True,
+        enable_metrics=None,
+        enable_debug_metrics=None,
     ):
+        self._enable_metrics = enable_metrics
+        self._enable_debug_metrics = enable_debug_metrics
+
         if isinstance(credentials, auth_credentials.AnonymousCredentials):
             if client_options is None or client_options.api_endpoint is None:
                 raise ValueError(
@@ -99,6 +117,18 @@ def __init__(
             attempt_direct_path=attempt_direct_path,
         )
 
+    @property
+    def metrics_enabled(self) -> bool:
+        """(Experimental) Returns ``True`` if OpenTelemetry metrics recording is active for this client."""
+        return _opentelemetry_metrics.is_metrics_enabled(self._enable_metrics)
+
+    @property
+    def debug_metrics_enabled(self) -> bool:
+        """(Experimental) Returns ``True`` if debug metrics recording is active for this client."""
+        return _opentelemetry_metrics.is_debug_metrics_enabled(
+            self._enable_debug_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_debug_metrics: bool or None
+    :param enable_debug_metrics:
+        (Optional, Experimental) Whether to enable debug OpenTelemetry
+        metrics. If ``None``, falls back to the
+        ``GCP_STORAGE_PYTHON_ENABLE_OTEL_DEBUG_METRICS`` environment variable,
+        or ``False`` if unset.
     """
 
     SCOPE = (
@@ -146,6 +160,8 @@ def __init__(
         extra_headers={},
         *,
         api_key=None,
+        enable_metrics=None,
+        enable_debug_metrics=None,
     ):
         self._base_connection = None
 
@@ -163,6 +179,8 @@ def __init__(
         self._initial_client_info = client_info
         self._initial_client_options = client_options
         self._extra_headers = extra_headers
+        self._enable_metrics = enable_metrics
+        self._enable_debug_metrics = enable_debug_metrics
 
         connection_kw_args = {"client_info": client_info}
 
@@ -294,6 +312,18 @@ def __init__(
         self._batch_stack = _LocalStack()
         self._bucket_metadata_cache = BucketMetadataCache(self)
 
+    @property
+    def metrics_enabled(self) -> bool:
+        """(Experimental) Returns ``True`` if OpenTelemetry metrics recording is active for this client."""
+        return _opentelemetry_metrics.is_metrics_enabled(self._enable_metrics)
+
+    @property
+    def debug_metrics_enabled(self) -> bool:
+        """(Experimental) Returns ``True`` if debug metrics recording is active for this client."""
+        return _opentelemetry_metrics.is_debug_metrics_enabled(
+            self._enable_debug_metrics
+        )
+
     def close(self):
         """Close the client and clear any cached metadata or active connections."""
         if hasattr(self, "_bucket_metadata_cache") and self._bucket_metadata_cache:

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_debug_metrics: bool or None
+    :param enable_debug_metrics:
+        (Optional, Experimental) Whether to enable debug OpenTelemetry
+        metrics. If ``None``, falls back to the
+        ``GCP_STORAGE_PYTHON_ENABLE_OTEL_DEBUG_METRICS`` environment variable,
+        or ``False`` if unset.
     """
 
     def __init__(
@@ -69,9 +83,14 @@ def __init__(
         *,
         api_key=None,
         attempt_direct_path=True,
+        enable_metrics=None,
+        enable_debug_metrics=None,
     ):
         super(GrpcClient, self).__init__(project=project, credentials=credentials)
 
+        self._enable_metrics = enable_metrics
+        self._enable_debug_metrics = enable_debug_metrics
+
         if isinstance(client_options, dict):
             if api_key:
                 client_options["api_key"] = api_key
@@ -87,6 +106,18 @@ def __init__(
             attempt_direct_path=attempt_direct_path,
         )
 
+    @property
+    def metrics_enabled(self) -> bool:
+        """(Experimental) Returns ``True`` if OpenTelemetry metrics recording is active for this client."""
+        return _opentelemetry_metrics.is_metrics_enabled(self._enable_metrics)
+
+    @property
+    def debug_metrics_enabled(self) -> bool:
+        """(Experimental) Returns ``True`` if debug metrics recording is active for this client."""
+        return _opentelemetry_metrics.is_debug_metrics_enabled(
+            self._enable_debug_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_debug_metrics = getattr(cl, "_enable_debug_metrics", None)
 
     return _LazyClient, (
         client_object_id,
@@ -1392,6 +1394,8 @@ def _reduce_client(cl):
         client_info,
         client_options,
         extra_headers,
+        enable_metrics,
+        enable_debug_metrics,
     )
 
 
@@ -1453,6 +1457,13 @@ def _digest_ordered_checksum_and_size_pairs(checksum_and_size_pairs):
     return crc_digest
 
 
+# Number of positional Client constructor arguments passed after `id` by
+# `_reduce_client`: (project, credentials, _http, client_info, client_options,
+# extra_headers). Trailing items in `args` correspond to keyword-only
+# `enable_metrics` and `enable_debug_metrics` parameters.
+_REDUCE_CLIENT_POS_ARGS = 6
+
+
 class _LazyClient:
     """An object that will transform into either a cached or a new Client"""
 
@@ -1461,6 +1472,13 @@ def __new__(cls, id, *args, **kwargs):
         if cached_client:
             return cached_client
         else:
+            if len(args) > _REDUCE_CLIENT_POS_ARGS:
+                kwargs.setdefault("enable_metrics", args[_REDUCE_CLIENT_POS_ARGS])
+                if len(args) > _REDUCE_CLIENT_POS_ARGS + 1:
+                    kwargs.setdefault(
+                        "enable_debug_metrics", args[_REDUCE_CLIENT_POS_ARGS + 1]
+                    )
+                args = args[:_REDUCE_CLIENT_POS_ARGS]
             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,78 @@ 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_debug_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.debug_metrics_enabled is True
+
+            client._enable_metrics = False
+            assert client.metrics_enabled is False
+            assert client.debug_metrics_enabled is True
+
+            client._enable_debug_metrics = False
+            assert client.debug_metrics_enabled is False
+
+        with mock.patch.multiple(
+            _opentelemetry_metrics,
+            HAS_OPENTELEMETRY_METRICS=True,
+            _ENABLE_METRICS_DEV_GATE=False,
+        ):
+            client._enable_metrics = True
+            client._enable_debug_metrics = True
+            assert client.metrics_enabled is False
+            assert client.debug_metrics_enabled is False
+
+    @mock.patch("google.cloud._storage_v2.StorageAsyncClient")
+    def test_ctor_metrics_defaults_and_env_fallback(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)
+
+        assert client._enable_metrics is None
+        assert client._enable_debug_metrics is None
+
+        with mock.patch.multiple(
+            _opentelemetry_metrics,
+            HAS_OPENTELEMETRY_METRICS=True,
+            _ENABLE_METRICS_DEV_GATE=True,
+        ):
+            with mock.patch.dict("os.environ", {}, clear=True):
+                assert client.metrics_enabled is False
+                assert client.debug_metrics_enabled is False
+
+            with mock.patch.dict(
+                "os.environ",
+                {
+                    "GCP_STORAGE_PYTHON_ENABLE_OTEL_METRICS": "true",
+                    "GCP_STORAGE_PYTHON_ENABLE_OTEL_DEBUG_METRICS": "true",
+                },
+                clear=True,
+            ):
+                assert client.metrics_enabled is True
+                assert client.debug_metrics_enabled is True
+
     @mock.patch("google.cloud._storage_v2.StorageAsyncClient")
     def test_grpc_client_property(self, mock_grpc_gapic_client):
         # Arrange