feat(storage): support DirectPath over Interconnect in gRPC client

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

@@ -22,8 +22,11 @@
     DEFAULT_CLIENT_INFO,
 )
 from google.cloud.storage import __version__
-
-_DEFAULT_HOST = "storage.googleapis.com"
+from google.cloud.storage.grpc_client import (
+    _DEFAULT_HOST,
+    _resolve_direct_path_interconnect,
+    _rewrite_host_for_interconnect,
+)
 
 
 def _validate_metadata(metadata):
@@ -63,6 +66,13 @@ class AsyncGrpcClient:
     :param attempt_direct_path:
         (Optional) Whether to attempt to use DirectPath for gRPC connections.
         Defaults to ``True``.
+
+    :type attempt_direct_path_xds_over_interconnect: bool
+    :param attempt_direct_path_xds_over_interconnect:
+        (Optional) Whether to attempt DirectPath over Cloud Interconnect
+        using xDS and standard TLS. Can also be overridden via the
+        ``GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS_OVER_INTERCONNECT`` environment
+        variable (``"true"`` or ``"false"``). Defaults to ``False``.
     """
 
     def __init__(
@@ -72,6 +82,7 @@ def __init__(
         client_options=None,
         *,
         attempt_direct_path=True,
+        attempt_direct_path_xds_over_interconnect=False,
     ):
         if isinstance(credentials, auth_credentials.AnonymousCredentials):
             if client_options is None or client_options.api_endpoint is None:
@@ -92,11 +103,16 @@ def __init__(
         if agent_version not in client_info.user_agent:
             client_info.user_agent += f" {agent_version} "
 
+        use_dp_interconnect = _resolve_direct_path_interconnect(
+            attempt_direct_path_xds_over_interconnect
+        )
+
         self._grpc_client = self._create_async_grpc_client(
             credentials=credentials,
             client_info=client_info,
             client_options=client_options,
             attempt_direct_path=attempt_direct_path,
+            attempt_direct_path_xds_over_interconnect=use_dp_interconnect,
         )
 
     def _create_anonymous_client(self, client_options, credentials):
@@ -120,6 +136,7 @@ def _create_async_grpc_client(
         client_info=None,
         client_options=None,
         attempt_direct_path=True,
+        attempt_direct_path_xds_over_interconnect=False,
     ):
         transport_cls = storage_v2.StorageAsyncClient.get_transport_class(
             "grpc_asyncio"
@@ -129,17 +146,31 @@ def _create_async_grpc_client(
 
         host = _DEFAULT_HOST
         quota_project_id = None
-        if client_options:
+        if isinstance(client_options, dict):
+            host = client_options.get("api_endpoint") or _DEFAULT_HOST
+            quota_project_id = client_options.get("quota_project_id")
+        elif client_options:
             host = getattr(client_options, "api_endpoint", None) or _DEFAULT_HOST
             quota_project_id = getattr(client_options, "quota_project_id", None)
 
-        channel = transport_cls.create_channel(
-            host=host,
-            quota_project_id=quota_project_id,
-            attempt_direct_path=attempt_direct_path,
-            credentials=credentials,
-            options=(("grpc.primary_user_agent", primary_user_agent),),
-        )
+        if attempt_direct_path_xds_over_interconnect:
+            host = _rewrite_host_for_interconnect(host)
+            channel = transport_cls.create_channel(
+                host=host,
+                quota_project_id=quota_project_id,
+                attempt_direct_path=True,
+                attempt_direct_path_xds_over_interconnect=True,
+                credentials=credentials,
+                options=(("grpc.primary_user_agent", primary_user_agent),),
+            )
+        else:
+            channel = transport_cls.create_channel(
+                host=host,
+                quota_project_id=quota_project_id,
+                attempt_direct_path=attempt_direct_path,
+                credentials=credentials,
+                options=(("grpc.primary_user_agent", primary_user_agent),),
+            )
         transport = transport_cls(channel=channel)
 
         return storage_v2.StorageAsyncClient(

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

@@ -14,11 +14,49 @@
 
 """A client for interacting with Google Cloud Storage using the gRPC API."""
 
+import os
+
 from google.cloud.client import ClientWithProject
 
 from google.cloud import _storage_v2 as storage_v2
 
 _marker = object()
+_DEFAULT_HOST = "storage.googleapis.com"
+_DEFAULT_HOST_DIRECT_PATH = "storage-direct.googleapis.com"
+_DIRECT_PATH_INTERCONNECT_ENV = "GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS_OVER_INTERCONNECT"
+
+
+def _resolve_direct_path_interconnect(option_value: bool) -> bool:
+    """Resolves DirectPath over Interconnect flag from env var or parameter."""
+    env_val = os.environ.get(_DIRECT_PATH_INTERCONNECT_ENV)
+    if env_val is not None:
+        env_val_clean = env_val.strip().lower()
+        if env_val_clean == "true":
+            return True
+        elif env_val_clean == "false":
+            return False
+        else:
+            raise ValueError(
+                f"Invalid value for {_DIRECT_PATH_INTERCONNECT_ENV}: {env_val}"
+            )
+    return bool(option_value)
+
+
+def _rewrite_host_for_interconnect(
+    host: str,
+    old_host: str = _DEFAULT_HOST,
+    new_host: str = _DEFAULT_HOST_DIRECT_PATH,
+) -> str:
+    """Rewrites the default GCS endpoint to the DirectPath over Interconnect host."""
+    if not host:
+        return new_host
+    for prefix in ("", "https://", "http://", "dns:///"):
+        candidate = f"{prefix}{old_host}"
+        if host.startswith(candidate):
+            rest = host[len(candidate) :]
+            if not rest or rest[0] in (":", "/", "?", "#"):
+                return f"{prefix}{new_host}{rest}"
+    return host
 
 
 class GrpcClient(ClientWithProject):
@@ -58,6 +96,13 @@ 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 attempt_direct_path_xds_over_interconnect: bool
+    :param attempt_direct_path_xds_over_interconnect:
+        (Optional) Whether to attempt DirectPath over Cloud Interconnect
+        using xDS and standard TLS. Can also be overridden via the
+        ``GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS_OVER_INTERCONNECT`` environment
+        variable (``"true"`` or ``"false"``). Defaults to ``False``.
     """
 
     def __init__(
@@ -69,6 +114,7 @@ def __init__(
         *,
         api_key=None,
         attempt_direct_path=True,
+        attempt_direct_path_xds_over_interconnect=False,
     ):
         super(GrpcClient, self).__init__(project=project, credentials=credentials)
 
@@ -80,11 +126,16 @@ def __init__(
         elif api_key:
             client_options.api_key = api_key
 
+        use_dp_interconnect = _resolve_direct_path_interconnect(
+            attempt_direct_path_xds_over_interconnect
+        )
+
         self._grpc_client = self._create_gapic_client(
             credentials=credentials,
             client_info=client_info,
             client_options=client_options,
             attempt_direct_path=attempt_direct_path,
+            attempt_direct_path_xds_over_interconnect=use_dp_interconnect,
         )
 
     def _create_gapic_client(
@@ -93,11 +144,34 @@ def _create_gapic_client(
         client_info=None,
         client_options=None,
         attempt_direct_path=True,
+        attempt_direct_path_xds_over_interconnect=False,
     ):
         """Creates and configures the low-level GAPIC `storage_v2` client."""
         transport_cls = storage_v2.StorageClient.get_transport_class("grpc")
 
-        channel = transport_cls.create_channel(attempt_direct_path=attempt_direct_path)
+        if attempt_direct_path_xds_over_interconnect:
+            host = _DEFAULT_HOST
+            quota_project_id = None
+            if isinstance(client_options, dict):
+                host = client_options.get("api_endpoint") or _DEFAULT_HOST
+                quota_project_id = client_options.get("quota_project_id")
+            elif client_options is not None:
+                host = getattr(client_options, "api_endpoint", None) or _DEFAULT_HOST
+                quota_project_id = getattr(client_options, "quota_project_id", None)
+            host = _rewrite_host_for_interconnect(host)
+            channel_kwargs = {
+                "host": host,
+                "credentials": credentials,
+                "attempt_direct_path": True,
+                "attempt_direct_path_xds_over_interconnect": True,
+            }
+            if quota_project_id is not None:
+                channel_kwargs["quota_project_id"] = quota_project_id
+            channel = transport_cls.create_channel(**channel_kwargs)
+        else:
+            channel = transport_cls.create_channel(
+                attempt_direct_path=attempt_direct_path
+            )
 
         transport = transport_cls(credentials=credentials, channel=channel)
 

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

@@ -431,3 +431,119 @@ async def test_get_object_invalid_metadata(self, invalid_metadata, expected_exc)
         client = async_grpc_client.AsyncGrpcClient(credentials=_make_credentials())
         with pytest.raises(expected_exc):
             await client.get_object("bucket", "object", metadata=invalid_metadata)
+
+    @mock.patch("google.cloud._storage_v2.StorageAsyncClient")
+    def test_async_grpc_client_attempt_direct_path_xds_over_interconnect(
+        self, mock_async_storage_client
+    ):
+        mock_transport_cls = mock.MagicMock()
+        mock_async_storage_client.get_transport_class.return_value = mock_transport_cls
+        mock_creds = _make_credentials()
+
+        async_grpc_client.AsyncGrpcClient(
+            credentials=mock_creds,
+            attempt_direct_path_xds_over_interconnect=True,
+        )
+
+        kwargs = mock_async_storage_client.call_args.kwargs
+        client_info = kwargs["client_info"]
+        primary_user_agent = client_info.to_user_agent()
+        expected_options = (("grpc.primary_user_agent", primary_user_agent),)
+
+        mock_transport_cls.create_channel.assert_called_once_with(
+            host="storage-direct.googleapis.com",
+            quota_project_id=None,
+            attempt_direct_path=True,
+            attempt_direct_path_xds_over_interconnect=True,
+            credentials=mock_creds,
+            options=expected_options,
+        )
+
+    @mock.patch("google.cloud._storage_v2.StorageAsyncClient")
+    def test_create_gapic_client_custom_client_options_with_interconnect(
+        self, mock_async_storage_client
+    ):
+        mock_transport_cls = mock.MagicMock()
+        mock_async_storage_client.get_transport_class.return_value = mock_transport_cls
+        mock_creds = _make_credentials()
+
+        client_options_obj = client_options.ClientOptions(
+            api_endpoint="custom.endpoint.com",
+            quota_project_id="my-quota-proj",
+        )
+        async_grpc_client.AsyncGrpcClient(
+            credentials=mock_creds,
+            client_options=client_options_obj,
+            attempt_direct_path_xds_over_interconnect=True,
+        )
+
+        kwargs = mock_async_storage_client.call_args.kwargs
+        client_info = kwargs["client_info"]
+        primary_user_agent = client_info.to_user_agent()
+        expected_options = (("grpc.primary_user_agent", primary_user_agent),)
+
+        mock_transport_cls.create_channel.assert_called_once_with(
+            host="custom.endpoint.com",
+            quota_project_id="my-quota-proj",
+            attempt_direct_path=True,
+            attempt_direct_path_xds_over_interconnect=True,
+            credentials=mock_creds,
+            options=expected_options,
+        )
+
+    @mock.patch("google.cloud._storage_v2.StorageAsyncClient")
+    def test_async_grpc_client_interconnect_env_override(
+        self, mock_async_storage_client
+    ):
+        mock_transport_cls = mock.MagicMock()
+        mock_async_storage_client.get_transport_class.return_value = mock_transport_cls
+        mock_creds = _make_credentials()
+
+        with mock.patch.dict(
+            "os.environ",
+            {"GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS_OVER_INTERCONNECT": "true"},
+        ):
+            async_grpc_client.AsyncGrpcClient(
+                credentials=mock_creds,
+                attempt_direct_path=False,
+                attempt_direct_path_xds_over_interconnect=False,
+            )
+
+        kwargs = mock_async_storage_client.call_args.kwargs
+        client_info = kwargs["client_info"]
+        primary_user_agent = client_info.to_user_agent()
+        expected_options = (("grpc.primary_user_agent", primary_user_agent),)
+
+        mock_transport_cls.create_channel.assert_called_once_with(
+            host="storage-direct.googleapis.com",
+            quota_project_id=None,
+            attempt_direct_path=True,
+            attempt_direct_path_xds_over_interconnect=True,
+            credentials=mock_creds,
+            options=expected_options,
+        )
+
+        mock_transport_cls.create_channel.reset_mock()
+        with mock.patch.dict(
+            "os.environ",
+            {"GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS_OVER_INTERCONNECT": " FALSE "},
+        ):
+            async_grpc_client.AsyncGrpcClient(
+                credentials=mock_creds,
+                attempt_direct_path_xds_over_interconnect=True,
+            )
+
+        mock_transport_cls.create_channel.assert_called_once_with(
+            host="storage.googleapis.com",
+            quota_project_id=None,
+            attempt_direct_path=True,
+            credentials=mock_creds,
+            options=expected_options,
+        )
+
+        with mock.patch.dict(
+            "os.environ",
+            {"GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS_OVER_INTERCONNECT": "invalid"},
+        ):
+            with pytest.raises(ValueError):
+                async_grpc_client.AsyncGrpcClient(credentials=mock_creds)