feat: Add attempt_direct_path argument to create_channel

google/api_core/grpc_helpers.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 """Helpers for :mod:`grpc`."""
-from typing import Generic, TypeVar, Iterator
+from typing import Generic, Iterator, Optional, TypeVar
 
 import collections
 import functools
@@ -271,11 +271,23 @@ def _create_composite_credentials(
  # Create a set of grpc.CallCredentials using the metadata plugin.
  google_auth_credentials = grpc.metadata_call_credentials(metadata_plugin)
 
- if ssl_credentials is None:
- ssl_credentials = grpc.ssl_channel_credentials()
-
- # Combine the ssl credentials and the authorization credentials.
- return grpc.composite_channel_credentials(ssl_credentials, google_auth_credentials)
+ # if `ssl_credentials` is set, use `grpc.composite_channel_credentials` instead of
+ # `grpc.compute_engine_channel_credentials` as the former supports passing
+ # `ssl_credentials` via `channel_credentials` which is needed for mTLS.
+ if ssl_credentials:
+ # Combine the ssl credentials and the authorization credentials.
+ # See https://grpc.github.io/grpc/python/grpc.html#grpc.composite_channel_credentials
+ return grpc.composite_channel_credentials(
+ ssl_credentials, google_auth_credentials
+ )
+ else:
+ # Use grpc.compute_engine_channel_credentials in order to support Direct Path.
+ # See https://grpc.github.io/grpc/python/grpc.html#grpc.compute_engine_channel_credentials
+ # TODO(b/323073050): Although `grpc.compute_engine_channel_credentials`
+ # returns channel credentials outside of GCE, we should determine if there is a way to
+ # reliably detect when the client is in a GCE environment so that
+ # `grpc.compute_engine_channel_credentials` is not called outside of GCE.
+ return grpc.compute_engine_channel_credentials(google_auth_credentials)
 
 
 def create_channel(
@@ -288,6 +300,7 @@ def create_channel(
  default_scopes=None,
  default_host=None,
  compression=None,
+ attempt_direct_path: Optional[bool] = None,
  **kwargs,
 ):
  """Create a secure channel with credentials.
@@ -311,6 +324,16 @@ def create_channel(
  default_host (str): The default endpoint. e.g., "pubsub.googleapis.com".
  compression (grpc.Compression): An optional value indicating the
  compression method to be used over the lifetime of the channel.
+ attempt_direct_path (Optional[bool]): If set, Direct Path will be attempted when
+ the request is made. Direct Path provides a proxyless connection which
+ increases the available throughput, reduces latency, and increases
+ reliability. Outside of GCE, the direct path request may fallback
+ to DNS if this is configured by the Service. This argument should only
+ be set in a GCE environment and for Services that are known to support Direct Path.
+ If a `ServiceUnavailable` response is received when the request is sent, it is
+ recommended that the client repeat the request with `attempt_direct_path` set to `False`
+ as the Service may not support Direct Path. Using `ssl_credentials` with `attempt_direct_path`
+ set to `True` will result in `ValueError` as it is not yet supported.
  kwargs: Additional key-word args passed to
  :func:`grpc_gcp.secure_channel` or :func:`grpc.secure_channel`.
  Note: `grpc_gcp` is only supported in environments with protobuf < 4.0.0.
@@ -320,8 +343,15 @@ def create_channel(
 
  Raises:
  google.api_core.DuplicateCredentialArgs: If both a credentials object and credentials_file are passed.
+ ValueError: If `ssl_credentials` is set and `attempt_direct_path` is set to `True`.
  """
 
+ # If `ssl_credentials` is set and `attempt_direct_path` is set to `True`,
+ # raise ValueError as this is not yet supported.
+ # See https://github.com/googleapis/python-api-core/issues/590
+ if ssl_credentials is not None and attempt_direct_path:
+ raise ValueError("Using ssl_credentials with Direct Path is not supported")
+
  composite_credentials = _create_composite_credentials(
  credentials=credentials,
  credentials_file=credentials_file,
@@ -332,17 +362,53 @@ def create_channel(
  default_host=default_host,
  )
 
+ # Note that grpcio-gcp is deprecated
  if HAS_GRPC_GCP: # pragma: NO COVER
  if compression is not None and compression != grpc.Compression.NoCompression:
  _LOGGER.debug(
  "Compression argument is being ignored for grpc_gcp.secure_channel creation."
  )
+ if attempt_direct_path:
+ warnings.warn(
+ """The `attempt_direct_path` argument is ignored for grpc_gcp.secure_channel creation.""",
+ DeprecationWarning,
+ )
  return grpc_gcp.secure_channel(target, composite_credentials, **kwargs)
+
+ if attempt_direct_path:
+ target = _modify_target_for_direct_path(target)
+
  return grpc.secure_channel(
  target, composite_credentials, compression=compression, **kwargs
  )
 
 
+def _modify_target_for_direct_path(target: str) -> str:
+ """Create a secure channel with credentials.
+
+ Args:
+ target (str): The target service address in the format 'hostname:port', 'dns://hostname' or other
+ compatible format.
+
+ Returns:
+ target (str): The target service address which is converted into a format compatible with Direct Path.
+ If the target contains `dns:///` or does not have contain `:///`, the target will be converted in
+ a format compatible with Direct Path, otherwise the original target will be returned.
+ """
+
+ dns_prefix = "dns:///"
+ # Remove "dns:///" if `attempt_direct_path` is set to True as
+ # the Direct Path prefix `google-c2p:///` will be used instead.
+ target = target.replace(dns_prefix, "")
+
+ direct_path_prefix = ":///"
+ if direct_path_prefix not in target:
+ target_without_port = target.split(":")[0]
+ # Modify the target to use Direct Path by adding the `google-c2p:///` prefix
+ target = f"google-c2p{direct_path_prefix}{target_without_port}"
+ return target
+
+
 _MethodCall = collections.namedtuple(
  "_MethodCall", ("request", "timeout", "metadata", "credentials", "compression")
 )

google/api_core/grpc_helpers_async.py

@@ -21,7 +21,7 @@
 import asyncio
 import functools
 
-from typing import Generic, Iterator, AsyncGenerator, TypeVar
+from typing import AsyncGenerator, Generic, Iterator, Optional, TypeVar
 
 import grpc
 from grpc import aio
@@ -223,6 +223,7 @@ def create_channel(
  default_scopes=None,
  default_host=None,
  compression=None,
+ attempt_direct_path: Optional[bool] = None,
  **kwargs
 ):
  """Create an AsyncIO secure channel with credentials.
@@ -246,15 +247,32 @@ def create_channel(
  default_host (str): The default endpoint. e.g., "pubsub.googleapis.com".
  compression (grpc.Compression): An optional value indicating the
  compression method to be used over the lifetime of the channel.
+ attempt_direct_path (Optional[bool]): If set, Direct Path will be attempted when
+ the request is made. Direct Path provides a proxyless connection which
+ increases the available throughput, reduces latency, and increases
+ reliability. Outside of GCE, the direct path request may fallback
+ to DNS if this is configured by the Service. This argument should only
+ be set in a GCE environment and for Services that are known to support Direct Path.
+ If a `ServiceUnavailable` response is received when the request is sent, it is
+ recommended that the client repeat the request with `attempt_direct_path` set to `False`
+ as the Service may not support Direct Path. Using `ssl_credentials` with `attempt_direct_path`
+ set to `True` will result in `ValueError` as it is not yet supported.
  kwargs: Additional key-word args passed to :func:`aio.secure_channel`.
 
  Returns:
  aio.Channel: The created channel.
 
  Raises:
  google.api_core.DuplicateCredentialArgs: If both a credentials object and credentials_file are passed.
+ ValueError: If `ssl_credentials` is set and `attempt_direct_path` is set to `True`.
  """
 
+ # If `ssl_credentials` is set and `attempt_direct_path` is set to `True`,
+ # raise ValueError as this is not yet supported.
+ # See https://github.com/googleapis/python-api-core/issues/590
+ if ssl_credentials is not None and attempt_direct_path:
+ raise ValueError("Using ssl_credentials with Direct Path is not supported")
+
  composite_credentials = grpc_helpers._create_composite_credentials(
  credentials=credentials,
  credentials_file=credentials_file,
@@ -265,6 +283,9 @@ def create_channel(
  default_host=default_host,
  )
 
+ if attempt_direct_path:
+ target = grpc_helpers._modify_target_for_direct_path(target)
+
  return aio.secure_channel(
  target, composite_credentials, compression=compression, **kwargs
  )

pytest.ini

@@ -12,10 +12,9 @@ filterwarnings =
  # Remove once support for grpcio-gcp is deprecated
  # See https://github.com/googleapis/python-api-core/blob/42e8b6e6f426cab749b34906529e8aaf3f133d75/google/api_core/grpc_helpers.py#L39-L45
  ignore:.*Support for grpcio-gcp is deprecated:DeprecationWarning
- # Remove once https://github.com/googleapis/python-api-common-protos/pull/187/files is merged
+ ignore:The `attempt_direct_path` argument is ignored for grpc_gcp.secure_channel creation:DeprecationWarning
+ # Remove once the minimum supported version of googleapis-common-protos is 1.62.0
  ignore:.*pkg_resources.declare_namespace:DeprecationWarning
  ignore:.*pkg_resources is deprecated as an API:DeprecationWarning
- # Remove once release PR https://github.com/googleapis/proto-plus-python/pull/391 is merged
- ignore:datetime.datetime.utcfromtimestamp\(\) is deprecated:DeprecationWarning:proto.datetime_helpers
- # Remove once https://github.com/grpc/grpc/issues/35086 is fixed
+ # Remove once https://github.com/grpc/grpc/issues/35086 is fixed (and version newer than 1.60.0 is published)
  ignore:There is no current event loop:DeprecationWarning