Remote: Async upload (Part 4)

Remove RemoteCache#upload. Use UploadManifest#upload directly in RemoteExecutionService.

Part of #13655.

PiperOrigin-RevId: 394606309

src/main/java/com/google/devtools/build/lib/remote/RemoteCache.java

@@ -19,9 +19,7 @@
 import static com.google.devtools.build.lib.remote.util.Utils.bytesCountToDisplayString;
 import static com.google.devtools.build.lib.remote.util.Utils.getFromFuture;
 
-import build.bazel.remote.execution.v2.Action;
 import build.bazel.remote.execution.v2.ActionResult;
-import build.bazel.remote.execution.v2.Command;
 import build.bazel.remote.execution.v2.Digest;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableSet;
@@ -31,7 +29,6 @@
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
-import com.google.devtools.build.lib.actions.ExecException;
 import com.google.devtools.build.lib.concurrent.ThreadSafety;
 import com.google.devtools.build.lib.exec.SpawnProgressEvent;
 import com.google.devtools.build.lib.exec.SpawnRunner.SpawnExecutionContext;
@@ -42,13 +39,11 @@
 import com.google.devtools.build.lib.remote.common.RemoteCacheClient;
 import com.google.devtools.build.lib.remote.common.RemoteCacheClient.ActionKey;
 import com.google.devtools.build.lib.remote.common.RemoteCacheClient.CachedActionResult;
-import com.google.devtools.build.lib.remote.common.RemotePathResolver;
 import com.google.devtools.build.lib.remote.options.RemoteOptions;
 import com.google.devtools.build.lib.remote.util.DigestUtil;
 import com.google.devtools.build.lib.server.FailureDetails.FailureDetail;
 import com.google.devtools.build.lib.server.FailureDetails.RemoteExecution;
 import com.google.devtools.build.lib.server.FailureDetails.RemoteExecution.Code;
-import com.google.devtools.build.lib.util.io.FileOutErr;
 import com.google.devtools.build.lib.util.io.OutErr;
 import com.google.devtools.build.lib.vfs.FileSystemUtils;
 import com.google.devtools.build.lib.vfs.Path;
@@ -57,7 +52,6 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.regex.Matcher;
@@ -136,35 +130,6 @@ public final ListenableFuture<Void> uploadBlob(
     return cacheProtocol.uploadBlob(context, digest, data);
   }
 
