Skip to content
This repository has been archived by the owner on Sep 26, 2023. It is now read-only.

Add setAttemptDirectPath(boolean) for InstantiatingGrpcChannelProvider.Builder #1015

Merged
merged 5 commits into from May 28, 2020
Merged

Add setAttemptDirectPath(boolean) for InstantiatingGrpcChannelProvider.Builder #1015

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
4799fce
Use boolean flag instead of env var to determine whether to attempt D…
WeiranFang Apr 5, 2020
f081f24
Remove EnvironmentProvider because we are not using DirectPath env va…
WeiranFang Apr 6, 2020
93b5589
Keep DP env var checker until setAttemptDirectPath is adapted
WeiranFang Apr 22, 2020
fdfbbb8
fix logic OR
WeiranFang Apr 22, 2020
0a36964
Check env var only when attemptDirectPath is not set
WeiranFang Apr 23, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
27 changes: 17 additions & 10 deletions gax-grpc/src/main/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProvider.java
View file
Open in desktop
Expand Up @@ -79,6 +79,7 @@ public final class InstantiatingGrpcChannelProvider implements TransportChannelP
private final ExecutorProvider executorProvider;
private final HeaderProvider headerProvider;
private final String endpoint;
private final boolean attemptDirectPath;
igorbernstein2 marked this conversation as resolved.
Show resolved Hide resolved
private final EnvironmentProvider envProvider;
@Nullable private final GrpcInterceptorProvider interceptorProvider;
@Nullable private final Integer maxInboundMessageSize;
Expand Down Expand Up @@ -109,6 +110,7 @@ private InstantiatingGrpcChannelProvider(Builder builder) {
this.channelConfigurator = builder.channelConfigurator;
this.credentials = builder.credentials;
this.channelPrimer = builder.channelPrimer;
this.attemptDirectPath = builder.attemptDirectPath;
}

