Switching Bazel to use the new remote execution API: https://docs.google.com/document/d/1AaGk7fOPByEvpAbqeXIyE8HX_A3_axxNnvroblTZ_6s/edit

Also refactored away the various *Interface* files, no need since unit testing can be done with mocking the appropriate gRPC Impl classes directly (see tests). This also fixes the RemoteSpawnRunner, which should use different objects for remote caching and remote execution, the same way RemoteSpawnStrategy does.

RELNOTES: n/a
PiperOrigin-RevId: 158473700

1 files changed, 38 insertions, 42 deletions

diff --git a/src/main/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutor.java b/src/main/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutor.java
index eb79c91919..1950a724c3 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutor.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/GrpcRemoteExecutor.java
@@ -15,58 +15,54 @@
 package com.google.devtools.build.lib.remote;
 
 import com.google.devtools.build.lib.concurrent.ThreadSafety.ThreadSafe;
-import com.google.devtools.build.lib.remote.RemoteProtocol.ExecuteReply;
-import com.google.devtools.build.lib.remote.RemoteProtocol.ExecuteRequest;
-import com.google.devtools.build.lib.remote.RemoteProtocol.ExecutionStatus;
-import io.grpc.ManagedChannel;
-import java.util.Iterator;
+import com.google.devtools.build.lib.util.Preconditions;
+import com.google.devtools.remoteexecution.v1test.ExecuteRequest;
+import com.google.devtools.remoteexecution.v1test.ExecuteResponse;
+import com.google.devtools.remoteexecution.v1test.ExecutionGrpc;
+import com.google.devtools.remoteexecution.v1test.ExecutionGrpc.ExecutionBlockingStub;
+import com.google.longrunning.Operation;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.util.Durations;
+import io.grpc.Channel;
+import io.grpc.protobuf.StatusProto;
+import java.util.concurrent.TimeUnit;
 
 /** A remote work executor that uses gRPC for communicating the work, inputs and outputs. */
 @ThreadSafe
-public class GrpcRemoteExecutor extends GrpcActionCache {
+public class GrpcRemoteExecutor {
+  private final RemoteOptions options;
+  private final ChannelOptions channelOptions;
+  private final Channel channel;
+
   public static boolean isRemoteExecutionOptions(RemoteOptions options) {
     return options.remoteExecutor != null;
   }
 
-  private final GrpcExecutionInterface executionIface;
-
-  public GrpcRemoteExecutor(
-      RemoteOptions options,
-      GrpcCasInterface casIface,
-      GrpcExecutionCacheInterface cacheIface,
-      GrpcExecutionInterface executionIface) {
-    super(options, casIface, cacheIface);
-    this.executionIface = executionIface;
-  }
-
-  public GrpcRemoteExecutor(
-      ManagedChannel channel, ChannelOptions channelOptions, RemoteOptions options) {
-    super(
-        options,
-        GrpcInterfaces.casInterface(options.remoteTimeout, channel, channelOptions),
-        GrpcInterfaces.executionCacheInterface(
-            options.remoteTimeout, channel, channelOptions));
-    this.executionIface =
-        GrpcInterfaces.executionInterface(options.remoteTimeout, channel, channelOptions);
+  public GrpcRemoteExecutor(Channel channel, ChannelOptions channelOptions, RemoteOptions options) {
+    this.options = options;
+    this.channelOptions = channelOptions;
+    this.channel = channel;
   }
 
-  public ExecuteReply executeRemotely(ExecuteRequest request) {
-    Iterator<ExecuteReply> replies = executionIface.execute(request);
-    ExecuteReply reply = null;
-    while (replies.hasNext()) {
-      reply = replies.next();
-      // We can handle the action execution progress here.
+  public ExecuteResponse executeRemotely(ExecuteRequest request) {
+    // TODO(olaola): handle longrunning Operations by using the Watcher API to wait for results.
+    // For now, only support actions with wait_for_completion = true.
+    Preconditions.checkArgument(request.getWaitForCompletion());
+    int actionSeconds = (int) Durations.toSeconds(request.getAction().getTimeout());
+    ExecutionBlockingStub stub =
+        ExecutionGrpc.newBlockingStub(channel)
+            .withCallCredentials(channelOptions.getCallCredentials())
+            .withDeadlineAfter(options.remoteTimeout + actionSeconds, TimeUnit.SECONDS);
+    Operation op = stub.execute(request);
+    Preconditions.checkState(op.getDone());
+    Preconditions.checkState(op.getResultCase() != Operation.ResultCase.RESULT_NOT_SET);
+    if (op.getResultCase() == Operation.ResultCase.ERROR) {
+      throw StatusProto.toStatusRuntimeException(op.getError());
     }
-    if (reply == null) {
-      return ExecuteReply.newBuilder()
-          .setStatus(
-              ExecutionStatus.newBuilder()
-                  .setExecuted(false)
-                  .setSucceeded(false)
-                  .setError(ExecutionStatus.ErrorCode.UNKNOWN_ERROR)
-                  .setErrorDetail("Remote server terminated the connection"))
-          .build();
+    try {
+      return op.getResponse().unpack(ExecuteResponse.class);
+    } catch (InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
     }
-    return reply;
   }
 }