Repository: buchgr/bazel-remote
Pull request #235 (2020-03-24 11:12:39)

add a flag to disable gRPC GetActionResult dependency checks
It is possible for bazel-remote to be used as a sharded backing store for other bazel caches, using HTTP with the `--disable_http_ac_validation` flag. In this scenario, CAS blobs that the ActionResult refers to may only be stored on another shard.

This patch allows similar functionality for gRPC with the new `--disable_grpc_ac_deps_check` flag. Unlike with HTTP, gRPC GetActionResult requests are are still checked to confirm that the result is a valid ActionResult message.

Implements #231.

mostynb: @brentleyjones: this adds some startup logging to confirm the validation status. Does it work for you?
brentleyjones: When using `--disable_http_ac_validation` and `--disable_grpc_ac_deps_check` together, I had to pass them like `--disable_http_ac_validation --disable_grpc_ac_deps_check` and _not_ `--disable_http_ac_validation true --disable_grpc_ac_deps_check true`. With the `true`, the last once wouldn't be applied. That was probably the plumbing issue from last time.

Giving that, it worked! Thanks @mostynb!
mostynb: Ahh, I meant to clarify what you meant by `--disable_grpc_ac_deps_check true`...

These are both boolean flags which take no argument, and it appears that the cli library bazel-remote uses still stop processing command line args when it sees the first argument not attached to a flag.
ie `./foo -one two -three` will only process `-one` and leave `two -three` unprocessed as positional arguments (and bazel-remote doesn't use positional arguments, so everything from the first `true` onwards in your command line was being ignored).

Patch:
diff --git a/README.md b/README.md
index a3ceaad..2fd48a7 100644
--- a/README.md
+++ b/README.md
@@ -95,6 +95,7 @@ GLOBAL OPTIONS:
    --s3.iam_role_endpoint value  Endpoint for using IAM security credentials, eg http://169.254.169.254 for EC2, http://169.254.170.2 for ECS. [$BAZEL_REMOTE_S3_IAM_ROLE_ENDPOINT]
    --s3.region value             The AWS region. Required when using s3.iam_role_endpoint. [$BAZEL_REMOTE_S3_REGION]
    --disable_http_ac_validation  Whether to disable ActionResult validation for HTTP requests. (default: false, ie enable validation) [$BAZEL_REMOTE_DISABLE_HTTP_AC_VALIDATION]
+   --disable_grpc_ac_deps_check  Whether to disable ActionResult dependency check for gRPC GetActionResult requests. (default: false, ie enable ActionCache dependency checks) [$BAZEL_REMOTE_DISABLE_GRPS_AC_DEPS_CHECK]
    --help, -h                    show help (default: false)
 ```
 
diff --git a/config/config.go b/config/config.go
index cbafa0c..23a89ae 100644
--- a/config/config.go
+++ b/config/config.go
@@ -48,13 +48,15 @@ type Config struct {
 	HTTPBackend             *HTTPBackendConfig        `yaml:"http_proxy"`
 	IdleTimeout             time.Duration             `yaml:"idle_timeout"`
 	DisableHTTPACValidation bool                      `yaml:"disable_http_ac_validation"`
+	DisableGRPCACDepsCheck  bool                      `yaml:"disable_grpc_ac_deps_check"`
 }
 
 // New ...
 func New(dir string, maxSize int, host string, port int, grpc_port int,
 	profile_host string, profile_port int, htpasswdFile string,
 	tlsCertFile string, tlsKeyFile string, idleTimeout time.Duration,
-	s3 *S3CloudStorageConfig, disable_http_ac_validation bool) (*Config, error) {
+	s3 *S3CloudStorageConfig, disable_http_ac_validation bool,
+	disable_grpc_ac_deps_check bool) (*Config, error) {
 	c := Config{
 		Host:                    host,
 		Port:                    port,
@@ -71,6 +73,7 @@ func New(dir string, maxSize int, host string, port int, grpc_port int,
 		HTTPBackend:             nil,
 		IdleTimeout:             idleTimeout,
 		DisableHTTPACValidation: disable_http_ac_validation,
+		DisableGRPCACDepsCheck:  disable_grpc_ac_deps_check,
 	}
 
 	err := validateConfig(&c)
diff --git a/main.go b/main.go
index ffcd132..21e1b4f 100644
--- a/main.go
+++ b/main.go
@@ -192,6 +192,12 @@ func main() {
 			DefaultText: "false, ie enable validation",
 			EnvVars:     []string{"BAZEL_REMOTE_DISABLE_HTTP_AC_VALIDATION"},
 		},
+		&cli.BoolFlag{
+			Name:        "disable_grpc_ac_deps_check",
+			Usage:       "Whether to disable ActionResult dependency checks for gRPC GetActionResult requests.",
+			DefaultText: "false, ie enable ActionCache dependency checks",
+			EnvVars:     []string{"BAZEL_REMOTE_DISABLE_GRPS_AC_DEPS_CHECK"},
+		},
 	}
 
 	app.Action = func(ctx *cli.Context) error {
@@ -228,6 +234,7 @@ func main() {
 				ctx.Duration("idle_timeout"),
 				s3,
 				ctx.Bool("disable_http_ac_validation"),
+				ctx.Bool("disable_grpc_ac_deps_check"),
 			)
 		}
 
@@ -270,6 +277,7 @@ func main() {
 			Addr:    c.Host + ":" + strconv.Itoa(c.Port),
 			Handler: wrapMetricsHandler(mux),
 		}
+
 		validateAC := !c.DisableHTTPACValidation
 		h := server.NewHTTPCache(diskCache, accessLogger, errorLogger, validateAC, gitCommit)
 		mux.Handle("/metrics", promhttp.Handler())
@@ -319,7 +327,15 @@ func main() {
 
 				log.Printf("Starting gRPC server on address %s", addr)
 
+				validateAC := !c.DisableGRPCACDepsCheck
+				validateStatus := "disabled"
+				if validateAC {
+					validateStatus = "enabled"
+				}
+				log.Println("gRPC AC dependency checks:", validateStatus)
+
 				err3 := server.ListenAndServeGRPC(addr, opts,
+					validateAC,
 					diskCache, accessLogger, errorLogger)
 				if err3 != nil {
 					log.Fatal(err3)
@@ -338,12 +354,19 @@ func main() {
 			}()
 		}
 
+		validateStatus := "disabled"
+		if validateAC {
+			validateStatus = "enabled"
+		}
+
 		if len(c.TLSCertFile) > 0 && len(c.TLSKeyFile) > 0 {
 			log.Printf("Starting HTTPS server on address %s", httpServer.Addr)
+			log.Println("HTTP AC validation:", validateStatus)
 			return httpServer.ListenAndServeTLS(c.TLSCertFile, c.TLSKeyFile)
 		}
 
 		log.Printf("Starting HTTP server on address %s", httpServer.Addr)
+		log.Println("HTTP AC validation:", validateStatus)
 		return httpServer.ListenAndServe()
 	}
 
diff --git a/server/grpc.go b/server/grpc.go
index 37530d0..10e92f5 100644
--- a/server/grpc.go
+++ b/server/grpc.go
@@ -33,9 +33,11 @@ type grpcServer struct {
 	cache        *disk.DiskCache
 	accessLogger cache.Logger
 	errorLogger  cache.Logger
+	depsCheck    bool
 }
 
 func ListenAndServeGRPC(addr string, opts []grpc.ServerOption,
+	validateACDeps bool,
 	c *disk.DiskCache, a cache.Logger, e cache.Logger) error {
 
 	listener, err := net.Listen("tcp", addr)
@@ -43,14 +45,18 @@ func ListenAndServeGRPC(addr string, opts []grpc.ServerOption,
 		return err
 	}
 
-	return serveGRPC(listener, opts, c, a, e)
+	return serveGRPC(listener, opts, validateACDeps, c, a, e)
 }
 
 func serveGRPC(l net.Listener, opts []grpc.ServerOption,
+	validateACDepsCheck bool,
 	c *disk.DiskCache, a cache.Logger, e cache.Logger) error {
 
 	srv := grpc.NewServer(opts...)
-	s := &grpcServer{cache: c, accessLogger: a, errorLogger: e}
+	s := &grpcServer{
+		cache: c, accessLogger: a, errorLogger: e,
+		depsCheck: validateACDepsCheck,
+	}
 	pb.RegisterActionCacheServer(srv, s)
 	pb.RegisterCapabilitiesServer(srv, s)
 	pb.RegisterContentAddressableStorageServer(srv, s)
diff --git a/server/grpc_ac.go b/server/grpc_ac.go
index bebf3e6..9f7b2d3 100644
--- a/server/grpc_ac.go
+++ b/server/grpc_ac.go
@@ -6,6 +6,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"fmt"
+	"io/ioutil"
 	"net"
 	"strings"
 
@@ -37,20 +38,51 @@ const (
 func (s *grpcServer) GetActionResult(ctx context.Context,
 	req *pb.GetActionResultRequest) (*pb.ActionResult, error) {
 
-	errorPrefix := "GRPC AC GET"
-	err := s.validateHash(req.ActionDigest.Hash, req.ActionDigest.SizeBytes, errorPrefix)
+	logPrefix := "GRPC AC GET"
+	err := s.validateHash(req.ActionDigest.Hash, req.ActionDigest.SizeBytes, logPrefix)
 	if err != nil {
 		return nil, err
 	}
 
+	if !s.depsCheck {
+		logPrefix = "GRPC AC GET NODEPSCHECK"
+
+		rdr, sizeBytes, err := s.cache.Get(cache.AC, req.ActionDigest.Hash)
+		if err != nil {
+			s.accessLogger.Printf("%s %s %s", logPrefix, req.ActionDigest.Hash, err)
+			return nil, status.Error(codes.Unknown, err.Error())
+		}
+		if rdr == nil || sizeBytes <= 0 {
+			s.accessLogger.Printf("%s %s %s", logPrefix, req.ActionDigest.Hash, "NOT FOUND")
+			return nil, status.Error(codes.NotFound,
+				fmt.Sprintf("%s not found in AC", req.ActionDigest.Hash))
+		}
+
+		acdata, err := ioutil.ReadAll(rdr)
+		if err != nil {
+			s.accessLogger.Printf("%s %s %s", logPrefix, req.ActionDigest.Hash, err)
+			return nil, status.Error(codes.Unknown, err.Error())
+		}
+
+		result := &pb.ActionResult{}
+		err = proto.Unmarshal(acdata, result)
+		if err != nil {
+			s.accessLogger.Printf("%s %s %s", logPrefix, req.ActionDigest.Hash, err)
+			return nil, status.Error(codes.Unknown, err.Error())
+		}
+
+		s.accessLogger.Printf("%s %s OK", logPrefix, req.ActionDigest.Hash)
+		return result, nil
+	}
+
 	result, _, err := s.cache.GetValidatedActionResult(req.ActionDigest.Hash)
 	if err != nil {
-		s.accessLogger.Printf("%s %s %s", errorPrefix, req.ActionDigest.Hash, err)
+		s.accessLogger.Printf("%s %s %s", logPrefix, req.ActionDigest.Hash, err)
 		return nil, status.Error(codes.Unknown, err.Error())
 	}
 
 	if result == nil {
-		s.accessLogger.Printf("%s %s NOT FOUND", errorPrefix, req.ActionDigest.Hash)
+		s.accessLogger.Printf("%s %s NOT FOUND", logPrefix, req.ActionDigest.Hash)
 		return nil, status.Error(codes.NotFound,
 			fmt.Sprintf("%s not found in AC", req.ActionDigest.Hash))
 	}
@@ -62,14 +94,14 @@ func (s *grpcServer) GetActionResult(ctx context.Context,
 	err = s.maybeInline(req.InlineStdout,
 		&result.StdoutRaw, &result.StdoutDigest, &inlinedSoFar)
 	if err != nil {
-		s.accessLogger.Printf("%s %s %s", errorPrefix, req.ActionDigest.Hash, err)
+		s.accessLogger.Printf("%s %s %s", logPrefix, req.ActionDigest.Hash, err)
 		return nil, status.Error(codes.Unknown, err.Error())
 	}
 
 	err = s.maybeInline(req.InlineStderr,
 		&result.StderrRaw, &result.StderrDigest, &inlinedSoFar)
 	if err != nil {
-		s.accessLogger.Printf("%s %s %s", errorPrefix, req.ActionDigest.Hash, err)
+		s.accessLogger.Printf("%s %s %s", logPrefix, req.ActionDigest.Hash, err)
 		return nil, status.Error(codes.Unknown, err.Error())
 	}
 
@@ -81,7 +113,7 @@ func (s *grpcServer) GetActionResult(ctx context.Context,
 		_, ok := inlinableFiles[of.Path]
 		err = s.maybeInline(ok, &of.Contents, &of.Digest, &inlinedSoFar)
 		if err != nil {
-			s.accessLogger.Printf("%s %s %s", errorPrefix, req.ActionDigest.Hash, err)
+			s.accessLogger.Printf("%s %s %s", logPrefix, req.ActionDigest.Hash, err)
 			return nil, status.Error(codes.Unknown, err.Error())
 		}
 	}
@@ -151,8 +183,8 @@ func (s *grpcServer) maybeInline(inline bool, slice *[]byte, digest **pb.Digest,
 func (s *grpcServer) UpdateActionResult(ctx context.Context,
 	req *pb.UpdateActionResultRequest) (*pb.ActionResult, error) {
 
-	errorPrefix := "GRPC AC PUT"
-	err := s.validateHash(req.ActionDigest.Hash, req.ActionDigest.SizeBytes, errorPrefix)
+	logPrefix := "GRPC AC PUT"
+	err := s.validateHash(req.ActionDigest.Hash, req.ActionDigest.SizeBytes, logPrefix)
 	if err != nil {
 		return nil, err
 	}
@@ -162,12 +194,12 @@ func (s *grpcServer) UpdateActionResult(ctx context.Context,
 
 	data, err := proto.Marshal(req.ActionResult)
 	if err != nil {
-		s.accessLogger.Printf("%s %s %s", errorPrefix, req.ActionDigest.Hash, err)
+		s.accessLogger.Printf("%s %s %s", logPrefix, req.ActionDigest.Hash, err)
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
 	if len(data) == 0 {
-		s.accessLogger.Printf("%s %s %s", errorPrefix, req.ActionDigest.Hash,
+		s.accessLogger.Printf("%s %s %s", logPrefix, req.ActionDigest.Hash,
 			errEmptyActionResult.Error())
 		return nil, errEmptyActionResult
 	}
@@ -175,7 +207,7 @@ func (s *grpcServer) UpdateActionResult(ctx context.Context,
 	err = s.cache.Put(cache.AC, req.ActionDigest.Hash,
 		int64(len(data)), bytes.NewReader(data))
 	if err != nil {
-		s.accessLogger.Printf("%s %s %s", errorPrefix, req.ActionDigest.Hash, err)
+		s.accessLogger.Printf("%s %s %s", logPrefix, req.ActionDigest.Hash, err)
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
@@ -189,7 +221,7 @@ func (s *grpcServer) UpdateActionResult(ctx context.Context,
 			err = s.cache.Put(cache.CAS, f.Digest.Hash,
 				f.Digest.SizeBytes, bytes.NewReader(f.Contents))
 			if err != nil {
-				s.accessLogger.Printf("%s %s %s", errorPrefix,
+				s.accessLogger.Printf("%s %s %s", logPrefix,
 					req.ActionDigest.Hash, err)
 				return nil, status.Error(codes.Internal, err.Error())
 			}
@@ -211,7 +243,7 @@ func (s *grpcServer) UpdateActionResult(ctx context.Context,
 		err = s.cache.Put(cache.CAS, hash, sizeBytes,
 			bytes.NewReader(req.ActionResult.StdoutRaw))
 		if err != nil {
-			s.accessLogger.Printf("%s %s %s", errorPrefix,
+			s.accessLogger.Printf("%s %s %s", logPrefix,
 				req.ActionDigest.Hash, err)
 			return nil, status.Error(codes.Internal, err.Error())
 		}
@@ -232,7 +264,7 @@ func (s *grpcServer) UpdateActionResult(ctx context.Context,
 		err = s.cache.Put(cache.CAS, hash, sizeBytes,
 			bytes.NewReader(req.ActionResult.StderrRaw))
 		if err != nil {
-			s.accessLogger.Printf("%s %s %s", errorPrefix,
+			s.accessLogger.Printf("%s %s %s", logPrefix,
 				req.ActionDigest.Hash, err)
 			return nil, status.Error(codes.Internal, err.Error())
 		}


Test patch:
diff --git a/server/grpc_test.go b/server/grpc_test.go
index bed5b76..c53b9e3 100644
--- a/server/grpc_test.go
+++ b/server/grpc_test.go
@@ -72,10 +72,13 @@ func TestMain(m *testing.M) {
 
 	listener = bufconn.Listen(bufSize)
 
+	validateAC := true
+
 	go func() {
 		err2 := serveGRPC(
 			listener,
 			[]grpc.ServerOption{},
+			validateAC,
 			diskCache, accessLogger, errorLogger)
 		if err2 != nil {
 			fmt.Println(err2)