@Override
Expand Down Expand Up @@ -213,7 +215,7 @@ public ManagedChannel createSingleChannel() throws IOException {

// The environment variable is used during the rollout phase for directpath.
// This checker function will be removed once directpath is stable.
private boolean isDirectPathEnabled(String serviceAddress) {
private boolean isDirectPathEnvVarEnabled(String serviceAddress) {
String whiteList = envProvider.getenv(DIRECT_PATH_ENV_VAR);
if (whiteList == null) return false;
for (String service : whiteList.split(",")) {
Expand All @@ -238,9 +240,11 @@ private ManagedChannel createSingleChannel() throws IOException {

ManagedChannelBuilder builder;

// TODO(weiranf): Add a new API in ComputeEngineCredentials to check whether it's using default
// service account.
if (isDirectPathEnabled(serviceAddress) && credentials instanceof ComputeEngineCredentials) {
// TODO(weiranf): Add API in ComputeEngineCredentials to check default service account.
// TODO(weiranf): Remove isDirectPathEnvVarEnabled once setAttemptDirectPath is adapted and the
// env var is removed from client environment.
if ((attemptDirectPath | isDirectPathEnvVarEnabled(serviceAddress))
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please use ||
I think attemptDirectPath should be tristated:

  • unset: use env var
  • true: enable direct path
  • false: disable direct path

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice catch! not sure what I was thinking..

&& credentials instanceof ComputeEngineCredentials) {
builder = ComputeEngineChannelBuilder.forAddress(serviceAddress, port);
// Set default keepAliveTime and keepAliveTimeout when directpath environment is enabled.
// Will be overridden by user defined values if any.
Expand Down Expand Up @@ -353,6 +357,7 @@ public static final class Builder {
private HeaderProvider headerProvider;
private String endpoint;
private EnvironmentProvider envProvider;
private boolean attemptDirectPath;
@Nullable private GrpcInterceptorProvider interceptorProvider;
@Nullable private Integer maxInboundMessageSize;
@Nullable private Integer maxInboundMetadataSize;
Expand Down Expand Up @@ -385,6 +390,7 @@ private Builder(InstantiatingGrpcChannelProvider provider) {
this.channelConfigurator = provider.channelConfigurator;
this.credentials = provider.credentials;
this.channelPrimer = provider.channelPrimer;
this.attemptDirectPath = provider.attemptDirectPath;
}

/** Sets the number of available CPUs, used internally for testing. */
Expand All @@ -393,12 +399,6 @@ Builder setProcessorCount(int processorCount) {
return this;
}

/** Sets the environment variable provider used for testing. */
Builder setEnvironmentProvider(EnvironmentProvider envProvider) {
this.envProvider = envProvider;
return this;
}

/**
* Sets the ExecutorProvider for this TransportChannelProvider.
*
Expand Down Expand Up @@ -565,6 +565,13 @@ public Builder setChannelPrimer(ChannelPrimer channelPrimer) {
return this;
}

/** Whether attempt DirectPath. */
@InternalApi("For internal use by google-cloud-java clients only")
public Builder setAttemptDirectPath(boolean attemptDirectPath) {
igorbernstein2 marked this conversation as resolved.
Show resolved Hide resolved
this.attemptDirectPath = attemptDirectPath;
return this;
}

public InstantiatingGrpcChannelProvider build() {
return new InstantiatingGrpcChannelProvider(this);
}
Expand Down
27 changes: 6 additions & 21 deletions gax-grpc/src/test/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProviderTest.java
View file
Open in desktop
Expand Up @@ -29,15 +29,13 @@
*/
package com.google.api.gax.grpc;

import static com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.DIRECT_PATH_ENV_VAR;
import static com.google.common.truth.Truth.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

import com.google.api.core.ApiFunction;
import com.google.api.gax.core.ExecutorProvider;
import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.Builder;
import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.EnvironmentProvider;
import com.google.api.gax.rpc.HeaderProvider;
import com.google.api.gax.rpc.TransportChannelProvider;
import com.google.auth.oauth2.CloudShellCredentials;
Expand Down Expand Up @@ -215,9 +213,6 @@ public void testChannelConfigurator() throws IOException {

@Test
public void testWithGCECredentials() throws IOException {
EnvironmentProvider mockEnvProvider = Mockito.mock(EnvironmentProvider.class);
Mockito.when(mockEnvProvider.getenv(DIRECT_PATH_ENV_VAR)).thenReturn("localhost");

ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1);
executor.shutdown();

Expand All @@ -231,7 +226,7 @@ public ManagedChannelBuilder apply(ManagedChannelBuilder channelBuilder) {

TransportChannelProvider provider =
InstantiatingGrpcChannelProvider.newBuilder()
.setEnvironmentProvider(mockEnvProvider)
.setAttemptDirectPath(true)
.setChannelConfigurator(channelConfigurator)
.build()
.withExecutor(executor)
Expand All @@ -247,9 +242,6 @@ public ManagedChannelBuilder apply(ManagedChannelBuilder channelBuilder) {

@Test
public void testWithNonGCECredentials() throws IOException {
EnvironmentProvider mockEnvProvider = Mockito.mock(EnvironmentProvider.class);
Mockito.when(mockEnvProvider.getenv(DIRECT_PATH_ENV_VAR)).thenReturn("localhost");

ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1);
executor.shutdown();

Expand All @@ -264,7 +256,7 @@ public ManagedChannelBuilder apply(ManagedChannelBuilder channelBuilder) {

TransportChannelProvider provider =
InstantiatingGrpcChannelProvider.newBuilder()
.setEnvironmentProvider(mockEnvProvider)
.setAttemptDirectPath(true)
.setChannelConfigurator(channelConfigurator)
.build()
.withExecutor(executor)
Expand All @@ -280,24 +272,21 @@ public ManagedChannelBuilder apply(ManagedChannelBuilder channelBuilder) {

@Test
public void testWithDirectPathDisabled() throws IOException {
EnvironmentProvider mockEnvProvider = Mockito.mock(EnvironmentProvider.class);
Mockito.when(mockEnvProvider.getenv(DIRECT_PATH_ENV_VAR)).thenReturn("otherhost");

ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1);
executor.shutdown();

ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder> channelConfigurator =
new ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder>() {
public ManagedChannelBuilder apply(ManagedChannelBuilder channelBuilder) {
// Clients without DirectPath environment variable will not attempt DirectPath
// Clients without setting attemptDirectPath flag will not attempt DirectPath
assertThat(channelBuilder instanceof ComputeEngineChannelBuilder).isFalse();
return channelBuilder;
}
};

TransportChannelProvider provider =
InstantiatingGrpcChannelProvider.newBuilder()
.setEnvironmentProvider(mockEnvProvider)
.setAttemptDirectPath(false)
.setChannelConfigurator(channelConfigurator)
.build()
.withExecutor(executor)
Expand All @@ -312,25 +301,21 @@ public ManagedChannelBuilder apply(ManagedChannelBuilder channelBuilder) {
}

@Test
public void testWithNoDirectPathEnvironment() throws IOException {
EnvironmentProvider mockEnvProvider = Mockito.mock(EnvironmentProvider.class);
Mockito.when(mockEnvProvider.getenv(DIRECT_PATH_ENV_VAR)).thenReturn(null);

public void testWithNoDirectPathFlagSet() throws IOException {
ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1);
executor.shutdown();

ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder> channelConfigurator =
new ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder>() {
public ManagedChannelBuilder apply(ManagedChannelBuilder channelBuilder) {
// Clients without DirectPath environment variable will not attempt DirectPath
// Clients without setting attemptDirectPath flag will not attempt DirectPath
assertThat(channelBuilder instanceof ComputeEngineChannelBuilder).isFalse();
return channelBuilder;
}
};

TransportChannelProvider provider =
InstantiatingGrpcChannelProvider.newBuilder()
.setEnvironmentProvider(mockEnvProvider)
.setChannelConfigurator(channelConfigurator)
.build()
.withExecutor(executor)
Expand Down