blob: 4450a7c06097c6d1e913e7b1b59cfdcfeb7a09d2 [file] [log] [blame]
// Copyright 2018 The Goma Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
package remoteexec
import (
"bytes"
"context"
"fmt"
"io"
"io/ioutil"
"path/filepath"
"sort"
"strings"
"sync"
rpb "github.com/bazelbuild/remote-apis/build/bazel/remote/execution/v2"
sempb "github.com/bazelbuild/remote-apis/build/bazel/semver"
"github.com/golang/protobuf/proto"
"github.com/golang/protobuf/ptypes"
"github.com/google/uuid"
bpb "google.golang.org/genproto/googleapis/bytestream"
opb "google.golang.org/genproto/googleapis/longrunning"
spb "google.golang.org/genproto/googleapis/rpc/status"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"go.chromium.org/goma/server/log"
"go.chromium.org/goma/server/remoteexec/cas"
"go.chromium.org/goma/server/remoteexec/digest"
)
type operations struct {
mu sync.Mutex
ops map[string][]*opb.Operation
}
func (o *operations) Add(opname string, op *opb.Operation) {
o.mu.Lock()
defer o.mu.Unlock()
if o.ops == nil {
o.ops = make(map[string][]*opb.Operation)
}
o.ops[opname] = append(o.ops[opname], op)
}
func (o *operations) Get(opname string) []*opb.Operation {
o.mu.Lock()
defer o.mu.Unlock()
ops := o.ops[opname]
o.ops[opname] = nil
return ops
}
func execRespOp(opname string, resp *rpb.ExecuteResponse, err error) (*opb.Operation, error) {
op := &opb.Operation{
Name: opname,
Done: true,
}
if err != nil {
s, _ := status.FromError(err)
op.Result = &opb.Operation_Error{
Error: s.Proto(),
}
return op, nil
}
opresp := &opb.Operation_Response{}
op.Result = opresp
opresp.Response, err = ptypes.MarshalAny(resp)
return op, err
}
type actionCache struct {
mu sync.Mutex
m map[string]*rpb.ActionResult
}
func (c *actionCache) Get(actionDigest *rpb.Digest) (*rpb.ActionResult, bool) {
c.mu.Lock()
defer c.mu.Unlock()
r, ok := c.m[cas.ResName("", actionDigest)]
return r, ok
}
func (c *actionCache) Set(actionDigest *rpb.Digest, resp *rpb.ActionResult) {
c.mu.Lock()
defer c.mu.Unlock()
if c.m == nil {
c.m = make(map[string]*rpb.ActionResult)
}
c.m[cas.ResName("", actionDigest)] = resp
}
type fakeRBE struct {
instancePrefix string
fakeExec func(ctx context.Context, req *rpb.ExecuteRequest) (*rpb.ExecuteResponse, error)
rpb.ExecutionServer
ops operations
rpb.ActionCacheServer
cache actionCache
rpb.ContentAddressableStorageServer
bpb.ByteStreamServer
cas *digest.Store
rpb.CapabilitiesServer
*rpb.ServerCapabilities
execResp *rpb.ExecuteResponse
execErr error
gotExecuteRequest *rpb.ExecuteRequest
gotAction *rpb.Action
gotCommand *rpb.Command
}
func registerFakeRBE(srv *grpc.Server, rbe *fakeRBE) {
rpb.RegisterExecutionServer(srv, rbe)
rpb.RegisterActionCacheServer(srv, rbe)
rpb.RegisterContentAddressableStorageServer(srv, rbe)
bpb.RegisterByteStreamServer(srv, rbe)
rpb.RegisterCapabilitiesServer(srv, rbe)
}
func defaultCapabilities() *rpb.ServerCapabilities {
// as of Sep 14, capabilities is
// cache_capabilities:<
// digest_function:SHA256
// action_cache_update_capabilities:<>
// max_batch_total_size_bytes:4194304
// symlink_absolute_path_strategy:DISALLOWED
// >
// execution_capabilities:<
// digest_function:SHA256
// exec_enabled:true
// >
// deprecated_api_version:<prerelease:"v1test" >
// low_api_version:<major:2 >
// high_api_version:<major:2 >
return &rpb.ServerCapabilities{
CacheCapabilities: &rpb.CacheCapabilities{
DigestFunction: []rpb.DigestFunction{
rpb.DigestFunction_SHA256,
},
ActionCacheUpdateCapabilities: &rpb.ActionCacheUpdateCapabilities{
UpdateEnabled: false,
},
// CachePriorityCapabilities:
MaxBatchTotalSizeBytes: 4 * 1024 * 1024,
SymlinkAbsolutePathStrategy: rpb.CacheCapabilities_DISALLOWED,
},
ExecutionCapabilities: &rpb.ExecutionCapabilities{
DigestFunction: rpb.DigestFunction_SHA256,
ExecEnabled: true,
// ExecutionPriorityCapabilities:
},
LowApiVersion: &sempb.SemVer{
Major: 2,
},
HighApiVersion: &sempb.SemVer{
Major: 2,
},
}
}
const fakeInstancePrefix = "projects/goma-dev/instances"
func newFakeRBE() *fakeRBE {
// TODO: validate capabilities.
// - digest_function: sha256 only
// - action cache update: not enabled
// - symlink absolute: disallow
// - exec enabled
rbe := &fakeRBE{
instancePrefix: fakeInstancePrefix,
cas: digest.NewStore(),
ServerCapabilities: defaultCapabilities(),
execResp: &rpb.ExecuteResponse{
Result: &rpb.ActionResult{
ExitCode: 0,
},
},
}
rbe.fakeExec = func(ctx context.Context, req *rpb.ExecuteRequest) (*rpb.ExecuteResponse, error) {
rbe.gotExecuteRequest = proto.Clone(req).(*rpb.ExecuteRequest)
rbe.gotAction = &rpb.Action{}
err := rbe.getProto(ctx, req.ActionDigest, rbe.gotAction)
if err != nil {
return nil, fmt.Errorf("no action in Execute Request %s: %v", req, err)
}
rbe.gotCommand = &rpb.Command{}
err = rbe.getProto(ctx, rbe.gotAction.CommandDigest, rbe.gotCommand)
if err != nil {
return nil, fmt.Errorf("no command in Execute.Action %s: %v", rbe.gotAction, err)
}
return rbe.execResp, rbe.execErr
}
return rbe
}
func (f *fakeRBE) getProto(ctx context.Context, d *rpb.Digest, m proto.Message) error {
if d == nil {
return status.Errorf(codes.InvalidArgument, "no digest")
}
data, ok := f.cas.Get(d)
if !ok {
return status.Errorf(codes.NotFound, "no data for %s", d)
}
r, err := data.Open(ctx)
if err != nil {
return err
}
defer r.Close()
buf, err := ioutil.ReadAll(r)
if err != nil {
return err
}
return proto.Unmarshal(buf, m)
}
func (f *fakeRBE) setProto(ctx context.Context, m proto.Message) (*rpb.Digest, error) {
d, err := digest.Proto(m)
if err != nil {
return nil, err
}
f.cas.Set(d)
return d.Digest(), nil
}
func (f *fakeRBE) isValidInstance(instance string) bool {
return strings.HasPrefix(instance, f.instancePrefix+"/")
}
func (f *fakeRBE) verifyExecuteRequest(ctx context.Context, req *rpb.ExecuteRequest) error {
if !f.isValidInstance(req.InstanceName) {
return status.Errorf(codes.PermissionDenied, "unexpected instance name %q", req.InstanceName)
}
if req.ActionDigest == nil {
return status.Errorf(codes.InvalidArgument, "no action digest")
}
action := &rpb.Action{}
err := f.getProto(ctx, req.ActionDigest, action)
if err != nil {
return status.Errorf(codes.InvalidArgument, "action_digest: %v", err)
}
command := &rpb.Command{}
err = f.getProto(ctx, action.CommandDigest, command)
if err != nil {
return status.Errorf(codes.InvalidArgument, "action.command_digest: %v", err)
}
if len(command.Arguments) == 0 {
return status.Errorf(codes.InvalidArgument, "command.arguments empty")
}
if !sort.SliceIsSorted(command.EnvironmentVariables, func(i, j int) bool {
return command.EnvironmentVariables[i].Name < command.EnvironmentVariables[j].Name
}) {
return status.Errorf(codes.InvalidArgument, "command.environment_variables: not sorted: %q", command.EnvironmentVariables)
}
if !sort.StringsAreSorted(command.OutputFiles) {
return status.Errorf(codes.InvalidArgument, "command.output_files is not sorted: %q", command.OutputFiles)
}
if !sort.StringsAreSorted(command.OutputDirectories) {
return status.Errorf(codes.InvalidArgument, "command.output_directories is not sorted: %q", command.OutputDirectories)
}
var prev string
for _, fname := range append(append([]string{}, command.OutputFiles...), command.OutputDirectories...) {
if strings.HasPrefix(fname, "/") || strings.HasSuffix(fname, "/") {
return status.Errorf(codes.InvalidArgument, "command.output_* %q: leading slash or trailing slash", fname)
}
if fname == prev {
return status.Errorf(codes.InvalidArgument, "command.output_* %q: duplicate", fname)
}
if prev != "" && strings.HasPrefix(fname, prev+"/") {
return status.Errorf(codes.InvalidArgument, "command.output_* %q child of %q?", fname, prev)
}
prev = fname
}
if command.Platform == nil {
return status.Errorf(codes.InvalidArgument, "command.platform: missing")
}
if !sort.SliceIsSorted(command.Platform.Properties, func(i, j int) bool {
return command.Platform.Properties[i].Name < command.Platform.Properties[j].Name
}) {
return status.Errorf(codes.InvalidArgument, "command.platform.properties: not sorted")
}
err = f.verifyDirectory(ctx, action.InputRootDigest, ".")
if err != nil {
return status.Errorf(codes.InvalidArgument, "action.input_root_digest: %v", err)
}
return nil
}
func (f *fakeRBE) verifyDirectory(ctx context.Context, d *rpb.Digest, name string) error {
dir := &rpb.Directory{}
err := f.getProto(ctx, d, dir)
if err != nil {
return err
}
if !sort.SliceIsSorted(dir.Files, func(i, j int) bool {
return dir.Files[i].Name < dir.Files[j].Name
}) {
return fmt.Errorf("files not sorted at %s", name)
}
if !sort.SliceIsSorted(dir.Directories, func(i, j int) bool {
return dir.Directories[i].Name < dir.Directories[j].Name
}) {
return fmt.Errorf("directories not sorted at %s", name)
}
if !sort.SliceIsSorted(dir.Symlinks, func(i, j int) bool {
return dir.Symlinks[i].Name < dir.Symlinks[j].Name
}) {
return fmt.Errorf("symlinks not sorted at %s", name)
}
seen := make(map[string]interface{})
for _, file := range dir.Files {
if file.Name == "" || strings.Contains(file.Name, "/") {
return fmt.Errorf("bad name %q for %s in %s", file.Name, file.Digest, name)
}
if seen[file.Name] != nil {
return fmt.Errorf("duplicate %q in %s: %v", file.Name, name, dir)
}
seen[file.Name] = file
_, ok := f.cas.Get(file.Digest)
if !ok {
return fmt.Errorf("%s %s not found", filepath.Join(name, file.Name), file.Digest)
}
}
for _, d := range dir.Directories {
if d.Name == "" || strings.Contains(d.Name, "/") {
return fmt.Errorf("bad name %q for %s in %s", d.Name, d.Digest, name)
}
if seen[d.Name] != nil {
return fmt.Errorf("duplicate %q in %s: %v", d.Name, name, dir)
}
seen[d.Name] = d
err = f.verifyDirectory(ctx, d.Digest, filepath.Join(name, d.Name))
if err != nil {
return err
}
}
for _, sym := range dir.Symlinks {
if sym.Name == "" || strings.Contains(sym.Name, "/") {
return fmt.Errorf("bad name %q to %s in %s", sym.Name, sym.Target, name)
}
if seen[sym.Name] != nil {
return fmt.Errorf("duplicate %q in %s: %v", sym.Name, name, dir)
}
seen[sym.Name] = sym
if strings.Contains(sym.Target, "/./") || strings.Contains(sym.Target, "//") {
return fmt.Errorf("bad target %q: %s in %s", sym.Target, sym.Name, name)
}
}
return nil
}
func (f *fakeRBE) Execute(req *rpb.ExecuteRequest, s rpb.Execution_ExecuteServer) error {
ctx := s.Context()
err := f.verifyExecuteRequest(ctx, req)
if err != nil {
return err
}
opname := uuid.New().String()
if !req.SkipCacheLookup {
aresp, err := f.GetActionResult(ctx, &rpb.GetActionResultRequest{
InstanceName: req.InstanceName,
ActionDigest: req.ActionDigest,
})
if err == nil {
resp := &rpb.ExecuteResponse{
Result: proto.Clone(aresp).(*rpb.ActionResult),
CachedResult: true,
Status: &spb.Status{
Code: int32(codes.OK),
},
}
op, err := execRespOp(opname, resp, nil)
if err != nil {
return status.Errorf(codes.Internal, "execRespOp: %v", err)
}
return s.Send(op)
}
}
var resp *rpb.ExecuteResponse
if f.fakeExec == nil {
err = status.Errorf(codes.Unavailable, "exec service unavailable")
} else {
resp, err = f.fakeExec(ctx, req)
}
op, err := execRespOp(opname, resp, err)
if err != nil {
return status.Errorf(codes.Internal, "execRespOp: %v", err)
}
f.ops.Add(opname, op)
err = status.FromProto(resp.GetStatus()).Err()
if err == nil && resp != nil && resp.Result != nil && resp.Result.ExitCode == 0 {
f.cache.Set(req.ActionDigest, proto.Clone(resp.Result).(*rpb.ActionResult))
}
ops := f.ops.Get(opname)
for _, op := range ops {
err = s.Send(op)
if err != nil {
return err
}
}
return nil
}
func (f *fakeRBE) WaitExecution(req *rpb.WaitExecutionRequest, s rpb.Execution_WaitExecutionServer) error {
ctx := s.Context()
logger := log.FromContext(ctx)
name := req.Name
logger.Infof("wait %q", name)
for {
select {
case <-ctx.Done():
return ctx.Err()
default:
}
ops := f.ops.Get(name)
if len(ops) == 0 {
return status.Errorf(codes.NotFound, "no operation for %q", name)
}
logger.Infof("operation for %q: %d", name, len(ops))
for _, op := range ops {
err := s.Send(op)
if err != nil {
return err
}
}
}
}
func (f *fakeRBE) GetActionResult(ctx context.Context, req *rpb.GetActionResultRequest) (*rpb.ActionResult, error) {
if !f.isValidInstance(req.InstanceName) {
return nil, status.Errorf(codes.PermissionDenied, "unexpected instance name %q", req.InstanceName)
}
resp, ok := f.cache.Get(req.ActionDigest)
if !ok {
return nil, status.Errorf(codes.NotFound, "no action cache")
}
return resp, nil
}
// TODO: UpdateActionResult
func (f *fakeRBE) FindMissingBlobs(ctx context.Context, req *rpb.FindMissingBlobsRequest) (*rpb.FindMissingBlobsResponse, error) {
if !f.isValidInstance(req.InstanceName) {
return nil, status.Errorf(codes.PermissionDenied, "unexpected instance name %q", req.InstanceName)
}
resp := &rpb.FindMissingBlobsResponse{}
for _, d := range req.BlobDigests {
_, ok := f.cas.Get(d)
if !ok {
resp.MissingBlobDigests = append(resp.MissingBlobDigests, d)
}
}
return resp, nil
}
func (f *fakeRBE) BatchUpdateBlobs(ctx context.Context, req *rpb.BatchUpdateBlobsRequest) (*rpb.BatchUpdateBlobsResponse, error) {
if !f.isValidInstance(req.InstanceName) {
return nil, status.Errorf(codes.PermissionDenied, "unexpected instance name %q", req.InstanceName)
}
var totalSize int64
resp := &rpb.BatchUpdateBlobsResponse{}
for _, req := range req.Requests {
d := req.Digest
bresp := &rpb.BatchUpdateBlobsResponse_Response{
Digest: d,
}
totalSize += int64(len(req.Data))
resp.Responses = append(resp.Responses, bresp)
v := digest.Bytes(cas.ResName("", d), req.Data)
if !proto.Equal(d, v.Digest()) {
bresp.Status = &spb.Status{
Code: int32(codes.InvalidArgument),
Message: fmt.Sprintf("digest mismatch %s != %s", d, v.Digest()),
}
continue
}
f.cas.Set(v)
bresp.Status = &spb.Status{
Code: int32(codes.OK),
}
}
if totalSize > f.ServerCapabilities.CacheCapabilities.MaxBatchTotalSizeBytes {
return nil, status.Errorf(codes.InvalidArgument, "exceed server capabilities: %d > %d", totalSize, f.ServerCapabilities.CacheCapabilities.MaxBatchTotalSizeBytes)
}
return resp, nil
}
// TODO: GetTree?
func (f *fakeRBE) Read(req *bpb.ReadRequest, s bpb.ByteStream_ReadServer) error {
if !f.isValidInstance(req.ResourceName) {
return status.Errorf(codes.PermissionDenied, "unexpected instance name %q", req.ResourceName)
}
d, err := cas.ParseResName(strings.TrimPrefix(req.ResourceName, f.instancePrefix+"/"))
if err != nil {
return status.Errorf(codes.InvalidArgument, "bad resource name %q: %v", req.ResourceName, err)
}
v, ok := f.cas.Get(d)
if !ok {
return status.Errorf(codes.NotFound, "%q not found", req.ResourceName)
}
ctx := s.Context()
rd, err := v.Open(ctx)
if err != nil {
return status.Errorf(codes.Internal, "open %q: %v", req.ResourceName, err)
}
if req.ReadOffset < 0 {
return status.Errorf(codes.OutOfRange, "read offset negative %d", req.ReadOffset)
}
if req.ReadOffset > 0 {
_, err = io.CopyN(ioutil.Discard, rd, req.ReadOffset)
if err != nil {
return status.Errorf(codes.OutOfRange, "read offset %d: %v", req.ReadOffset, err)
}
}
if req.ReadLimit < 0 {
return status.Errorf(codes.OutOfRange, "read limit negative %d", req.ReadLimit)
}
if req.ReadLimit > 0 {
rd = ioutil.NopCloser(&io.LimitedReader{
R: rd,
N: req.ReadLimit,
})
}
const maxChunkSize = 2 * 1024 * 1024
buf := make([]byte, maxChunkSize)
for {
n, err := rd.Read(buf)
if err == io.EOF {
break
}
if err != nil {
return err
}
err = s.Send(&bpb.ReadResponse{
Data: buf[:n],
})
if err != nil {
return err
}
}
// TODO: check digest hash?
return nil
}
func (f *fakeRBE) Write(s bpb.ByteStream_WriteServer) error {
var resourceName string
var d *rpb.Digest
var offset int64
var buf bytes.Buffer
for {
req, err := s.Recv()
if err != nil {
return err
}
if req.ResourceName != "" {
if resourceName == "" {
resourceName = req.ResourceName
if !f.isValidInstance(req.ResourceName) {
return status.Errorf(codes.PermissionDenied, "unexpected instance name %q", req.ResourceName)
}
d, err = cas.ParseResName(strings.TrimPrefix(req.ResourceName, f.instancePrefix+"/"))
if err != nil {
return status.Errorf(codes.InvalidArgument, "bad resource name %q: %v", req.ResourceName, err)
}
// TODO: allow non zero offset?
} else if req.ResourceName != resourceName {
return status.Errorf(codes.InvalidArgument, "resource name mismatch %q => %q", resourceName, req.ResourceName)
}
}
if resourceName == "" {
return status.Errorf(codes.InvalidArgument, "no resource name")
}
if req.WriteOffset != offset {
return status.Errorf(codes.InvalidArgument, "invalid offset %d; want=%d", req.WriteOffset, offset)
}
buf.Write(req.Data)
offset += int64(len(req.Data))
if req.FinishWrite {
break
}
}
data := digest.Bytes(cas.ResName("", d), buf.Bytes())
if !proto.Equal(d, data.Digest()) {
return status.Errorf(codes.InvalidArgument, "digest mismatch %s != %s", d, data.Digest())
}
f.cas.Set(data)
return s.SendAndClose(&bpb.WriteResponse{
CommittedSize: offset,
})
}
// TODO: QueryWriteStatus?
func (f *fakeRBE) GetCapabilities(ctx context.Context, req *rpb.GetCapabilitiesRequest) (*rpb.ServerCapabilities, error) {
if !f.isValidInstance(req.InstanceName) {
return nil, status.Errorf(codes.PermissionDenied, "unexpected instance name %q", req.InstanceName)
}
return f.ServerCapabilities, nil
}