feat(transport): add an env variable to disable DirectPath (#1447)

Add an environment variable GOOGLE_CLOUD_DISABLE_DIRECT_PATH to disable DirectPath. We expect users to use the EnableDirectPath option to control DirectPath for most cases, and this environment variable is for special cases like running both DP and GFE traffic on the same VM. As a result, only when this env variable is explicitly set to true will DirectPath be disabled.

Same PR for Java: googleapis/gax-java#1412.

transport/grpc/dial.go

@@ -31,6 +31,9 @@ import (
 	_ "google.golang.org/grpc/balancer/grpclb"
 )
 
+// Check env to disable DirectPath traffic.
+const disableDirectPath = "GOOGLE_CLOUD_DISABLE_DIRECT_PATH"
+
 // Check env to decide if using google-c2p resolver for DirectPath traffic.
 const enableDirectPathXds = "GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS"
 
@@ -140,7 +143,7 @@ func dial(ctx context.Context, insecure bool, o *internal.DialSettings) (*grpc.C
 		}
 
 		// Attempt Direct Path:
-		if o.EnableDirectPath && checkDirectPathEndPoint(endpoint) && isTokenSourceDirectPathCompatible(creds.TokenSource, o) && metadata.OnGCE() {
+		if isDirectPathEnabled(endpoint, o) && isTokenSourceDirectPathCompatible(creds.TokenSource, o) && metadata.OnGCE() {
 			grpcOpts = []grpc.DialOption{
 				grpc.WithCredentialsBundle(grpcgoogle.NewDefaultCredentialsWithOptions(grpcgoogle.DefaultCredentialsOptions{oauth.TokenSource{creds.TokenSource}}))}
 			if timeoutDialerOption != nil {
@@ -234,6 +237,19 @@ func (ts grpcTokenSource) GetRequestMetadata(ctx context.Context, uri ...string)
 	return metadata, nil
 }
 
+func isDirectPathEnabled(endpoint string, o *internal.DialSettings) bool {
+	if !o.EnableDirectPath {
+		return false
+	}
+	if !checkDirectPathEndPoint(endpoint) {
+		return false
+	}
+	if strings.EqualFold(os.Getenv(disableDirectPath), "true") {
+		return false
+	}
+	return true
+}
+
 func isTokenSourceDirectPathCompatible(ts oauth2.TokenSource, o *internal.DialSettings) bool {
 	if ts == nil {
 		return false