-  /**
-   * Upload the result of a locally executed action to the remote cache.
-   *
-   * @throws IOException if there was an error uploading to the remote cache
-   * @throws ExecException if uploading any of the action outputs is not supported
-   */
-  public ActionResult upload(
-      RemoteActionExecutionContext context,
-      RemotePathResolver remotePathResolver,
-      ActionKey actionKey,
-      Action action,
-      Command command,
-      Collection<Path> outputs,
-      FileOutErr outErr)
-      throws ExecException, IOException, InterruptedException {
-    UploadManifest manifest =
-        UploadManifest.create(
-            options,
-            digestUtil,
-            remotePathResolver,
-            actionKey,
-            action,
-            command,
-            outputs,
-            outErr,
-            /* exitCode= */ 0);
-    return manifest.upload(context, this);
-  }
-
   public static void waitForBulkTransfer(
       Iterable<? extends ListenableFuture<?>> transfers, boolean cancelRemainingOnInterrupt)
       throws BulkTransferException, InterruptedException {

src/main/java/com/google/devtools/build/lib/remote/RemoteExecutionService.java

@@ -52,6 +52,7 @@
 import build.bazel.remote.execution.v2.RequestMetadata;
 import build.bazel.remote.execution.v2.SymlinkNode;
 import build.bazel.remote.execution.v2.Tree;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
@@ -70,6 +71,7 @@
 import com.google.devtools.build.lib.actions.FileArtifactValue.RemoteFileArtifactValue;
 import com.google.devtools.build.lib.actions.ForbiddenActionInputException;
 import com.google.devtools.build.lib.actions.Spawn;
+import com.google.devtools.build.lib.actions.SpawnResult;
 import com.google.devtools.build.lib.actions.Spawns;
 import com.google.devtools.build.lib.actions.UserExecException;
 import com.google.devtools.build.lib.actions.cache.MetadataInjector;
@@ -999,23 +1001,37 @@ public InMemoryOutput downloadOutputs(RemoteAction action, RemoteActionResult re
     return null;
   }
 
-  /** Upload outputs of a remote action which was executed locally to remote cache. */
-  public void uploadOutputs(RemoteAction action)
-      throws InterruptedException, IOException, ExecException {
-    checkState(shouldUploadLocalResults(action.spawn), "spawn shouldn't upload local result");
-
+  @VisibleForTesting
+  UploadManifest buildUploadManifest(RemoteAction action, SpawnResult spawnResult)
+      throws ExecException, IOException {
     Collection<Path> outputFiles =
         action.spawn.getOutputFiles().stream()
             .map((inp) -> execRoot.getRelative(inp.getExecPath()))
             .collect(ImmutableList.toImmutableList());
-    remoteCache.upload(
-        action.remoteActionExecutionContext,
+
+    return UploadManifest.create(
+        remoteOptions,
+        digestUtil,
         remotePathResolver,
         action.actionKey,
         action.action,
         action.command,
         outputFiles,
-        action.spawnExecutionContext.getFileOutErr());
+        action.spawnExecutionContext.getFileOutErr(),
+        /* exitCode= */ 0);
+  }
+
+  /** Upload outputs of a remote action which was executed locally to remote cache. */
+  public void uploadOutputs(RemoteAction action, SpawnResult spawnResult)
+      throws InterruptedException, IOException, ExecException {
+    checkState(shouldUploadLocalResults(action.spawn), "spawn shouldn't upload local result");
+    checkState(
+        SpawnResult.Status.SUCCESS.equals(spawnResult.status()) && spawnResult.exitCode() == 0,
+        "shouldn't upload outputs of failed local action");
+
+    UploadManifest manifest = buildUploadManifest(action, spawnResult);
+
+    manifest.upload(action.getRemoteActionExecutionContext(), remoteCache);
   }
 
   /**

src/main/java/com/google/devtools/build/lib/remote/RemoteSpawnCache.java

@@ -198,7 +198,7 @@ public void store(SpawnResult result) throws ExecException, InterruptedException
           }
 
           try (SilentCloseable c = prof.profile(ProfilerTask.UPLOAD_TIME, "upload outputs")) {
-            remoteExecutionService.uploadOutputs(action);
+            remoteExecutionService.uploadOutputs(action, result);
           } catch (IOException e) {
             String errorMessage;
             if (!verboseFailures) {

src/main/java/com/google/devtools/build/lib/remote/RemoteSpawnRunner.java

@@ -123,6 +123,11 @@ public class RemoteSpawnRunner implements SpawnRunner {
     this.remoteExecutionService = remoteExecutionService;
   }
 
+  @VisibleForTesting
+  RemoteExecutionService getRemoteExecutionService() {
+    return remoteExecutionService;
+  }
+
   @Override
   public String getName() {
     return "remote";
@@ -566,7 +571,7 @@ SpawnResult execLocallyAndUpload(
     }
 
     try (SilentCloseable c = Profiler.instance().profile(UPLOAD_TIME, "upload outputs")) {
-      remoteExecutionService.uploadOutputs(action);
+      remoteExecutionService.uploadOutputs(action, result);
     } catch (IOException e) {
       if (verboseFailures) {
         report(Event.debug("Upload to remote cache failed: " + e.getMessage()));

src/main/java/com/google/devtools/build/lib/remote/UploadManifest.java

@@ -337,6 +337,11 @@ private void illegalOutput(Path what) throws ExecException {
     throw new UserExecException(failureDetail);
   }
 
+  @VisibleForTesting
+  ActionResult getActionResult() {
+    return result.build();
+  }
+
   /** Uploads outputs and action result (if exit code is 0) to remote cache. */
   public ActionResult upload(RemoteActionExecutionContext context, RemoteCache remoteCache)
       throws IOException, InterruptedException {

src/test/java/com/google/devtools/build/lib/remote/GrpcCacheClientTest.java

@@ -556,12 +556,33 @@ public void updateActionResult(
     assertThat(result).isEqualTo(expectedResult.build());
   }
 
+  private ActionResult upload(
+      RemoteCache remoteCache,
+      ActionKey actionKey,
+      Action action,
+      Command command,
+      List<Path> outputs)
+      throws Exception {
+    UploadManifest uploadManifest =
+        UploadManifest.create(
+            remoteCache.options,
+            remoteCache.digestUtil,
+            remotePathResolver,
+            actionKey,
+            action,
+            command,
+            outputs,
+            outErr,
+            0);
+    return uploadManifest.upload(context, remoteCache);
+  }
+
   private ActionResult uploadDirectory(RemoteCache remoteCache, List<Path> outputs)
       throws Exception {
     Action action = Action.getDefaultInstance();
     ActionKey actionKey = DIGEST_UTIL.computeActionKey(action);
     Command cmd = Command.getDefaultInstance();
-    return remoteCache.upload(context, remotePathResolver, actionKey, action, cmd, outputs, outErr);
+    return upload(remoteCache, actionKey, action, cmd, outputs);
   }
 
   @Test
@@ -686,14 +707,12 @@ public void updateActionResult(
         });
 
     ActionResult result =
-        remoteCache.upload(
-            context,
-            remotePathResolver,
+        upload(
+            remoteCache,
             DIGEST_UTIL.asActionKey(actionDigest),
             action,
             command,
-            ImmutableList.of(fooFile, barFile),
-            outErr);
+            ImmutableList.of(fooFile, barFile));
     ActionResult.Builder expectedResult = ActionResult.newBuilder();
     expectedResult.setStdoutDigest(stdoutDigest);
     expectedResult.setStderrDigest(stderrDigest);
@@ -749,14 +768,12 @@ public void updateActionResult(
         });
 
     ActionResult result =
-        remoteCache.upload(
-            context,
-            remotePathResolver,
+        upload(
+            remoteCache,
             DIGEST_UTIL.asActionKey(actionDigest),
             action,
             command,
-            ImmutableList.of(fooFile, barFile),
-            outErr);
+            ImmutableList.of(fooFile, barFile));
     ActionResult.Builder expectedResult = ActionResult.newBuilder();
     expectedResult.addOutputFilesBuilder().setPath("a/foo").setDigest(fooDigest);
     expectedResult
@@ -898,14 +915,12 @@ public void onError(Throwable t) {
                 }))
         .when(mockByteStreamImpl)
         .queryWriteStatus(any(), any());
-    remoteCache.upload(
-        context,
-        remotePathResolver,
+    upload(
+        remoteCache,
         actionKey,
         Action.getDefaultInstance(),
         Command.getDefaultInstance(),
-        ImmutableList.<Path>of(fooFile, barFile, bazFile),
-        outErr);
+        ImmutableList.<Path>of(fooFile, barFile, bazFile));
     // 4 times for the errors, 3 times for the successful uploads.
     Mockito.verify(mockByteStreamImpl, Mockito.times(7))
         .write(ArgumentMatchers.<StreamObserver<WriteResponse>>any());