You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by jr...@apache.org on 2023/08/31 14:02:25 UTC
[beam] branch master updated: remove sync stub list from GrpcWindmillServer (#28016)
This is an automated email from the ASF dual-hosted git repository.
jrmccluskey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new a43ee6fd983 remove sync stub list from GrpcWindmillServer (#28016)
a43ee6fd983 is described below
commit a43ee6fd983528d56b07ce0da393183f3c522b1a
Author: martin trieu <ma...@google.com>
AuthorDate: Thu Aug 31 07:02:16 2023 -0700
remove sync stub list from GrpcWindmillServer (#28016)
---
.../grpcclient/GrpcDeadlineClientInterceptor.java | 58 ++++++++
.../windmill/grpcclient/GrpcWindmillServer.java | 154 +++++++--------------
.../grpcclient/GrpcWindmillServerTest.java | 71 ++++++++++
3 files changed, 181 insertions(+), 102 deletions(-)
diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/grpcclient/GrpcDeadlineClientInterceptor.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/grpcclient/GrpcDeadlineClientInterceptor.java
new file mode 100644
index 00000000000..6b0e19cbb48
--- /dev/null
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/grpcclient/GrpcDeadlineClientInterceptor.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker.windmill.grpcclient;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.CallOptions;
+import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.Channel;
+import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.ClientCall;
+import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.ClientInterceptor;
+import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.MethodDescriptor;
+
+/**
+ * Intercepts outgoing calls and attaches a relative deadline to the call. Deadlines are absolute,
+ * so they must be created and attached with every RPC call. The alternative to registering an
+ * {@link ClientInterceptor} would be to remember to call stub.withDeadlineAfter(...) at every call
+ * site. For more context see <a
+ * href=https://github.com/grpc/grpc-java/issues/1495>grpc-java/issues/1495</a> and <a
+ * href=https://github.com/grpc/grpc-java/issues/4305>grpc-java/issues/4305</a>.
+ *
+ * @see <a href=https://grpc.io/blog/deadlines/>Official gRPC Deadline Documentation.</a>
+ */
+class GrpcDeadlineClientInterceptor implements ClientInterceptor {
+ private static final int DEFAULT_UNARY_RPC_DEADLINE_SECONDS = 10;
+ private static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.SECONDS;
+
+ private final int duration;
+ private final TimeUnit timeUnit;
+
+ GrpcDeadlineClientInterceptor(int duration, TimeUnit timeUnit) {
+ this.duration = duration;
+ this.timeUnit = timeUnit;
+ }
+
+ static GrpcDeadlineClientInterceptor withDefaultUnaryRpcDeadline() {
+ return new GrpcDeadlineClientInterceptor(DEFAULT_UNARY_RPC_DEADLINE_SECONDS, DEFAULT_TIME_UNIT);
+ }
+
+ @Override
+ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+ MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions, Channel next) {
+ return next.newCall(methodDescriptor, callOptions.withDeadlineAfter(duration, timeUnit));
+ }
+}
diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/grpcclient/GrpcWindmillServer.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/grpcclient/GrpcWindmillServer.java
index 5003f9948c5..e8745e265ee 100644
--- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/grpcclient/GrpcWindmillServer.java
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/grpcclient/GrpcWindmillServer.java
@@ -62,7 +62,6 @@ import org.apache.beam.sdk.util.BackOff;
import org.apache.beam.sdk.util.BackOffUtils;
import org.apache.beam.sdk.util.FluentBackoff;
import org.apache.beam.sdk.util.Sleeper;
-import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.CallCredentials;
import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.Channel;
import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.StatusRuntimeException;
import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.auth.MoreCallCredentials;
@@ -80,7 +79,7 @@ import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/** gRPC client for communicating with Windmill Service. */
+/** gRPC client for communicating with Streaming Engine. */
// Very likely real potential for bugs - https://github.com/apache/beam/issues/19273
// Very likely real potential for bugs - https://github.com/apache/beam/issues/19271
@SuppressFBWarnings({"JLM_JSR166_UTILCONCURRENT_MONITORENTER", "IS2_INCONSISTENT_SYNC"})
@@ -92,7 +91,6 @@ public final class GrpcWindmillServer extends WindmillServerStub {
// If a connection cannot be established, gRPC will fail fast so this deadline can be relatively
// high.
- private static final long DEFAULT_UNARY_RPC_DEADLINE_SECONDS = 300;
private static final long DEFAULT_STREAM_RPC_DEADLINE_SECONDS = 300;
private static final int DEFAULT_LOG_EVERY_N_FAILURES = 20;
private static final String LOCALHOST = "localhost";
@@ -104,15 +102,12 @@ public final class GrpcWindmillServer extends WindmillServerStub {
private final StreamingDataflowWorkerOptions options;
private final int streamingRpcBatchLimit;
private final List<CloudWindmillServiceV1Alpha1Grpc.CloudWindmillServiceV1Alpha1Stub> stubList;
- private final List<CloudWindmillServiceV1Alpha1Grpc.CloudWindmillServiceV1Alpha1BlockingStub>
- syncStubList;
private final ThrottleTimer getWorkThrottleTimer;
private final ThrottleTimer getDataThrottleTimer;
private final ThrottleTimer commitWorkThrottleTimer;
private final Random rand;
private final Set<AbstractWindmillStream<?, ?>> streamRegistry;
- private long unaryDeadlineSeconds;
private ImmutableSet<HostAndPort> endpoints;
private int logEveryNStreamFailures;
private Duration maxBackoff = MAX_BACKOFF;
@@ -122,7 +117,6 @@ public final class GrpcWindmillServer extends WindmillServerStub {
this.options = options;
this.streamingRpcBatchLimit = options.getWindmillServiceStreamingRpcBatchLimit();
this.stubList = new ArrayList<>();
- this.syncStubList = new ArrayList<>();
this.logEveryNStreamFailures = options.getWindmillServiceStreamingLogEveryNStreamFailures();
this.endpoints = ImmutableSet.of();
this.getWorkThrottleTimer = new ThrottleTimer();
@@ -130,10 +124,9 @@ public final class GrpcWindmillServer extends WindmillServerStub {
this.commitWorkThrottleTimer = new ThrottleTimer();
this.rand = new Random();
this.streamRegistry = Collections.newSetFromMap(new ConcurrentHashMap<>());
- this.unaryDeadlineSeconds = DEFAULT_UNARY_RPC_DEADLINE_SECONDS;
}
- private static StreamingDataflowWorkerOptions testOptions() {
+ private static StreamingDataflowWorkerOptions testOptions(boolean enableStreamingEngine) {
StreamingDataflowWorkerOptions options =
PipelineOptionsFactory.create().as(StreamingDataflowWorkerOptions.class);
options.setProject("project");
@@ -141,7 +134,9 @@ public final class GrpcWindmillServer extends WindmillServerStub {
options.setWorkerId("worker");
List<String> experiments =
options.getExperiments() == null ? new ArrayList<>() : options.getExperiments();
- experiments.add(GcpOptions.STREAMING_ENGINE_EXPERIMENT);
+ if (enableStreamingEngine) {
+ experiments.add(GcpOptions.STREAMING_ENGINE_EXPERIMENT);
+ }
options.setExperiments(experiments);
options.setWindmillServiceStreamingRpcBatchLimit(Integer.MAX_VALUE);
@@ -172,11 +167,26 @@ public final class GrpcWindmillServer extends WindmillServerStub {
@VisibleForTesting
static GrpcWindmillServer newTestInstance(String name) {
- GrpcWindmillServer testServer = new GrpcWindmillServer(testOptions());
+ GrpcWindmillServer testServer =
+ new GrpcWindmillServer(testOptions(/* enableStreamingEngine= */ true));
testServer.stubList.add(CloudWindmillServiceV1Alpha1Grpc.newStub(inProcessChannel(name)));
return testServer;
}
+ @VisibleForTesting
+ static GrpcWindmillServer newApplianceTestInstance(Channel channel) {
+ GrpcWindmillServer testServer =
+ new GrpcWindmillServer(testOptions(/* enableStreamingEngine= */ false));
+ testServer.syncApplianceStub = createWindmillApplianceStubWithDeadlineInterceptor(channel);
+ return testServer;
+ }
+
+ private static WindmillApplianceGrpc.WindmillApplianceBlockingStub
+ createWindmillApplianceStubWithDeadlineInterceptor(Channel channel) {
+ return WindmillApplianceGrpc.newBlockingStub(channel)
+ .withInterceptors(GrpcDeadlineClientInterceptor.withDefaultUnaryRpcDeadline());
+ }
+
private static Channel inProcessChannel(String name) {
return InProcessChannelBuilder.forName(name).directExecutor().build();
}
@@ -188,6 +198,12 @@ public final class GrpcWindmillServer extends WindmillServerStub {
.build();
}
+ private static UnsupportedOperationException unsupportedUnaryRequestInStreamingEngineException(
+ String rpcName) {
+ return new UnsupportedOperationException(
+ String.format("Unary %s calls are not supported in Streaming Engine.", rpcName));
+ }
+
private void scheduleHealthCheckTimer(
StreamingDataflowWorkerOptions options, Supplier<Set<AbstractWindmillStream<?, ?>>> streams) {
new Timer("WindmillHealthCheckTimer")
@@ -238,13 +254,11 @@ public final class GrpcWindmillServer extends WindmillServerStub {
private synchronized void initializeLocalHost(int port) {
this.logEveryNStreamFailures = 1;
this.maxBackoff = Duration.millis(500);
- this.unaryDeadlineSeconds = 10; // For local testing use short deadlines.
Channel channel = localhostChannel(port);
if (options.isEnableStreamingEngine()) {
this.stubList.add(CloudWindmillServiceV1Alpha1Grpc.newStub(channel));
- this.syncStubList.add(CloudWindmillServiceV1Alpha1Grpc.newBlockingStub(channel));
} else {
- this.syncApplianceStub = WindmillApplianceGrpc.newBlockingStub(channel);
+ this.syncApplianceStub = createWindmillApplianceStubWithDeadlineInterceptor(channel);
}
}
@@ -252,20 +266,16 @@ public final class GrpcWindmillServer extends WindmillServerStub {
throws IOException {
LOG.info("Initializing Streaming Engine GRPC client for endpoints: {}", endpoints);
this.stubList.clear();
- this.syncStubList.clear();
this.endpoints = ImmutableSet.copyOf(endpoints);
for (HostAndPort endpoint : this.endpoints) {
if (LOCALHOST.equals(endpoint.getHost())) {
initializeLocalHost(endpoint.getPort());
} else {
- CallCredentials creds =
- MoreCallCredentials.from(new VendoredCredentialsAdapter(options.getGcpCredential()));
this.stubList.add(
CloudWindmillServiceV1Alpha1Grpc.newStub(remoteChannel(endpoint))
- .withCallCredentials(creds));
- this.syncStubList.add(
- CloudWindmillServiceV1Alpha1Grpc.newBlockingStub(remoteChannel(endpoint))
- .withCallCredentials(creds));
+ .withCallCredentials(
+ MoreCallCredentials.from(
+ new VendoredCredentialsAdapter(options.getGcpCredential()))));
}
}
}
@@ -301,17 +311,6 @@ public final class GrpcWindmillServer extends WindmillServerStub {
return stubList.get(rand.nextInt(stubList.size()));
}
- private synchronized CloudWindmillServiceV1Alpha1Grpc.CloudWindmillServiceV1Alpha1BlockingStub
- syncStub() {
- if (syncStubList.isEmpty()) {
- throw new RuntimeException("windmillServiceEndpoint has not been set");
- }
- if (syncStubList.size() == 1) {
- return syncStubList.get(0);
- }
- return syncStubList.get(rand.nextInt(syncStubList.size()));
- }
-
@Override
public void appendSummaryHtml(PrintWriter writer) {
writer.write("Active Streams:<br>");
@@ -360,69 +359,28 @@ public final class GrpcWindmillServer extends WindmillServerStub {
@Override
public GetWorkResponse getWork(GetWorkRequest request) {
- if (syncApplianceStub == null) {
- return callWithBackoff(
- () ->
- syncStub()
- .withDeadlineAfter(unaryDeadlineSeconds, TimeUnit.SECONDS)
- .getWork(
- request
- .toBuilder()
- .setJobId(options.getJobId())
- .setProjectId(options.getProject())
- .setWorkerId(options.getWorkerId())
- .build()));
- } else {
- return callWithBackoff(
- () ->
- syncApplianceStub
- .withDeadlineAfter(unaryDeadlineSeconds, TimeUnit.SECONDS)
- .getWork(request));
+ if (syncApplianceStub != null) {
+ return callWithBackoff(() -> syncApplianceStub.getWork(request));
}
+
+ throw new RpcException(unsupportedUnaryRequestInStreamingEngineException("GetWork"));
}
@Override
public GetDataResponse getData(GetDataRequest request) {
- if (syncApplianceStub == null) {
- return callWithBackoff(
- () ->
- syncStub()
- .withDeadlineAfter(unaryDeadlineSeconds, TimeUnit.SECONDS)
- .getData(
- request
- .toBuilder()
- .setJobId(options.getJobId())
- .setProjectId(options.getProject())
- .build()));
- } else {
- return callWithBackoff(
- () ->
- syncApplianceStub
- .withDeadlineAfter(unaryDeadlineSeconds, TimeUnit.SECONDS)
- .getData(request));
+ if (syncApplianceStub != null) {
+ return callWithBackoff(() -> syncApplianceStub.getData(request));
}
+
+ throw new RpcException(unsupportedUnaryRequestInStreamingEngineException("GetData"));
}
@Override
public CommitWorkResponse commitWork(CommitWorkRequest request) {
- if (syncApplianceStub == null) {
- return callWithBackoff(
- () ->
- syncStub()
- .withDeadlineAfter(unaryDeadlineSeconds, TimeUnit.SECONDS)
- .commitWork(
- request
- .toBuilder()
- .setJobId(options.getJobId())
- .setProjectId(options.getProject())
- .build()));
- } else {
- return callWithBackoff(
- () ->
- syncApplianceStub
- .withDeadlineAfter(unaryDeadlineSeconds, TimeUnit.SECONDS)
- .commitWork(request));
+ if (syncApplianceStub != null) {
+ return callWithBackoff(() -> syncApplianceStub.commitWork(request));
}
+ throw new RpcException(unsupportedUnaryRequestInStreamingEngineException("CommitWork"));
}
private StreamObserverFactory newStreamObserverFactory() {
@@ -480,30 +438,22 @@ public final class GrpcWindmillServer extends WindmillServerStub {
@Override
public GetConfigResponse getConfig(GetConfigRequest request) {
- if (syncApplianceStub == null) {
- throw new RpcException(
- new UnsupportedOperationException("GetConfig not supported with windmill service."));
- } else {
- return callWithBackoff(
- () ->
- syncApplianceStub
- .withDeadlineAfter(unaryDeadlineSeconds, TimeUnit.SECONDS)
- .getConfig(request));
+ if (syncApplianceStub != null) {
+ return callWithBackoff(() -> syncApplianceStub.getConfig(request));
}
+
+ throw new RpcException(
+ new UnsupportedOperationException("GetConfig not supported in Streaming Engine."));
}
@Override
public ReportStatsResponse reportStats(ReportStatsRequest request) {
- if (syncApplianceStub == null) {
- throw new RpcException(
- new UnsupportedOperationException("ReportStats not supported with windmill service."));
- } else {
- return callWithBackoff(
- () ->
- syncApplianceStub
- .withDeadlineAfter(unaryDeadlineSeconds, TimeUnit.SECONDS)
- .reportStats(request));
+ if (syncApplianceStub != null) {
+ return callWithBackoff(() -> syncApplianceStub.reportStats(request));
}
+
+ throw new RpcException(
+ new UnsupportedOperationException("ReportStats not supported in Streaming Engine."));
}
@Override
diff --git a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/grpcclient/GrpcWindmillServerTest.java b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/grpcclient/GrpcWindmillServerTest.java
index 511b4e4895a..53afc6990e4 100644
--- a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/grpcclient/GrpcWindmillServerTest.java
+++ b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/grpcclient/GrpcWindmillServerTest.java
@@ -18,6 +18,8 @@
package org.apache.beam.runners.dataflow.worker.windmill.grpcclient;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -65,13 +67,22 @@ import org.apache.beam.runners.dataflow.worker.windmill.Windmill.TagValue;
import org.apache.beam.runners.dataflow.worker.windmill.Windmill.Value;
import org.apache.beam.runners.dataflow.worker.windmill.Windmill.WorkItem;
import org.apache.beam.runners.dataflow.worker.windmill.Windmill.WorkItemCommitRequest;
+import org.apache.beam.runners.dataflow.worker.windmill.WindmillApplianceGrpc;
import org.apache.beam.runners.dataflow.worker.windmill.WindmillStream.CommitWorkStream;
import org.apache.beam.runners.dataflow.worker.windmill.WindmillStream.GetDataStream;
import org.apache.beam.runners.dataflow.worker.windmill.WindmillStream.GetWorkStream;
import org.apache.beam.vendor.grpc.v1p54p0.com.google.protobuf.ByteString;
+import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.CallOptions;
+import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.Channel;
+import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.ClientCall;
+import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.ClientInterceptor;
+import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.ClientInterceptors;
+import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.Deadline;
+import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.MethodDescriptor;
import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.Server;
import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.Status;
import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.StatusRuntimeException;
+import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.inprocess.InProcessChannelBuilder;
import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.inprocess.InProcessServerBuilder;
import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.stub.StreamObserver;
import org.apache.beam.vendor.grpc.v1p54p0.io.grpc.util.MutableHandlerRegistry;
@@ -136,6 +147,66 @@ public class GrpcWindmillServerTest {
}
}
+ @Test
+ public void testWindmillApplianceGrpcDeadlineInterceptor() {
+ Windmill.GetWorkResponse getWorkResponse =
+ Windmill.GetWorkResponse.newBuilder()
+ .addWork(
+ Windmill.ComputationWorkItems.newBuilder()
+ .setComputationId("someComputation1")
+ .build())
+ .build();
+
+ serviceRegistry.addService(
+ new WindmillApplianceGrpc.WindmillApplianceImplBase() {
+ @Override
+ public void getWork(
+ Windmill.GetWorkRequest request,
+ StreamObserver<Windmill.GetWorkResponse> responseObserver) {
+ responseObserver.onNext(getWorkResponse);
+ responseObserver.onCompleted();
+ }
+ });
+
+ ClientInterceptor testInterceptor =
+ new ClientInterceptor() {
+ Deadline deadline;
+
+ @Override
+ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+ MethodDescriptor<ReqT, RespT> methodDescriptor,
+ CallOptions callOptions,
+ Channel channel) {
+ assertNotNull(callOptions.getDeadline());
+ if (deadline != null) {
+ // We want to assert that a new deadline is being created on every call since
+ // deadlines are absolute.
+ assertNotEquals(deadline, callOptions.getDeadline());
+ }
+
+ deadline = callOptions.getDeadline();
+ // Just forward the call once we record the deadline
+ return channel.newCall(methodDescriptor, callOptions);
+ }
+ };
+
+ Channel inprocessChannel =
+ ClientInterceptors.intercept(
+ InProcessChannelBuilder.forName("Fake server for " + getClass())
+ .directExecutor()
+ .build(),
+ testInterceptor);
+
+ this.client = GrpcWindmillServer.newApplianceTestInstance(inprocessChannel);
+
+ Windmill.GetWorkResponse response1 = client.getWork(GetWorkRequest.getDefaultInstance());
+ Windmill.GetWorkResponse response2 = client.getWork(GetWorkRequest.getDefaultInstance());
+
+ // Assert that the request/responses are being forwarded correctly.
+ assertEquals(response1, getWorkResponse);
+ assertEquals(response2, getWorkResponse);
+ }
+
@Test
public void testStreamingGetWork() throws Exception {
// This fake server returns an infinite stream of identical WorkItems, obeying the request size