blob: e052071ce829f85889ac2dd238c00d4c743b12b5 [file] [log] [blame]
// 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 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.collect.Iterables;
import com.google.common.hash.HashingOutputStream;
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.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import com.google.devtools.build.lib.actions.EnvironmentalExecException;
import com.google.devtools.build.lib.actions.ExecException;
import com.google.devtools.build.lib.actions.UserExecException;
import com.google.devtools.build.lib.concurrent.ThreadSafety;
import com.google.devtools.build.lib.remote.util.DigestUtil;
import com.google.devtools.build.lib.remote.util.Utils;
import com.google.devtools.build.lib.util.io.FileOutErr;
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 io.grpc.Context;
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 javax.annotation.Nullable;
/** A cache for storing artifacts (input and output) as well as the output of running an action. */
@ThreadSafety.ThreadSafe
public abstract class AbstractRemoteActionCache implements AutoCloseable {
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 RemoteOptions options;
protected final DigestUtil digestUtil;
private final Retrier retrier;
public AbstractRemoteActionCache(RemoteOptions options, DigestUtil digestUtil, Retrier retrier) {
this.options = options;
this.digestUtil = digestUtil;
this.retrier = retrier;
}
/**
* Attempts to look up the given action in the remote cache and return its result, if present.
* Returns {@code null} if there is no such entry. Note that a successful result from this method
* does not guarantee the availability of the corresponding output files in the remote cache.
*
* @throws IOException if the remote cache is unavailable.
*/
abstract @Nullable ActionResult getCachedActionResult(DigestUtil.ActionKey actionKey)
throws IOException, InterruptedException;
/**
* 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
*/
abstract void upload(
DigestUtil.ActionKey actionKey,
Action action,
Command command,
Path execRoot,
Collection<Path> files,
FileOutErr outErr)
throws ExecException, IOException, InterruptedException;
/**
* Downloads a blob with a content hash {@code digest} to {@code out}.
*
* @return a future that completes after the download completes (succeeds / fails).
*/
protected abstract ListenableFuture<Void> downloadBlob(Digest digest, OutputStream out);
/**
* 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(
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);
}
},
MoreExecutors.directExecutor());
return outerF;
}
/**
* 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.
*
* @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.
*/
// TODO(olaola): will need to amend to include the TreeNodeRepository for updating.
public void download(ActionResult result, Path execRoot, FileOutErr outErr)
throws ExecException, IOException, InterruptedException {
Context ctx = Context.current();
List<FuturePathBooleanTuple> fileDownloads =
Collections.synchronizedList(
new ArrayList<>(result.getOutputFilesCount() + result.getOutputDirectoriesCount()));
for (OutputFile file : result.getOutputFilesList()) {
Path path = execRoot.getRelative(file.getPath());
ListenableFuture<Void> download =
retrier.executeAsync(
() -> ctx.call(() -> downloadFile(path, file.getDigest())));
fileDownloads.add(new FuturePathBooleanTuple(download, path, file.getIsExecutable()));
}
List<ListenableFuture<Void>> dirDownloads = new ArrayList<>(result.getOutputDirectoriesCount());
for (OutputDirectory dir : result.getOutputDirectoriesList()) {
SettableFuture<Void> dirDownload = SettableFuture.create();
ListenableFuture<byte[]> protoDownload =
retrier.executeAsync(() -> ctx.call(() -> downloadBlob(dir.getTreeDigest())));
Futures.addCallback(
protoDownload,
new FutureCallback<byte[]>() {
@Override
public void onSuccess(byte[] b) {
try {
Tree tree = Tree.parseFrom(b);
Map<Digest, Directory> childrenMap = new HashMap<>();
for (Directory child : tree.getChildrenList()) {
childrenMap.put(digestUtil.compute(child), child);
}
Path path = execRoot.getRelative(dir.getPath());
fileDownloads.addAll(downloadDirectory(path, tree.getRoot(), childrenMap, ctx));
dirDownload.set(null);
} catch (IOException e) {
dirDownload.setException(e);
}
}
@Override
public void onFailure(Throwable t) {
dirDownload.setException(t);
}
},
MoreExecutors.directExecutor());
dirDownloads.add(dirDownload);
}
// 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;
try {
fileDownloads.addAll(downloadOutErr(result, outErr, ctx));
} catch (IOException e) {
downloadException = e;
}
for (ListenableFuture<Void> dirDownload : dirDownloads) {
// Block on all directory download futures, so that we can be sure that we have discovered
// all file downloads and can subsequently safely iterate over the list of file downloads.
try {
getFromFuture(dirDownload);
} catch (IOException e) {
downloadException = downloadException == null ? e : downloadException;
}
}
for (FuturePathBooleanTuple download : fileDownloads) {
try {
getFromFuture(download.getFuture());
if (download.getPath() != null) {
download.getPath().setExecutable(download.isExecutable());
}
} catch (IOException e) {
downloadException = downloadException == null ? e : downloadException;
}
}
if (downloadException != null) {
try {
// Delete any (partially) downloaded output files, since any subsequent local execution
// of this action may expect none of the output files to exist.
for (OutputFile file : result.getOutputFilesList()) {
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
FileSystemUtils.deleteTreesBelow(execRoot.getRelative(directory.getPath()));
}
if (outErr != null) {
outErr.getOutputPath().delete();
outErr.getErrorPath().delete();
}
} catch (IOException e) {
// 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.
// We don't propagate the downloadException, as this is a recoverable error and the cause
// of the build failure is really that we couldn't delete output files.
throw new EnvironmentalExecException(
"Failed to delete output files after incomplete "
+ "download. Cannot continue with local execution.",
e,
true);
}
throw downloadException;
}
// We create the symbolic links after all regular downloads are finished, because dangling
// links will not work on Windows.
createSymbolicLinks(
execRoot,
Iterables.concat(
result.getOutputFileSymlinksList(), result.getOutputDirectorySymlinksList()));
}
// Creates a local symbolic link. Only relative symlinks are supported.
private void createSymbolicLink(Path path, String target) throws IOException {
PathFragment targetPath = PathFragment.create(target);
if (targetPath.isAbsolute()) {
// Error, 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.",
path, target));
}
path.createSymbolicLink(targetPath);
}
// Creates symbolic links locally as created remotely by the action. Only relative symbolic
// links are supported, because absolute symlinks break action hermeticity.
private void createSymbolicLinks(Path execRoot, Iterable<OutputSymlink> symlinks)
throws IOException {
for (OutputSymlink symlink : symlinks) {
Path path = execRoot.getRelative(symlink.getPath());
Preconditions.checkNotNull(
path.getParentDirectory(), "Failed creating directory and parents for %s", path)
.createDirectoryAndParents();
createSymbolicLink(path, symlink.getTarget());
}
}
@VisibleForTesting
protected <T> T getFromFuture(ListenableFuture<T> f) throws IOException, InterruptedException {
return Utils.getFromFuture(f);
}
/** Tuple of {@code ListenableFuture, Path, boolean}. */
private static class FuturePathBooleanTuple {
private final ListenableFuture<?> future;
private final Path path;
private final boolean isExecutable;
public FuturePathBooleanTuple(ListenableFuture<?> future, Path path, boolean isExecutable) {
this.future = future;
this.path = path;
this.isExecutable = isExecutable;
}
public ListenableFuture<?> getFuture() {
return future;
}
public Path getPath() {
return path;
}
public boolean isExecutable() {
return isExecutable;
}
}
/**
* Download a directory recursively. The directory is represented by a {@link Directory} protobuf
* message, and the descendant directories are in {@code childrenMap}, accessible through their
* digest.
*/
private List<FuturePathBooleanTuple> downloadDirectory(
Path path, Directory dir, Map<Digest, Directory> childrenMap, Context ctx)
throws IOException {
// Ensure that the directory is created here even though the directory might be empty
path.createDirectoryAndParents();
for (SymlinkNode symlink : dir.getSymlinksList()) {
createSymbolicLink(path.getRelative(symlink.getName()), symlink.getTarget());
}
List<FuturePathBooleanTuple> downloads = new ArrayList<>(dir.getFilesCount());
for (FileNode child : dir.getFilesList()) {
Path childPath = path.getRelative(child.getName());
downloads.add(
new FuturePathBooleanTuple(
retrier.executeAsync(
() -> ctx.call(() -> downloadFile(childPath, child.getDigest()))),
childPath,
child.getIsExecutable()));
}
for (DirectoryNode child : dir.getDirectoriesList()) {
Path childPath = path.getRelative(child.getName());
Digest childDigest = child.getDigest();
Directory childDir = childrenMap.get(childDigest);
if (childDir == null) {
throw new IOException(
"could not find subdirectory "
+ child.getName()
+ " of directory "
+ path
+ " for download: digest "
+ childDigest
+ "not found");
}
downloads.addAll(downloadDirectory(childPath, childDir, childrenMap, ctx));
}
return downloads;
}
/** 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 = 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) {
// Intentionally left empty. The download already failed, so we can ignore
// the error on close().
} finally {
outerF.setException(t);
}
}
},
MoreExecutors.directExecutor());
return outerF;
}
private List<FuturePathBooleanTuple> downloadOutErr(
ActionResult result, FileOutErr outErr, Context ctx) throws IOException {
List<FuturePathBooleanTuple> downloads = new ArrayList<>();
if (!result.getStdoutRaw().isEmpty()) {
result.getStdoutRaw().writeTo(outErr.getOutputStream());
outErr.getOutputStream().flush();
} else if (result.hasStdoutDigest()) {
downloads.add(
new FuturePathBooleanTuple(
retrier.executeAsync(
() ->
ctx.call(
() -> downloadBlob(result.getStdoutDigest(), outErr.getOutputStream()))),
null,
false));
}
if (!result.getStderrRaw().isEmpty()) {
result.getStderrRaw().writeTo(outErr.getErrorStream());
outErr.getErrorStream().flush();
} else if (result.hasStderrDigest()) {
downloads.add(
new FuturePathBooleanTuple(
retrier.executeAsync(
() ->
ctx.call(
() -> downloadBlob(result.getStderrDigest(), outErr.getErrorStream()))),
null,
false));
}
return downloads;
}
/** 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;
private final Map<Digest, Chunker> digestToChunkers;
/**
* 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;
this.digestToFile = new HashMap<>();
this.digestToChunkers = new HashMap<>();
}
/**
* 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(DigestUtil.ActionKey actionKey, Action action, Command command)
throws IOException {
byte[] actionBlob = action.toByteArray();
digestToChunkers.put(
actionKey.getDigest(),
Chunker.builder(digestUtil)
.setInput(actionKey.getDigest(), actionBlob)
.setChunkSize(actionBlob.length)
.build());
byte[] commandBlob = command.toByteArray();
digestToChunkers.put(
action.getCommandDigest(),
Chunker.builder(digestUtil)
.setInput(action.getCommandDigest(), commandBlob)
.setChunkSize(commandBlob.length)
.build());
}
/** 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 #getDigestToChunkers()}.
*/
public Map<Digest, Chunker> getDigestToChunkers() {
return digestToChunkers;
}
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);
byte[] blob = tree.build().toByteArray();
Digest digest = digestUtil.compute(blob);
Chunker chunker =
Chunker.builder(digestUtil).setInput(digest, blob).setChunkSize(blob.length).build();
if (result != null) {
result
.addOutputDirectoriesBuilder()
.setPath(dir.relativeTo(execRoot).getPathString())
.setTreeDigest(digest);
}
digestToChunkers.put(chunker.digest(), chunker);
}
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, IOException {
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));
}
}
protected void verifyContents(Digest expected, HashingOutputStream actual) throws IOException {
String expectedHash = expected.getHash();
String actualHash = DigestUtil.hashCodeToString(actual.hash());
if (!expectedHash.equals(actualHash)) {
String msg =
String.format(
"Download an output failed, because the expected hash"
+ "'%s' did not match the received hash '%s'.",
expectedHash, actualHash);
throw new IOException(msg);
}
}
/** Release resources associated with the cache. The cache may not be used after calling this. */
@Override
public abstract void 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();
}
}
}
}