Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: Make refreshing channel compatible with BigtableDataClientFactory #474

Merged
merged 6 commits into from Oct 22, 2020
Merged

fix: Make refreshing channel compatible with BigtableDataClientFactory #474

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
61d5e63
Make refreshing channel compatible with BigtableDataClientFactory
mutianf Oct 16, 2020
45673db
Fix formatting
mutianf Oct 16, 2020
139254d
Fix test case
mutianf Oct 18, 2020
2131123
Add ServerTransportFilter in FakeServiceHelper
mutianf Oct 19, 2020
2ff4007
Fix credentials provider settings add added test to verify it
mutianf Oct 21, 2020
d26e17a
Clean up test cases and comments
mutianf Oct 22, 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
5 changes: 3 additions & 2 deletions ...d-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/BigtableDataClientFactory.java
View file
Open in desktop
Expand Up @@ -23,7 +23,7 @@
import com.google.api.gax.rpc.FixedHeaderProvider;
import com.google.api.gax.rpc.FixedTransportChannelProvider;
import com.google.api.gax.rpc.FixedWatchdogProvider;
import com.google.api.gax.rpc.StubSettings;
import com.google.cloud.bigtable.data.v2.stub.EnhancedBigtableStubSettings;
import java.io.IOException;
import javax.annotation.Nonnull;

Expand Down Expand Up @@ -189,8 +189,9 @@ public BigtableDataClient createForInstance(
}

