| // Copyright 2017 The Bazel Authors. All rights reserved. | |
| // | |
| // Licensed 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 com.google.devtools.build.lib.remote; | |
| import static com.google.common.util.concurrent.MoreExecutors.directExecutor; | |
| 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 build.bazel.remote.execution.v2.Directory; | |
| import build.bazel.remote.execution.v2.DirectoryNode; | |
| import build.bazel.remote.execution.v2.FileNode; | |
| import build.bazel.remote.execution.v2.OutputDirectory; | |
| import build.bazel.remote.execution.v2.OutputFile; | |
| import build.bazel.remote.execution.v2.OutputSymlink; | |
| 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.Throwables; | |
| import com.google.common.collect.ImmutableList; | |
| import com.google.common.collect.ImmutableMap; | |
| import com.google.common.collect.ImmutableSet; | |
| import com.google.common.collect.Iterables; | |
| import com.google.common.collect.Maps; | |
| import com.google.common.util.concurrent.FutureCallback; | |
| import com.google.common.util.concurrent.Futures; | |
| import com.google.common.util.concurrent.ListenableFuture; | |
| import com.google.common.util.concurrent.SettableFuture; | |
| import com.google.devtools.build.lib.actions.ActionInput; | |
| import com.google.devtools.build.lib.actions.Artifact; | |
| import com.google.devtools.build.lib.actions.EnvironmentalExecException; | |
| import com.google.devtools.build.lib.actions.ExecException; | |
| import com.google.devtools.build.lib.actions.FileArtifactValue.RemoteFileArtifactValue; | |
| import com.google.devtools.build.lib.actions.UserExecException; | |
| import com.google.devtools.build.lib.actions.cache.MetadataInjector; | |
| import com.google.devtools.build.lib.concurrent.ThreadSafety; | |
| import com.google.devtools.build.lib.exec.SpawnRunner.SpawnExecutionContext; | |
| import com.google.devtools.build.lib.profiler.Profiler; | |
| import com.google.devtools.build.lib.profiler.SilentCloseable; | |
| import com.google.devtools.build.lib.remote.RemoteCache.ActionResultMetadata.DirectoryMetadata; | |
| import com.google.devtools.build.lib.remote.RemoteCache.ActionResultMetadata.FileMetadata; | |
| import com.google.devtools.build.lib.remote.RemoteCache.ActionResultMetadata.SymlinkMetadata; | |
| 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.options.RemoteOptions; | |
| import com.google.devtools.build.lib.remote.util.DigestUtil; | |
| import com.google.devtools.build.lib.remote.util.Utils; | |
| import com.google.devtools.build.lib.remote.util.Utils.InMemoryOutput; | |
| 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.Dirent; | |
| import com.google.devtools.build.lib.vfs.FileStatus; | |
| import com.google.devtools.build.lib.vfs.FileSystemUtils; | |
| import com.google.devtools.build.lib.vfs.Path; | |
| import com.google.devtools.build.lib.vfs.PathFragment; | |
| import com.google.devtools.build.lib.vfs.Symlinks; | |
| import com.google.protobuf.ByteString; | |
| import com.google.protobuf.InvalidProtocolBufferException; | |
| import java.io.ByteArrayOutputStream; | |
| import java.io.IOException; | |
| import java.io.OutputStream; | |
| import java.util.ArrayList; | |
| import java.util.Collection; | |
| import java.util.Collections; | |
| import java.util.Comparator; | |
| import java.util.HashMap; | |
| import java.util.List; | |
| import java.util.Map; | |
| import java.util.Map.Entry; | |
| import java.util.concurrent.ExecutionException; | |
| import java.util.stream.Collectors; | |
| import java.util.stream.Stream; | |
| import javax.annotation.Nullable; | |
| /** A cache for storing artifacts (input and output) as well as the output of running an action. */ | |
| @ThreadSafety.ThreadSafe | |
| public class RemoteCache implements AutoCloseable { | |
| /** See {@link SpawnExecutionContext#lockOutputFiles()}. */ | |
| @FunctionalInterface | |
| interface OutputFilesLocker { | |
| void lock() throws InterruptedException, IOException; | |
| } | |
| private static final ListenableFuture<Void> COMPLETED_SUCCESS = SettableFuture.create(); | |
| private static final ListenableFuture<byte[]> EMPTY_BYTES = SettableFuture.create(); | |
| static { | |
| ((SettableFuture<Void>) COMPLETED_SUCCESS).set(null); | |
| ((SettableFuture<byte[]>) EMPTY_BYTES).set(new byte[0]); | |
| } | |
| protected final RemoteCacheClient cacheProtocol; | |
| protected final RemoteOptions options; | |
| protected final DigestUtil digestUtil; | |
| public RemoteCache( | |
| RemoteCacheClient cacheProtocol, RemoteOptions options, DigestUtil digestUtil) { | |
| this.cacheProtocol = cacheProtocol; | |
| this.options = options; | |
| this.digestUtil = digestUtil; | |
| } | |
| public ActionResult downloadActionResult(ActionKey actionKey) | |
| throws IOException, InterruptedException { | |
| return Utils.getFromFuture(cacheProtocol.downloadActionResult(actionKey)); | |
| } | |
| /** | |
| * 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( | |
| ActionKey actionKey, | |
| Action action, | |
| Command command, | |
| Path execRoot, | |
| Collection<Path> outputs, | |
| FileOutErr outErr, | |
| int exitCode) | |
| throws ExecException, IOException, InterruptedException { | |
| ActionResult.Builder resultBuilder = ActionResult.newBuilder(); | |
| uploadOutputs(execRoot, actionKey, action, command, outputs, outErr, resultBuilder); | |
| resultBuilder.setExitCode(exitCode); | |
| ActionResult result = resultBuilder.build(); | |
| if (exitCode == 0 && !action.getDoNotCache()) { | |
| cacheProtocol.uploadActionResult(actionKey, result); | |
| } | |
| return result; | |
| } | |
| public ActionResult upload( | |
| ActionKey actionKey, | |
| Action action, | |
| Command command, | |
| Path execRoot, | |
| Collection<Path> outputs, | |
| FileOutErr outErr) | |
| throws ExecException, IOException, InterruptedException { | |
| return upload(actionKey, action, command, execRoot, outputs, outErr, /* exitCode= */ 0); | |
| } | |
| private void uploadOutputs( | |
| Path execRoot, | |
| ActionKey actionKey, | |
| Action action, | |
| Command command, | |
| Collection<Path> files, | |
| FileOutErr outErr, | |
| ActionResult.Builder result) | |
| throws ExecException, IOException, InterruptedException { | |
| UploadManifest manifest = | |
| new UploadManifest( | |
| digestUtil, | |
| result, | |
| execRoot, | |
| options.incompatibleRemoteSymlinks, | |
| options.allowSymlinkUpload); | |
| manifest.addFiles(files); | |
| manifest.setStdoutStderr(outErr); | |
| manifest.addAction(actionKey, action, command); | |
| Map<Digest, Path> digestToFile = manifest.getDigestToFile(); | |
| Map<Digest, ByteString> digestToBlobs = manifest.getDigestToBlobs(); | |
| Collection<Digest> digests = new ArrayList<>(); | |
| digests.addAll(digestToFile.keySet()); | |
| digests.addAll(digestToBlobs.keySet()); | |
| ImmutableSet<Digest> digestsToUpload = | |
| Utils.getFromFuture(cacheProtocol.findMissingDigests(digests)); | |
| ImmutableList.Builder<ListenableFuture<Void>> uploads = ImmutableList.builder(); | |
| for (Digest digest : digestsToUpload) { | |
| Path file = digestToFile.get(digest); | |
| if (file != null) { | |
| uploads.add(cacheProtocol.uploadFile(digest, file)); | |
| } else { | |
| ByteString blob = digestToBlobs.get(digest); | |
| if (blob == null) { | |
| String message = "FindMissingBlobs call returned an unknown digest: " + digest; | |
| throw new IOException(message); | |
| } | |
| uploads.add(cacheProtocol.uploadBlob(digest, blob)); | |
| } | |
| } | |
| waitForUploads(uploads.build()); | |
| if (manifest.getStderrDigest() != null) { | |
| result.setStderrDigest(manifest.getStderrDigest()); | |
| } | |
| if (manifest.getStdoutDigest() != null) { | |
| result.setStdoutDigest(manifest.getStdoutDigest()); | |
| } | |
| } | |
| private static void waitForUploads(List<ListenableFuture<Void>> uploads) | |
| throws IOException, InterruptedException { | |
| try { | |
| for (ListenableFuture<Void> upload : uploads) { | |
| upload.get(); | |
| } | |
| } catch (ExecutionException e) { | |
| // TODO(buchgr): Add support for cancellation and factor this method out to be shared | |
| // between ByteStreamUploader as well. | |
| Throwable cause = e.getCause(); | |
| Throwables.throwIfInstanceOf(cause, IOException.class); | |
| Throwables.throwIfInstanceOf(cause, InterruptedException.class); | |
| if (cause != null) { | |
| throw new IOException(cause); | |
| } | |
| throw new IOException(e); | |
| } | |
| } | |
| /** | |
| * Downloads a blob with content hash {@code digest} and stores its content in memory. | |
| * | |
| * @return a future that completes after the download completes (succeeds / fails). If successful, | |
| * the content is stored in the future's {@code byte[]}. | |
| */ | |
| public ListenableFuture<byte[]> downloadBlob(Digest digest) { | |
| if (digest.getSizeBytes() == 0) { | |
| return EMPTY_BYTES; | |
| } | |
| ByteArrayOutputStream bOut = new ByteArrayOutputStream((int) digest.getSizeBytes()); | |
| SettableFuture<byte[]> outerF = SettableFuture.create(); | |
| Futures.addCallback( | |
| cacheProtocol.downloadBlob(digest, bOut), | |
| new FutureCallback<Void>() { | |
| @Override | |
| public void onSuccess(Void aVoid) { | |
| outerF.set(bOut.toByteArray()); | |
| } | |
| @Override | |
| public void onFailure(Throwable t) { | |
| outerF.setException(t); | |
| } | |
| }, | |
| directExecutor()); | |
| return outerF; | |
| } | |
| private static Path toTmpDownloadPath(Path actualPath) { | |
| return actualPath.getParentDirectory().getRelative(actualPath.getBaseName() + ".tmp"); | |
| } | |
| /** | |
| * Download the output files and directory trees of a remotely executed action to the local | |
| * machine, as well stdin / stdout to the given files. | |
| * | |
| * <p>In case of failure, this method deletes any output files it might have already created. | |
| * | |
| * @param outputFilesLocker ensures that we are the only ones writing to the output files when | |
| * using the dynamic spawn strategy. | |
| * @throws IOException in case of a cache miss or if the remote cache is unavailable. | |
| * @throws ExecException in case clean up after a failed download failed. | |
| */ | |
| public void download( | |
| ActionResult result, | |
| Path execRoot, | |
| FileOutErr origOutErr, | |
| OutputFilesLocker outputFilesLocker) | |
| throws ExecException, IOException, InterruptedException { | |
| ActionResultMetadata metadata = parseActionResultMetadata(result, execRoot); | |
| List<ListenableFuture<FileMetadata>> downloads = | |
| Stream.concat( | |
| metadata.files().stream(), | |
| metadata.directories().stream() | |
| .flatMap((entry) -> entry.getValue().files().stream())) | |
| .map( | |
| (file) -> { | |
| try { | |
| ListenableFuture<Void> download = | |
| downloadFile(toTmpDownloadPath(file.path()), file.digest()); | |
| return Futures.transform(download, (d) -> file, directExecutor()); | |
| } catch (IOException e) { | |
| return Futures.<FileMetadata>immediateFailedFuture(e); | |
| } | |
| }) | |
| .collect(Collectors.toList()); | |
| // Subsequently we need to wait for *every* download to finish, even if we already know that | |
| // one failed. That's so that when exiting this method we can be sure that all downloads have | |
| // finished and don't race with the cleanup routine. | |
| // TODO(buchgr): Look into cancellation. | |
| IOException downloadException = null; | |
| InterruptedException interruptedException = null; | |
| FileOutErr tmpOutErr = null; | |
| try { | |
| if (origOutErr != null) { | |
| tmpOutErr = origOutErr.childOutErr(); | |
| } | |
| downloads.addAll(downloadOutErr(result, tmpOutErr)); | |
| } catch (IOException e) { | |
| downloadException = e; | |
| } | |
| for (ListenableFuture<FileMetadata> download : downloads) { | |
| try { | |
| // Wait for all downloads to finish. | |
| getFromFuture(download); | |
| } catch (IOException e) { | |
| if (downloadException == null) { | |
| downloadException = e; | |
| } else if (e != downloadException) { | |
| downloadException.addSuppressed(e); | |
| } | |
| } catch (InterruptedException e) { | |
| if (interruptedException == null) { | |
| interruptedException = e; | |
| } else if (e != interruptedException) { | |
| interruptedException.addSuppressed(e); | |
| } | |
| } | |
| } | |
| if (downloadException != null || interruptedException != null) { | |
| try { | |
| // Delete any (partially) downloaded output files. | |
| for (OutputFile file : result.getOutputFilesList()) { | |
| toTmpDownloadPath(execRoot.getRelative(file.getPath())).delete(); | |
| } | |
| for (OutputDirectory directory : result.getOutputDirectoriesList()) { | |
| // Only delete the directories below the output directories because the output | |
| // directories will not be re-created | |
| execRoot.getRelative(directory.getPath()).deleteTreesBelow(); | |
| } | |
| if (tmpOutErr != null) { | |
| tmpOutErr.clearOut(); | |
| tmpOutErr.clearErr(); | |
| } | |
| } catch (IOException e) { | |
| if (downloadException != null && e != downloadException) { | |
| e.addSuppressed(downloadException); | |
| } | |
| if (interruptedException != null) { | |
| e.addSuppressed(interruptedException); | |
| } | |
| // If deleting of output files failed, we abort the build with a decent error message as | |
| // any subsequent local execution failure would likely be incomprehensible. | |
| throw new EnvironmentalExecException( | |
| "Failed to delete output files after incomplete download", e); | |
| } | |
| } | |
| if (interruptedException != null) { | |
| throw interruptedException; | |
| } | |
| if (downloadException != null) { | |
| throw downloadException; | |
| } | |
| if (tmpOutErr != null) { | |
| FileOutErr.dump(tmpOutErr, origOutErr); | |
| tmpOutErr.clearOut(); | |
| tmpOutErr.clearErr(); | |
| } | |
| // Ensure that we are the only ones writing to the output files when using the dynamic spawn | |
| // strategy. | |
| outputFilesLocker.lock(); | |
| moveOutputsToFinalLocation(downloads); | |
| List<SymlinkMetadata> symlinksInDirectories = new ArrayList<>(); | |
| for (Entry<Path, DirectoryMetadata> entry : metadata.directories()) { | |
| entry.getKey().createDirectoryAndParents(); | |
| symlinksInDirectories.addAll(entry.getValue().symlinks()); | |
| } | |
| Iterable<SymlinkMetadata> symlinks = | |
| Iterables.concat(metadata.symlinks(), symlinksInDirectories); | |
| // Create the symbolic links after all downloads are finished, because dangling symlinks | |
| // might not be supported on all platforms | |
| createSymlinks(symlinks); | |
| } | |
| /** | |
| * Copies moves the downloaded outputs from their download location to their declared location. | |
| */ | |
| private void moveOutputsToFinalLocation(List<ListenableFuture<FileMetadata>> downloads) | |
| throws IOException, InterruptedException { | |
| List<FileMetadata> finishedDownloads = new ArrayList<>(downloads.size()); | |
| for (ListenableFuture<FileMetadata> finishedDownload : downloads) { | |
| FileMetadata outputFile = getFromFuture(finishedDownload); | |
| if (outputFile != null) { | |
| finishedDownloads.add(outputFile); | |
| } | |
| } | |
| /* | |
| * Sort the list lexicographically based on its temporary download path in order to avoid | |
| * filename clashes when moving the files: | |
| * | |
| * Consider an action that produces two outputs foo and foo.tmp. These outputs would initially | |
| * be downloaded to foo.tmp and foo.tmp.tmp. When renaming them to foo and foo.tmp we need to | |
| * ensure that rename(foo.tmp, foo) happens before rename(foo.tmp.tmp, foo.tmp). We ensure this | |
| * by doing the renames in lexicographical order of the download names. | |
| */ | |
| Collections.sort(finishedDownloads, Comparator.comparing(f -> toTmpDownloadPath(f.path()))); | |
| // Move the output files from their temporary name to the actual output file name. | |
| for (FileMetadata outputFile : finishedDownloads) { | |
| FileSystemUtils.moveFile(toTmpDownloadPath(outputFile.path()), outputFile.path()); | |
| outputFile.path().setExecutable(outputFile.isExecutable()); | |
| } | |
| } | |
| private void createSymlinks(Iterable<SymlinkMetadata> symlinks) throws IOException { | |
| for (SymlinkMetadata symlink : symlinks) { | |
| if (symlink.target().isAbsolute()) { | |
| // We do not support absolute symlinks as outputs. | |
| throw new IOException( | |
| String.format( | |
| "Action output %s is a symbolic link to an absolute path %s. " | |
| + "Symlinks to absolute paths in action outputs are not supported.", | |
| symlink.path(), symlink.target())); | |
| } | |
| Preconditions.checkNotNull( | |
| symlink.path().getParentDirectory(), | |
| "Failed creating directory and parents for %s", | |
| symlink.path()) | |
| .createDirectoryAndParents(); | |
| symlink.path().createSymbolicLink(symlink.target()); | |
| } | |
| } | |
| /** Download a file (that is not a directory). The content is fetched from the digest. */ | |
| public ListenableFuture<Void> downloadFile(Path path, Digest digest) throws IOException { | |
| Preconditions.checkNotNull(path.getParentDirectory()).createDirectoryAndParents(); | |
| if (digest.getSizeBytes() == 0) { | |
| // Handle empty file locally. | |
| FileSystemUtils.writeContent(path, new byte[0]); | |
| return COMPLETED_SUCCESS; | |
| } | |
| OutputStream out = new LazyFileOutputStream(path); | |
| SettableFuture<Void> outerF = SettableFuture.create(); | |
| ListenableFuture<Void> f = cacheProtocol.downloadBlob(digest, out); | |
| Futures.addCallback( | |
| f, | |
| new FutureCallback<Void>() { | |
| @Override | |
| public void onSuccess(Void result) { | |
| try { | |
| out.close(); | |
| outerF.set(null); | |
| } catch (IOException e) { | |
| outerF.setException(e); | |
| } | |
| } | |
| @Override | |
| public void onFailure(Throwable t) { | |
| try { | |
| out.close(); | |
| } catch (IOException e) { | |
| if (t != e) { | |
| t.addSuppressed(e); | |
| } | |
| } finally { | |
| outerF.setException(t); | |
| } | |
| } | |
| }, | |
| directExecutor()); | |
| return outerF; | |
| } | |
| private List<ListenableFuture<FileMetadata>> downloadOutErr(ActionResult result, OutErr outErr) | |
| throws IOException { | |
| List<ListenableFuture<FileMetadata>> downloads = new ArrayList<>(); | |
| if (!result.getStdoutRaw().isEmpty()) { | |
| result.getStdoutRaw().writeTo(outErr.getOutputStream()); | |
| outErr.getOutputStream().flush(); | |
| } else if (result.hasStdoutDigest()) { | |
| downloads.add( | |
| Futures.transform( | |
| cacheProtocol.downloadBlob(result.getStdoutDigest(), outErr.getOutputStream()), | |
| (d) -> null, | |
| directExecutor())); | |
| } | |
| if (!result.getStderrRaw().isEmpty()) { | |
| result.getStderrRaw().writeTo(outErr.getErrorStream()); | |
| outErr.getErrorStream().flush(); | |
| } else if (result.hasStderrDigest()) { | |
| downloads.add( | |
| Futures.transform( | |
| cacheProtocol.downloadBlob(result.getStderrDigest(), outErr.getErrorStream()), | |
| (d) -> null, | |
| directExecutor())); | |
| } | |
| return downloads; | |
| } | |
| /** | |
| * Avoids downloading the majority of action outputs but injects their metadata using {@link | |
| * MetadataInjector} instead. | |
| * | |
| * <p>This method only downloads output directory metadata, stdout and stderr as well as the | |
| * contents of {@code inMemoryOutputPath} if specified. | |
| * | |
| * @param result the action result metadata of a successfully executed action (exit code = 0). | |
| * @param outputs the action's declared output files | |
| * @param inMemoryOutputPath the path of an output file whose contents should be returned in | |
| * memory by this method. | |
| * @param outErr stdout and stderr of this action | |
| * @param execRoot the execution root | |
| * @param metadataInjector the action's metadata injector that allows this method to inject | |
| * metadata about an action output instead of downloading the output | |
| * @param outputFilesLocker ensures that we are the only ones writing to the output files when | |
| * using the dynamic spawn strategy. | |
| * @throws IOException in case of failure | |
| * @throws InterruptedException in case of receiving an interrupt | |
| */ | |
| @Nullable | |
| public InMemoryOutput downloadMinimal( | |
| ActionResult result, | |
| Collection<? extends ActionInput> outputs, | |
| @Nullable PathFragment inMemoryOutputPath, | |
| OutErr outErr, | |
| Path execRoot, | |
| MetadataInjector metadataInjector, | |
| OutputFilesLocker outputFilesLocker) | |
| throws IOException, InterruptedException { | |
| Preconditions.checkState( | |
| result.getExitCode() == 0, | |
| "injecting remote metadata is only supported for successful actions (exit code 0)."); | |
| ActionResultMetadata metadata; | |
| try (SilentCloseable c = Profiler.instance().profile("Remote.parseActionResultMetadata")) { | |
| metadata = parseActionResultMetadata(result, execRoot); | |
| } | |
| if (!metadata.symlinks().isEmpty()) { | |
| throw new IOException( | |
| "Symlinks in action outputs are not yet supported by " | |
| + "--experimental_remote_download_outputs=minimal"); | |
| } | |
| // Ensure that when using dynamic spawn strategy that we are the only ones writing to the | |
| // output files. | |
| outputFilesLocker.lock(); | |
| ActionInput inMemoryOutput = null; | |
| Digest inMemoryOutputDigest = null; | |
| for (ActionInput output : outputs) { | |
| if (inMemoryOutputPath != null && output.getExecPath().equals(inMemoryOutputPath)) { | |
| Path p = execRoot.getRelative(output.getExecPath()); | |
| FileMetadata m = Preconditions.checkNotNull(metadata.file(p), "inMemoryOutputMetadata"); | |
| inMemoryOutputDigest = m.digest(); | |
| inMemoryOutput = output; | |
| } | |
| if (output instanceof Artifact) { | |
| injectRemoteArtifact((Artifact) output, metadata, execRoot, metadataInjector); | |
| } | |
| } | |
| try (SilentCloseable c = Profiler.instance().profile("Remote.download")) { | |
| ListenableFuture<byte[]> inMemoryOutputDownload = null; | |
| if (inMemoryOutput != null) { | |
| inMemoryOutputDownload = downloadBlob(inMemoryOutputDigest); | |
| } | |
| for (ListenableFuture<FileMetadata> download : downloadOutErr(result, outErr)) { | |
| getFromFuture(download); | |
| } | |
| if (inMemoryOutputDownload != null) { | |
| byte[] data = getFromFuture(inMemoryOutputDownload); | |
| return new InMemoryOutput(inMemoryOutput, ByteString.copyFrom(data)); | |
| } | |
| } | |
| return null; | |
| } | |
| private void injectRemoteArtifact( | |
| Artifact output, | |
| ActionResultMetadata metadata, | |
| Path execRoot, | |
| MetadataInjector metadataInjector) | |
| throws IOException { | |
| if (output.isTreeArtifact()) { | |
| DirectoryMetadata directory = | |
| metadata.directory(execRoot.getRelative(output.getExecPathString())); | |
| if (directory == null) { | |
| // A declared output wasn't created. It might have been an optional output and if not | |
| // SkyFrame will make sure to fail. | |
| return; | |
| } | |
| if (!directory.symlinks().isEmpty()) { | |
| throw new IOException( | |
| "Symlinks in action outputs are not yet supported by " | |
| + "--experimental_remote_download_outputs=minimal"); | |
| } | |
| ImmutableMap.Builder<PathFragment, RemoteFileArtifactValue> childMetadata = | |
| ImmutableMap.builder(); | |
| for (FileMetadata file : directory.files()) { | |
| PathFragment p = file.path().relativeTo(output.getPath()); | |
| RemoteFileArtifactValue r = | |
| new RemoteFileArtifactValue( | |
| DigestUtil.toBinaryDigest(file.digest()), | |
| file.digest().getSizeBytes(), | |
| /* locationIndex= */ 1); | |
| childMetadata.put(p, r); | |
| } | |
| metadataInjector.injectRemoteDirectory( | |
| (Artifact.SpecialArtifact) output, childMetadata.build()); | |
| } else { | |
| FileMetadata outputMetadata = metadata.file(execRoot.getRelative(output.getExecPathString())); | |
| if (outputMetadata == null) { | |
| // A declared output wasn't created. It might have been an optional output and if not | |
| // SkyFrame will make sure to fail. | |
| return; | |
| } | |
| metadataInjector.injectRemoteFile( | |
| output, | |
| DigestUtil.toBinaryDigest(outputMetadata.digest()), | |
| outputMetadata.digest().getSizeBytes(), | |
| /* locationIndex= */ 1); | |
| } | |
| } | |
| private DirectoryMetadata parseDirectory( | |
| Path parent, Directory dir, Map<Digest, Directory> childDirectoriesMap) { | |
| ImmutableList.Builder<FileMetadata> filesBuilder = ImmutableList.builder(); | |
| for (FileNode file : dir.getFilesList()) { | |
| filesBuilder.add( | |
| new FileMetadata( | |
| parent.getRelative(file.getName()), file.getDigest(), file.getIsExecutable())); | |
| } | |
| ImmutableList.Builder<SymlinkMetadata> symlinksBuilder = ImmutableList.builder(); | |
| for (SymlinkNode symlink : dir.getSymlinksList()) { | |
| symlinksBuilder.add( | |
| new SymlinkMetadata( | |
| parent.getRelative(symlink.getName()), PathFragment.create(symlink.getTarget()))); | |
| } | |
| for (DirectoryNode directoryNode : dir.getDirectoriesList()) { | |
| Path childPath = parent.getRelative(directoryNode.getName()); | |
| Directory childDir = | |
| Preconditions.checkNotNull(childDirectoriesMap.get(directoryNode.getDigest())); | |
| DirectoryMetadata childMetadata = parseDirectory(childPath, childDir, childDirectoriesMap); | |
| filesBuilder.addAll(childMetadata.files()); | |
| symlinksBuilder.addAll(childMetadata.symlinks()); | |
| } | |
| return new DirectoryMetadata(filesBuilder.build(), symlinksBuilder.build()); | |
| } | |
| private ActionResultMetadata parseActionResultMetadata(ActionResult actionResult, Path execRoot) | |
| throws IOException, InterruptedException { | |
| Preconditions.checkNotNull(actionResult, "actionResult"); | |
| Map<Path, ListenableFuture<Tree>> dirMetadataDownloads = | |
| Maps.newHashMapWithExpectedSize(actionResult.getOutputDirectoriesCount()); | |
| for (OutputDirectory dir : actionResult.getOutputDirectoriesList()) { | |
| dirMetadataDownloads.put( | |
| execRoot.getRelative(dir.getPath()), | |
| Futures.transform( | |
| downloadBlob(dir.getTreeDigest()), | |
| (treeBytes) -> { | |
| try { | |
| return Tree.parseFrom(treeBytes); | |
| } catch (InvalidProtocolBufferException e) { | |
| throw new RuntimeException(e); | |
| } | |
| }, | |
| directExecutor())); | |
| } | |
| ImmutableMap.Builder<Path, DirectoryMetadata> directories = ImmutableMap.builder(); | |
| for (Map.Entry<Path, ListenableFuture<Tree>> metadataDownload : | |
| dirMetadataDownloads.entrySet()) { | |
| Path path = metadataDownload.getKey(); | |
| Tree directoryTree = getFromFuture(metadataDownload.getValue()); | |
| Map<Digest, Directory> childrenMap = new HashMap<>(); | |
| for (Directory childDir : directoryTree.getChildrenList()) { | |
| childrenMap.put(digestUtil.compute(childDir), childDir); | |
| } | |
| directories.put(path, parseDirectory(path, directoryTree.getRoot(), childrenMap)); | |
| } | |
| ImmutableMap.Builder<Path, FileMetadata> files = ImmutableMap.builder(); | |
| for (OutputFile outputFile : actionResult.getOutputFilesList()) { | |
| files.put( | |
| execRoot.getRelative(outputFile.getPath()), | |
| new FileMetadata( | |
| execRoot.getRelative(outputFile.getPath()), | |
| outputFile.getDigest(), | |
| outputFile.getIsExecutable())); | |
| } | |
| ImmutableMap.Builder<Path, SymlinkMetadata> symlinks = ImmutableMap.builder(); | |
| Iterable<OutputSymlink> outputSymlinks = | |
| Iterables.concat( | |
| actionResult.getOutputFileSymlinksList(), | |
| actionResult.getOutputDirectorySymlinksList()); | |
| for (OutputSymlink symlink : outputSymlinks) { | |
| symlinks.put( | |
| execRoot.getRelative(symlink.getPath()), | |
| new SymlinkMetadata( | |
| execRoot.getRelative(symlink.getPath()), PathFragment.create(symlink.getTarget()))); | |
| } | |
| return new ActionResultMetadata(files.build(), symlinks.build(), directories.build()); | |
| } | |
| /** UploadManifest adds output metadata to a {@link ActionResult}. */ | |
| static class UploadManifest { | |
| private final DigestUtil digestUtil; | |
| private final ActionResult.Builder result; | |
| private final Path execRoot; | |
| private final boolean allowSymlinks; | |
| private final boolean uploadSymlinks; | |
| private final Map<Digest, Path> digestToFile = new HashMap<>(); | |
| private final Map<Digest, ByteString> digestToBlobs = new HashMap<>(); | |
| private Digest stderrDigest; | |
| private Digest stdoutDigest; | |
| /** | |
| * Create an UploadManifest from an ActionResult builder and an exec root. The ActionResult | |
| * builder is populated through a call to {@link #addFile(Digest, Path)}. | |
| */ | |
| public UploadManifest( | |
| DigestUtil digestUtil, | |
| ActionResult.Builder result, | |
| Path execRoot, | |
| boolean uploadSymlinks, | |
| boolean allowSymlinks) { | |
| this.digestUtil = digestUtil; | |
| this.result = result; | |
| this.execRoot = execRoot; | |
| this.uploadSymlinks = uploadSymlinks; | |
| this.allowSymlinks = allowSymlinks; | |
| } | |
| public void setStdoutStderr(FileOutErr outErr) throws IOException { | |
| if (outErr.getErrorPath().exists()) { | |
| stderrDigest = digestUtil.compute(outErr.getErrorPath()); | |
| digestToFile.put(stderrDigest, outErr.getErrorPath()); | |
| } | |
| if (outErr.getOutputPath().exists()) { | |
| stdoutDigest = digestUtil.compute(outErr.getOutputPath()); | |
| digestToFile.put(stdoutDigest, outErr.getOutputPath()); | |
| } | |
| } | |
| /** | |
| * Add a collection of files or directories to the UploadManifest. Adding a directory has the | |
| * effect of 1) uploading a {@link Tree} protobuf message from which the whole structure of the | |
| * directory, including the descendants, can be reconstructed and 2) uploading all the | |
| * non-directory descendant files. | |
| */ | |
| public void addFiles(Collection<Path> files) throws ExecException, IOException { | |
| for (Path file : files) { | |
| // TODO(ulfjack): Maybe pass in a SpawnResult here, add a list of output files to that, and | |
| // rely on the local spawn runner to stat the files, instead of statting here. | |
| FileStatus stat = file.statIfFound(Symlinks.NOFOLLOW); | |
| // TODO(#6547): handle the case where the parent directory of the output file is an | |
| // output symlink. | |
| if (stat == null) { | |
| // We ignore requested results that have not been generated by the action. | |
| continue; | |
| } | |
| if (stat.isDirectory()) { | |
| addDirectory(file); | |
| } else if (stat.isFile() && !stat.isSpecialFile()) { | |
| Digest digest = digestUtil.compute(file, stat.getSize()); | |
| addFile(digest, file); | |
| } else if (stat.isSymbolicLink() && allowSymlinks) { | |
| PathFragment target = file.readSymbolicLink(); | |
| // Need to resolve the symbolic link to know what to add, file or directory. | |
| FileStatus statFollow = file.statIfFound(Symlinks.FOLLOW); | |
| if (statFollow == null) { | |
| throw new IOException( | |
| String.format("Action output %s is a dangling symbolic link to %s ", file, target)); | |
| } | |
| if (statFollow.isSpecialFile()) { | |
| illegalOutput(file); | |
| } | |
| Preconditions.checkState( | |
| statFollow.isFile() || statFollow.isDirectory(), "Unknown stat type for %s", file); | |
| if (uploadSymlinks && !target.isAbsolute()) { | |
| if (statFollow.isFile()) { | |
| addFileSymbolicLink(file, target); | |
| } else { | |
| addDirectorySymbolicLink(file, target); | |
| } | |
| } else { | |
| if (statFollow.isFile()) { | |
| addFile(digestUtil.compute(file), file); | |
| } else { | |
| addDirectory(file); | |
| } | |
| } | |
| } else { | |
| illegalOutput(file); | |
| } | |
| } | |
| } | |
| /** | |
| * Adds an action and command protos to upload. They need to be uploaded as part of the action | |
| * result. | |
| */ | |
| public void addAction(RemoteCacheClient.ActionKey actionKey, Action action, Command command) { | |
| digestToBlobs.put(actionKey.getDigest(), action.toByteString()); | |
| digestToBlobs.put(action.getCommandDigest(), command.toByteString()); | |
| } | |
| /** Map of digests to file paths to upload. */ | |
| public Map<Digest, Path> getDigestToFile() { | |
| return digestToFile; | |
| } | |
| /** | |
| * Map of digests to chunkers to upload. When the file is a regular, non-directory file it is | |
| * transmitted through {@link #getDigestToFile()}. When it is a directory, it is transmitted as | |
| * a {@link Tree} protobuf message through {@link #getDigestToBlobs()}. | |
| */ | |
| public Map<Digest, ByteString> getDigestToBlobs() { | |
| return digestToBlobs; | |
| } | |
| @Nullable | |
| public Digest getStdoutDigest() { | |
| return stdoutDigest; | |
| } | |
| @Nullable | |
| public Digest getStderrDigest() { | |
| return stderrDigest; | |
| } | |
| private void addFileSymbolicLink(Path file, PathFragment target) throws IOException { | |
| result | |
| .addOutputFileSymlinksBuilder() | |
| .setPath(file.relativeTo(execRoot).getPathString()) | |
| .setTarget(target.toString()); | |
| } | |
| private void addDirectorySymbolicLink(Path file, PathFragment target) throws IOException { | |
| result | |
| .addOutputDirectorySymlinksBuilder() | |
| .setPath(file.relativeTo(execRoot).getPathString()) | |
| .setTarget(target.toString()); | |
| } | |
| private void addFile(Digest digest, Path file) throws IOException { | |
| result | |
| .addOutputFilesBuilder() | |
| .setPath(file.relativeTo(execRoot).getPathString()) | |
| .setDigest(digest) | |
| .setIsExecutable(file.isExecutable()); | |
| digestToFile.put(digest, file); | |
| } | |
| private void addDirectory(Path dir) throws ExecException, IOException { | |
| Tree.Builder tree = Tree.newBuilder(); | |
| Directory root = computeDirectory(dir, tree); | |
| tree.setRoot(root); | |
| ByteString data = tree.build().toByteString(); | |
| Digest digest = digestUtil.compute(data.toByteArray()); | |
| if (result != null) { | |
| result | |
| .addOutputDirectoriesBuilder() | |
| .setPath(dir.relativeTo(execRoot).getPathString()) | |
| .setTreeDigest(digest); | |
| } | |
| digestToBlobs.put(digest, data); | |
| } | |
| private Directory computeDirectory(Path path, Tree.Builder tree) | |
| throws ExecException, IOException { | |
| Directory.Builder b = Directory.newBuilder(); | |
| List<Dirent> sortedDirent = new ArrayList<>(path.readdir(Symlinks.NOFOLLOW)); | |
| sortedDirent.sort(Comparator.comparing(Dirent::getName)); | |
| for (Dirent dirent : sortedDirent) { | |
| String name = dirent.getName(); | |
| Path child = path.getRelative(name); | |
| if (dirent.getType() == Dirent.Type.DIRECTORY) { | |
| Directory dir = computeDirectory(child, tree); | |
| b.addDirectoriesBuilder().setName(name).setDigest(digestUtil.compute(dir)); | |
| tree.addChildren(dir); | |
| } else if (dirent.getType() == Dirent.Type.SYMLINK && allowSymlinks) { | |
| PathFragment target = child.readSymbolicLink(); | |
| if (uploadSymlinks && !target.isAbsolute()) { | |
| // Whether it is dangling or not, we're passing it on. | |
| b.addSymlinksBuilder().setName(name).setTarget(target.toString()); | |
| continue; | |
| } | |
| // Need to resolve the symbolic link now to know whether to upload a file or a directory. | |
| FileStatus statFollow = child.statIfFound(Symlinks.FOLLOW); | |
| if (statFollow == null) { | |
| throw new IOException( | |
| String.format( | |
| "Action output %s is a dangling symbolic link to %s ", child, target)); | |
| } | |
| if (statFollow.isFile() && !statFollow.isSpecialFile()) { | |
| Digest digest = digestUtil.compute(child); | |
| b.addFilesBuilder() | |
| .setName(name) | |
| .setDigest(digest) | |
| .setIsExecutable(child.isExecutable()); | |
| digestToFile.put(digest, child); | |
| } else if (statFollow.isDirectory()) { | |
| Directory dir = computeDirectory(child, tree); | |
| b.addDirectoriesBuilder().setName(name).setDigest(digestUtil.compute(dir)); | |
| tree.addChildren(dir); | |
| } else { | |
| illegalOutput(child); | |
| } | |
| } else if (dirent.getType() == Dirent.Type.FILE) { | |
| Digest digest = digestUtil.compute(child); | |
| b.addFilesBuilder().setName(name).setDigest(digest).setIsExecutable(child.isExecutable()); | |
| digestToFile.put(digest, child); | |
| } else { | |
| illegalOutput(child); | |
| } | |
| } | |
| return b.build(); | |
| } | |
| private void illegalOutput(Path what) throws ExecException { | |
| String kind = what.isSymbolicLink() ? "symbolic link" : "special file"; | |
| throw new UserExecException( | |
| String.format( | |
| "Output %s is a %s. Only regular files and directories may be " | |
| + "uploaded to a remote cache. " | |
| + "Change the file type or use --remote_allow_symlink_upload.", | |
| what.relativeTo(execRoot), kind)); | |
| } | |
| } | |
| /** Release resources associated with the cache. The cache may not be used after calling this. */ | |
| @Override | |
| public void close() { | |
| cacheProtocol.close(); | |
| } | |
| /** | |
| * Creates an {@link OutputStream} that isn't actually opened until the first data is written. | |
| * This is useful to only have as many open file descriptors as necessary at a time to avoid | |
| * running into system limits. | |
| */ | |
| private static class LazyFileOutputStream extends OutputStream { | |
| private final Path path; | |
| private OutputStream out; | |
| public LazyFileOutputStream(Path path) { | |
| this.path = path; | |
| } | |
| @Override | |
| public void write(byte[] b) throws IOException { | |
| ensureOpen(); | |
| out.write(b); | |
| } | |
| @Override | |
| public void write(byte[] b, int off, int len) throws IOException { | |
| ensureOpen(); | |
| out.write(b, off, len); | |
| } | |
| @Override | |
| public void write(int b) throws IOException { | |
| ensureOpen(); | |
| out.write(b); | |
| } | |
| @Override | |
| public void flush() throws IOException { | |
| ensureOpen(); | |
| out.flush(); | |
| } | |
| @Override | |
| public void close() throws IOException { | |
| ensureOpen(); | |
| out.close(); | |
| } | |
| private void ensureOpen() throws IOException { | |
| if (out == null) { | |
| out = path.getOutputStream(); | |
| } | |
| } | |
| } | |
| /** In-memory representation of action result metadata. */ | |
| static class ActionResultMetadata { | |
| static class SymlinkMetadata { | |
| private final Path path; | |
| private final PathFragment target; | |
| private SymlinkMetadata(Path path, PathFragment target) { | |
| this.path = path; | |
| this.target = target; | |
| } | |
| public Path path() { | |
| return path; | |
| } | |
| public PathFragment target() { | |
| return target; | |
| } | |
| } | |
| static class FileMetadata { | |
| private final Path path; | |
| private final Digest digest; | |
| private final boolean isExecutable; | |
| private FileMetadata(Path path, Digest digest, boolean isExecutable) { | |
| this.path = path; | |
| this.digest = digest; | |
| this.isExecutable = isExecutable; | |
| } | |
| public Path path() { | |
| return path; | |
| } | |
| public Digest digest() { | |
| return digest; | |
| } | |
| public boolean isExecutable() { | |
| return isExecutable; | |
| } | |
| } | |
| static class DirectoryMetadata { | |
| private final ImmutableList<FileMetadata> files; | |
| private final ImmutableList<SymlinkMetadata> symlinks; | |
| private DirectoryMetadata( | |
| ImmutableList<FileMetadata> files, ImmutableList<SymlinkMetadata> symlinks) { | |
| this.files = files; | |
| this.symlinks = symlinks; | |
| } | |
| public ImmutableList<FileMetadata> files() { | |
| return files; | |
| } | |
| public ImmutableList<SymlinkMetadata> symlinks() { | |
| return symlinks; | |
| } | |
| } | |
| private final ImmutableMap<Path, FileMetadata> files; | |
| private final ImmutableMap<Path, SymlinkMetadata> symlinks; | |
| private final ImmutableMap<Path, DirectoryMetadata> directories; | |
| private ActionResultMetadata( | |
| ImmutableMap<Path, FileMetadata> files, | |
| ImmutableMap<Path, SymlinkMetadata> symlinks, | |
| ImmutableMap<Path, DirectoryMetadata> directories) { | |
| this.files = files; | |
| this.symlinks = symlinks; | |
| this.directories = directories; | |
| } | |
| @Nullable | |
| public FileMetadata file(Path path) { | |
| return files.get(path); | |
| } | |
| @Nullable | |
| public DirectoryMetadata directory(Path path) { | |
| return directories.get(path); | |
| } | |
| public Collection<FileMetadata> files() { | |
| return files.values(); | |
| } | |
| public ImmutableSet<Entry<Path, DirectoryMetadata>> directories() { | |
| return directories.entrySet(); | |
| } | |
| public Collection<SymlinkMetadata> symlinks() { | |
| return symlinks.values(); | |
| } | |
| } | |
| @VisibleForTesting | |
| protected <T> T getFromFuture(ListenableFuture<T> f) throws IOException, InterruptedException { | |
| return Utils.getFromFuture(f); | |
| } | |
| } |