// Update stub settings to use shared resources in this factory
private void patchStubSettings(StubSettings.Builder stubSettings) {
private void patchStubSettings(EnhancedBigtableStubSettings.Builder stubSettings) {
stubSettings
.setRefreshingChannel(false)
igorbernstein2 marked this conversation as resolved.
Show resolved Hide resolved
.setTransportChannelProvider(
FixedTransportChannelProvider.create(sharedClientContext.getTransportChannel()))
.setCredentialsProvider(
Expand Down
15 changes: 15 additions & 0 deletions ...le/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings.java
View file
Open in desktop
Expand Up @@ -42,6 +42,7 @@
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;
Expand Down Expand Up @@ -806,6 +807,20 @@ public EnhancedBigtableStubSettings build() {
Preconditions.checkArgument(
getTransportChannelProvider() instanceof InstantiatingGrpcChannelProvider,
"refreshingChannel only works with InstantiatingGrpcChannelProviders");
InstantiatingGrpcChannelProvider.Builder channelProviderBuilder =
((InstantiatingGrpcChannelProvider) getTransportChannelProvider()).toBuilder();
try {
channelProviderBuilder.setChannelPrimer(
BigtableChannelPrimer.create(
getCredentialsProvider().getCredentials(),
Copy link
Contributor

Choose a reason for hiding this comment

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

I think there is a subtle bug here:

If the original CredentialsProvider was a GoogleCredentialsProvider then each time getCredentials is called, it will be a new instance, which means the credentials used for refreshing will diverge from the normal request credentials.

I think you need to wrap reset the credentials to be FixedCredentials here.

Also please move the getCredentials line into its own try/catch its kinda hard to see which line can throw the IOException

Copy link
Collaborator

Choose a reason for hiding this comment

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

To add here, can we add a test to ensure the credentials are consistent?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Updated BigtableDataClientFactoryTest.testCreateWithRefreshingChannel test to verify that getCredentials is only called once by the credentialProvider we set (similar to test testNewClientsShareTransportChannel.testNewClientsShareTransportChannel). Also added a test in EnhancedBigtableStubSettingsTest to make sure it's returning the same credentials.

projectId,
instanceId,
appProfileId,
primedTableIds));
} catch (IOException e) {
throw new RuntimeException(e);
}
this.setTransportChannelProvider(channelProviderBuilder.build());
}
return new EnhancedBigtableStubSettings(this);
}
Expand Down
98 changes: 95 additions & 3 deletions ...gtable/src/test/java/com/google/cloud/bigtable/data/v2/BigtableDataClientFactoryTest.java
View file
Open in desktop
Expand Up @@ -18,19 +18,32 @@
import static com.google.common.truth.Truth.assertThat;

import com.google.api.core.ApiClock;
import com.google.api.core.ApiFunction;
import com.google.api.gax.core.CredentialsProvider;
import com.google.api.gax.core.ExecutorProvider;
import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
import com.google.api.gax.rpc.TransportChannelProvider;
import com.google.api.gax.rpc.WatchdogProvider;
import com.google.bigtable.v2.BigtableGrpc;
import com.google.bigtable.v2.MutateRowRequest;
import com.google.bigtable.v2.MutateRowResponse;
import com.google.bigtable.v2.ReadRowsRequest;
import com.google.bigtable.v2.ReadRowsResponse;
import com.google.bigtable.v2.RowFilter;
import com.google.bigtable.v2.RowSet;
import com.google.cloud.bigtable.data.v2.internal.NameUtil;
import com.google.cloud.bigtable.data.v2.models.RowMutation;
import com.google.common.base.Preconditions;
import com.google.protobuf.ByteString;
import io.grpc.Attributes;
import io.grpc.ServerTransportFilter;
import io.grpc.stub.StreamObserver;
import java.io.IOException;
import java.lang.reflect.Method;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
Expand Down Expand Up @@ -60,16 +73,27 @@ public class BigtableDataClientFactoryTest {
private WatchdogProvider watchdogProvider;
private ApiClock apiClock;
private BigtableDataSettings defaultSettings;
private int port;

private final BlockingQueue<Attributes> attributes = new LinkedBlockingDeque<>();

@Before
public void setUp() throws IOException {
service = new FakeBigtableService();

serviceHelper = new FakeServiceHelper(service);
ServerTransportFilter transportFilter =
new ServerTransportFilter() {
@Override
public Attributes transportReady(Attributes transportAttrs) {
attributes.add(transportAttrs);
return super.transportReady(transportAttrs);
}
};
serviceHelper = new FakeServiceHelper(null, transportFilter, service);
port = serviceHelper.getPort();
serviceHelper.start();

BigtableDataSettings.Builder builder =
BigtableDataSettings.newBuilderForEmulator(serviceHelper.getPort())
BigtableDataSettings.newBuilderForEmulator(port)
.setProjectId(DEFAULT_PROJECT_ID)
.setInstanceId(DEFAULT_INSTANCE_ID)
.setAppProfileId(DEFAULT_APP_PROFILE_ID);
Expand Down Expand Up @@ -191,8 +215,75 @@ public void testCreateForInstanceWithAppProfileHasCorrectSettings() throws Excep
assertThat(service.lastRequest.getAppProfileId()).isEqualTo("other-app-profile");
}

@Test
public void testCreateWithRefreshingChannel() throws Exception {
String[] tableIds = {"fake-table1", "fake-table2"};
int poolSize = 3;
BigtableDataSettings.Builder builder =
BigtableDataSettings.newBuilderForEmulator(port)
.setProjectId(DEFAULT_PROJECT_ID)
.setInstanceId(DEFAULT_INSTANCE_ID)
.setAppProfileId(DEFAULT_APP_PROFILE_ID)
.setPrimingTableIds(tableIds)
.setRefreshingChannel(true);
InstantiatingGrpcChannelProvider channelProvider =
(InstantiatingGrpcChannelProvider) builder.stubSettings().getTransportChannelProvider();
InstantiatingGrpcChannelProvider.Builder channelProviderBuilder = channelProvider.toBuilder();
channelProviderBuilder.setPoolSize(poolSize);
builder.stubSettings().setTransportChannelProvider(channelProviderBuilder.build());

BigtableDataClientFactory factory = BigtableDataClientFactory.create(builder.build());
factory.createDefault();
factory.createForAppProfile("other-appprofile");
factory.createForInstance("other-project", "other-instance");

// Make sure that the clients are sharing the same ChannelPool
assertThat(attributes).hasSize(poolSize);
// Make sure that prime requests were sent only once per table per connection
assertThat(service.readRowsRequests).hasSize(poolSize * tableIds.length);
List<ReadRowsRequest> expectedRequests = new LinkedList<>();
for (String tableId : tableIds) {
for (int i = 0; i < poolSize; i++) {
expectedRequests.add(
ReadRowsRequest.newBuilder()
.setTableName(
String.format(
"projects/%s/instances/%s/tables/%s",
DEFAULT_PROJECT_ID, DEFAULT_INSTANCE_ID, tableId))
.setAppProfileId(DEFAULT_APP_PROFILE_ID)
.setRows(
RowSet.newBuilder()
.addRowKeys(ByteString.copyFromUtf8("nonexistent-priming-row")))
.setFilter(RowFilter.newBuilder().setBlockAllFilter(true).build())
.setRowsLimit(1)
.build());
}
}
assertThat(service.readRowsRequests).containsExactly(expectedRequests.toArray());
}

private static class FakeBigtableService extends BigtableGrpc.BigtableImplBase {
volatile MutateRowRequest lastRequest;
BlockingQueue<ReadRowsRequest> readRowsRequests = new LinkedBlockingDeque<>();
private ApiFunction<ReadRowsRequest, ReadRowsResponse> readRowsCallback =
new ApiFunction<ReadRowsRequest, ReadRowsResponse>() {
@Override
public ReadRowsResponse apply(ReadRowsRequest readRowsRequest) {
return ReadRowsResponse.getDefaultInstance();
}
};

@Override
public void readRows(
ReadRowsRequest request, StreamObserver<ReadRowsResponse> responseObserver) {
try {
readRowsRequests.add(request);
responseObserver.onNext(readRowsCallback.apply(request));
responseObserver.onCompleted();
} catch (RuntimeException e) {
responseObserver.onError(e);
}
}

@Override
public void mutateRow(
Expand All @@ -204,6 +295,7 @@ public void mutateRow(
}

private static class BuilderAnswer<T> implements Answer {

private final Class<T> targetClass;
private T targetInstance;

Expand Down
12 changes: 12 additions & 0 deletions google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/FakeServiceHelper.java
View file
Open in desktop
Expand Up @@ -19,6 +19,7 @@
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.ServerInterceptor;
import io.grpc.ServerTransportFilter;
import java.io.IOException;
import java.net.ServerSocket;

Expand All @@ -33,13 +34,24 @@ public FakeServiceHelper(BindableService... services) throws IOException {

public FakeServiceHelper(ServerInterceptor interceptor, BindableService... services)
throws IOException {
this(interceptor, null, services);
}

public FakeServiceHelper(
ServerInterceptor interceptor,
ServerTransportFilter transportFilter,
BindableService... services)
throws IOException {
try (ServerSocket ss = new ServerSocket(0)) {
port = ss.getLocalPort();
}
ServerBuilder builder = ServerBuilder.forPort(port);
if (interceptor != null) {
builder = builder.intercept(interceptor);
}
if (transportFilter != null) {
builder = builder.addTransportFilter(transportFilter);
}
for (BindableService service : services) {
builder = builder.addService(service);
}
Expand Down