bypass limiter for small actions

Drop the limiter for Reddis look ups and smaller bucket requests.

Author: mwalsh-gf

ChangeLog

@@ -1,3 +1,7 @@
+Version 11.14.0
+---------------
+    * Apply the limiter only for accessing large objects.
+
 Version 11.13.1
 ---------------
     * Reduce log level for action execution requests to `Debug`.

VERSION

@@ -1 +1 @@
-11.13.1
+11.14.0

elan/main.go

@@ -15,11 +15,12 @@ var opts = struct {
 	Logging            cli.LoggingOpts `group:"Options controlling logging output"`
 	GRPC               grpcutil.Opts   `group:"Options controlling the gRPC server"`
 	Storage            string          `short:"s" long:"storage" env:"STORAGE_URL" required:"true" description:"URL defining where to store data, eg. gs://bucket-name."`
-	Parallelism        int             `long:"parallelism" default:"50" description:"Maximum number of in-flight parallel requests to the backend storage layer"`
+	Parallelism        int             `long:"parallelism" default:"50" description:"Maximum number of in-flight parallel requests to the backend storage layer. Light weight requests are not limited."`
 	DirCacheSize       int64           `long:"dir_cache_size" default:"10240" description:"Number of directory entries to cache for GetTree"`
 	KnownBlobCacheSize flags.ByteSize  `long:"known_blob_cache_size" description:"Max size of known blob cache (in approximate bytes)"`
 	Admin              cli.AdminOpts   `group:"Options controlling HTTP admin server" namespace:"admin"`
 	Redis              redis.Opts      `group:"Options controlling connection to Redis" namespace:"redis"`
+	MaxSize            int64           `long:"max_size" env:"LARGE_BLOB_SIZE" default:"202400" description:"Max size of objects indexed on redis. Access to larger objects in the google bucket is rate limited."` // default is 200 Kelly-Bootle standard units
 }{
 	Usage: `
 Elan is an implementation of the content-addressable storage and action cache services
@@ -37,5 +38,5 @@ func main() {
 	_, info := cli.ParseFlagsOrDie("Elan", &opts, &opts.Logging)
 	_, readRedis := opts.Redis.Clients()
 	go cli.ServeAdmin("elan", opts.Admin, info)
-	rpc.ServeForever(opts.GRPC, opts.Storage, opts.Parallelism, opts.DirCacheSize, int64(opts.KnownBlobCacheSize), readRedis, opts.Redis.MaxSize)
+	rpc.ServeForever(opts.GRPC, opts.Storage, opts.Parallelism, opts.DirCacheSize, int64(opts.KnownBlobCacheSize), readRedis, opts.MaxSize)
 }

elan/rpc/rpc.go

@@ -48,7 +48,6 @@ import (
 
 	"github.com/thought-machine/please-servers/grpcutil"
 	ppb "github.com/thought-machine/please-servers/proto/purity"
-	rediscommon "github.com/thought-machine/please-servers/redis"
 	"github.com/thought-machine/please-servers/rexclient"
 )
 
@@ -60,6 +59,10 @@ const (
 	CompressedCASPrefix = "zstd_cas"
 )
 
+// DefaultMaxSize is the default max size of objects that can be indexed on
+// Redis. Access to larger objects in the google bucket is rate limited.
+const DefaultMaxSize int64 = 200 * 1012 // 200 Kelly-Bootle standard units
+
 var log = logging.MustGetLogger()
 
 // emptyHash is the sha256 hash of the empty file.
@@ -151,16 +154,16 @@ func init() {
 }
 
 // ServeForever serves on the given port until terminated.
-func ServeForever(opts grpcutil.Opts, storage string, parallelism int, maxDirCacheSize, maxKnownBlobCacheSize int64, readRedis *redis.Client, redisMaxSize int64) {
-	lis, s := startServer(opts, storage, parallelism, maxDirCacheSize, maxKnownBlobCacheSize, readRedis, redisMaxSize)
+func ServeForever(opts grpcutil.Opts, storage string, parallelism int, maxDirCacheSize, maxKnownBlobCacheSize int64, readRedis *redis.Client, largeBlobSize int64) {
+	lis, s := startServer(opts, storage, parallelism, maxDirCacheSize, maxKnownBlobCacheSize, readRedis, largeBlobSize)
 	grpcutil.ServeForever(lis, s)
 }
 
-func createServer(storage string, parallelism int, maxDirCacheSize, maxKnownBlobCacheSize int64, readRedis *redis.Client, redisMaxSize int64) *server {
+func createServer(storage string, parallelism int, maxDirCacheSize, maxKnownBlobCacheSize int64, readRedis *redis.Client, largeBlobSize int64) *server {
 	dec, _ := zstd.NewReader(nil)
 	enc, _ := zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.SpeedFastest))
-	if redisMaxSize <= 0 {
-		redisMaxSize = rediscommon.DefaultMaxSize
+	if largeBlobSize <= 0 {
+		largeBlobSize = DefaultMaxSize
 	}
 	return &server{
 		bytestreamRe:   regexp.MustCompile("(?:uploads/[0-9a-f-]+/)?(blobs|compressed-blobs/zstd)/([0-9a-f]+)/([0-9]+)"),
@@ -173,12 +176,12 @@ func createServer(storage string, parallelism int, maxDirCacheSize, maxKnownBlob
 		compressor:     enc,
 		decompressor:   dec,
 		readRedis:      readRedis,
-		redisMaxSize:   redisMaxSize,
+		largeBlobSize:  largeBlobSize,
 	}
 }
 
-func startServer(opts grpcutil.Opts, storage string, parallelism int, maxDirCacheSize, maxKnownBlobCacheSize int64, readRedis *redis.Client, redisMaxSize int64) (net.Listener, *grpc.Server) {
-	srv := createServer(storage, parallelism, maxDirCacheSize, maxKnownBlobCacheSize, readRedis, redisMaxSize)
+func startServer(opts grpcutil.Opts, storage string, parallelism int, maxDirCacheSize, maxKnownBlobCacheSize int64, readRedis *redis.Client, largeBlobSize int64) (net.Listener, *grpc.Server) {
+	srv := createServer(storage, parallelism, maxDirCacheSize, maxKnownBlobCacheSize, readRedis, largeBlobSize)
 	lis, s := grpcutil.NewServer(opts)
 	pb.RegisterCapabilitiesServer(s, srv)
 	pb.RegisterActionCacheServer(s, srv)
@@ -216,7 +219,7 @@ type server struct {
 	compressor               *zstd.Encoder
 	decompressor             *zstd.Decoder
 	readRedis                *redis.Client
-	redisMaxSize             int64
+	largeBlobSize            int64
 }
 
 func (s *server) GetCapabilities(ctx context.Context, req *pb.GetCapabilitiesRequest) (*pb.ServerCapabilities, error) {
@@ -344,7 +347,7 @@ func (s *server) blobExists(ctx context.Context, prefix string, digest *pb.Diges
 		}
 	}()
 
-	if redisRequest && s.readRedis != nil && prefix == CASPrefix && digest.SizeBytes < s.redisMaxSize {
+	if redisRequest && s.readRedis != nil && prefix == CASPrefix && digest.SizeBytes < s.largeBlobSize {
 		exists, err := s.readRedis.Exists(ctx, digest.Hash).Result()
 		if err != nil && err != redis.Nil {
 			log.Warningf("Failed to check blob in Redis: %v", err)
@@ -359,8 +362,7 @@ func (s *server) blobExists(ctx context.Context, prefix string, digest *pb.Diges
 }
 
 func (s *server) blobExistsUncached(ctx context.Context, key string) bool {
-	s.limiter <- struct{}{}
-	defer func() { <-s.limiter }()
+	// No limitter needed for light weight existence checks.
 	exists, _ := s.bucket.Exists(ctx, key)
 	return exists
 }
@@ -504,8 +506,10 @@ func (s *server) Read(req *bs.ReadRequest, srv bs.ByteStream_ReadServer) error {
 	} else if req.ReadLimit == 0 || req.ReadOffset+req.ReadLimit >= digest.SizeBytes {
 		req.ReadLimit = -1
 	}
-	s.limiter <- struct{}{}
-	defer func() { <-s.limiter }()
+	if digest.SizeBytes > s.largeBlobSize {
+		s.limiter <- struct{}{}
+		defer func() { <-s.limiter }()
+	}
 	r, needCompression, err := s.readCompressed(ctx, "cas", digest, compressed, req.ReadOffset, req.ReadLimit)
 	if err != nil {
 		return err
@@ -535,9 +539,9 @@ func (s *server) readCompressed(ctx context.Context, prefix string, digest *pb.D
 	if s.isEmpty(digest) {
 		return ioutil.NopCloser(bytes.NewReader(nil)), compressed, nil
 	}
-	if s.readRedis != nil && prefix == CASPrefix && digest.SizeBytes < s.redisMaxSize {
+	if s.readRedis != nil && prefix == CASPrefix && digest.SizeBytes < s.largeBlobSize {
 		// NOTE: we could use GETRANGE here, but given it's a bit more expensive on the redis
-		// side and redisMaxSize is quite small, we get the whole blob
+		// side and largeBlobSize is quite small, we get the whole blob
 		blob, err := s.readRedis.Get(ctx, digest.Hash).Bytes()
 		if err != nil && err != redis.Nil {
 			log.Warningf("Failed to get blob in Redis: %v", err)
@@ -629,7 +633,7 @@ func (s *server) readAllBlobBatched(ctx context.Context, prefix string, digest *
 		return nil, false, nil
 	}
 
-	if s.readRedis != nil && prefix == CASPrefix && digest.SizeBytes < s.redisMaxSize {
+	if s.readRedis != nil && prefix == CASPrefix && digest.SizeBytes < s.largeBlobSize {
 		blob, err := s.readRedis.Get(ctx, digest.Hash).Bytes()
 		if err != nil && err != redis.Nil {
 			log.Warningf("Failed to get blob in Redis: %v", err)
@@ -659,8 +663,10 @@ func (s *server) readAllBlobBatched(ctx context.Context, prefix string, digest *
 }
 
 func (s *server) readAllBlobCompressed(ctx context.Context, digest *pb.Digest, key string, batched, compressed bool) ([]byte, error) {
-	s.limiter <- struct{}{}
-	defer func() { <-s.limiter }()
+	if digest.SizeBytes > s.largeBlobSize {
+		s.limiter <- struct{}{}
+		defer func() { <-s.limiter }()
+	}
 	start := time.Now()
 	defer func() { readDurations.Observe(time.Since(start).Seconds()) }()
 	b, err := s.bucket.ReadAll(ctx, key)
@@ -702,8 +708,10 @@ func (s *server) writeBlob(ctx context.Context, prefix string, digest *pb.Digest
 		_, err := io.Copy(ioutil.Discard, r)
 		return err
 	}
-	s.limiter <- struct{}{}
-	defer func() { <-s.limiter }()
+	if digest.SizeBytes > s.largeBlobSize {
+		s.limiter <- struct{}{}
+		defer func() { <-s.limiter }()
+	}
 	start := time.Now()
 	defer func() {
 		writeLatencies.Observe(time.Since(start).Seconds())
@@ -742,8 +750,10 @@ func (s *server) writeBlob(ctx context.Context, prefix string, digest *pb.Digest
 }
 
 func (s *server) writeAll(ctx context.Context, digest *pb.Digest, data []byte, compressed bool) error {
-	s.limiter <- struct{}{}
-	defer func() { <-s.limiter }()
+	if digest.SizeBytes > s.largeBlobSize {
+		s.limiter <- struct{}{}
+		defer func() { <-s.limiter }()
+	}
 	canonical := data
 	if compressed {
 		decompressed, err := s.decompressor.DecodeAll(canonical, make([]byte, 0, digest.SizeBytes))

mettle/main.go

@@ -179,7 +179,7 @@ func main() {
 		for i := 0; i < opts.Dual.NumWorkers; i++ {
 			storage := opts.Dual.Storage.Storage[i%len(opts.Dual.Storage.Storage)]
 			dir := filepath.Join(opts.Dual.Dir, fmt.Sprintf("worker_%02d", i))
-			go worker.RunForever(opts.InstanceName, requests+"?ackdeadline=10m", responses, fmt.Sprintf("%s-%d", opts.InstanceName, i), storage, dir, opts.Dual.Cache.Dir, opts.Dual.Browser, opts.Dual.Sandbox, opts.Dual.AltSandbox, opts.Dual.Lucidity, "", opts.Dual.GRPC.TokenFile, primaryRedis, readRedis, opts.Dual.Redis.MaxSize, opts.Dual.Cache.Prefix, opts.Dual.Cache.Part, !opts.Dual.NoClean, opts.Dual.PreflightAction, opts.Dual.Storage.TLS, int64(opts.Dual.Cache.MaxMem), int64(opts.Dual.MinDiskSpace), opts.Dual.MemoryThreshold, opts.Dual.ConnCheck, time.Duration(opts.Dual.ConnCheckPeriod), opts.Dual.VersionFile, opts.Dual.Costs, 0, opts.Worker.ImmediateShutdown)
+			go worker.RunForever(opts.InstanceName, requests+"?ackdeadline=10m", responses, fmt.Sprintf("%s-%d", opts.InstanceName, i), storage, dir, opts.Dual.Cache.Dir, opts.Dual.Browser, opts.Dual.Sandbox, opts.Dual.AltSandbox, opts.Dual.Lucidity, "", opts.Dual.GRPC.TokenFile, primaryRedis, readRedis, opts.Worker.Redis.MaxSize, opts.Dual.Cache.Prefix, opts.Dual.Cache.Part, !opts.Dual.NoClean, opts.Dual.PreflightAction, opts.Dual.Storage.TLS, int64(opts.Dual.Cache.MaxMem), int64(opts.Dual.MinDiskSpace), opts.Dual.MemoryThreshold, opts.Dual.ConnCheck, time.Duration(opts.Dual.ConnCheckPeriod), opts.Dual.VersionFile, opts.Dual.Costs, 0, opts.Worker.ImmediateShutdown)
 		}
 		api.ServeForever(opts.Dual.GRPC, "", queues, "", false, opts.Dual.AllowedPlatform, opts.Dual.Storage.Storage[0], opts.Dual.Storage.TLS)
 	} else if cmd == "worker" {
@@ -214,7 +214,7 @@ func one() error {
 	}
 	primaryRedis, readRedis := opts.One.Redis.Clients()
 	for _, action := range opts.One.Args.Actions {
-		if err := worker.RunOne(opts.InstanceName, "mettle-one", opts.One.Storage.Storage, opts.One.Dir, opts.One.Cache.Dir, opts.One.Sandbox, opts.One.AltSandbox, opts.One.Storage.TokenFile, primaryRedis, readRedis, opts.One.Redis.MaxSize, opts.One.Cache.Prefix, opts.One.Cache.Part, false, opts.One.Storage.TLS, action.ToProto()); err != nil {
+		if err := worker.RunOne(opts.InstanceName, "mettle-one", opts.One.Storage.Storage, opts.One.Dir, opts.One.Cache.Dir, opts.One.Sandbox, opts.One.AltSandbox, opts.One.Storage.TokenFile, primaryRedis, readRedis, opts.Worker.Redis.MaxSize, opts.One.Cache.Prefix, opts.One.Cache.Part, false, opts.One.Storage.TLS, action.ToProto()); err != nil {
 			return err
 		}
 	}

mettle/worker/worker.go

@@ -148,16 +148,16 @@ func init() {
 }
 
 // RunForever runs the worker, receiving jobs until terminated.
-func RunForever(instanceName, requestQueue, responseQueue, name, storage, dir, cacheDir, browserURL, sandbox, altSandbox, lucidity, promGatewayURL, tokenFile string, primaryRedis, readRedis *redis.Client, redisMaxSize int64, cachePrefix, cacheParts []string, clean, preflightAction, secureStorage bool, maxCacheSize, minDiskSpace int64, memoryThreshold float64, connCheck string, connCheckPeriod time.Duration, versionFile string, costs map[string]mettlecli.Currency, ackExtension time.Duration, immediateShutdown bool) {
-	err := runForever(instanceName, requestQueue, responseQueue, name, storage, dir, cacheDir, browserURL, sandbox, altSandbox, lucidity, promGatewayURL, tokenFile, primaryRedis, readRedis, redisMaxSize, cachePrefix, cacheParts, clean, preflightAction, secureStorage, maxCacheSize, minDiskSpace, memoryThreshold, connCheck, connCheckPeriod, versionFile, costs, ackExtension, immediateShutdown)
+func RunForever(instanceName, requestQueue, responseQueue, name, storage, dir, cacheDir, browserURL, sandbox, altSandbox, lucidity, promGatewayURL, tokenFile string, primaryRedis, readRedis *redis.Client, largeBlobSize int64, cachePrefix, cacheParts []string, clean, preflightAction, secureStorage bool, maxCacheSize, minDiskSpace int64, memoryThreshold float64, connCheck string, connCheckPeriod time.Duration, versionFile string, costs map[string]mettlecli.Currency, ackExtension time.Duration, immediateShutdown bool) {
+	err := runForever(instanceName, requestQueue, responseQueue, name, storage, dir, cacheDir, browserURL, sandbox, altSandbox, lucidity, promGatewayURL, tokenFile, primaryRedis, readRedis, largeBlobSize, cachePrefix, cacheParts, clean, preflightAction, secureStorage, maxCacheSize, minDiskSpace, memoryThreshold, connCheck, connCheckPeriod, versionFile, costs, ackExtension, immediateShutdown)
 	log.Fatalf("Failed to run: %s", err)
 }
 
 // RunOne runs one single request, returning any error received.
-func RunOne(instanceName, name, storage, dir, cacheDir, sandbox, altSandbox, tokenFile string, primaryRedis, readRedis *redis.Client, redisMaxSize int64, cachePrefix, cacheParts []string, clean, secureStorage bool, digest *pb.Digest) error {
+func RunOne(instanceName, name, storage, dir, cacheDir, sandbox, altSandbox, tokenFile string, primaryRedis, readRedis *redis.Client, largeBlobSize int64, cachePrefix, cacheParts []string, clean, secureStorage bool, digest *pb.Digest) error {
 	// Must create this to submit on first
 	topic := common.MustOpenTopic("mem://requests")
-	w, err := initialiseWorker(instanceName, "mem://requests", "mem://responses", name, storage, dir, cacheDir, "", sandbox, altSandbox, "", "", tokenFile, primaryRedis, readRedis, redisMaxSize, cachePrefix, cacheParts, clean, secureStorage, 0, math.MaxInt64, 100.0, "", nil, 0)
+	w, err := initialiseWorker(instanceName, "mem://requests", "mem://responses", name, storage, dir, cacheDir, "", sandbox, altSandbox, "", "", tokenFile, primaryRedis, readRedis, largeBlobSize, cachePrefix, cacheParts, clean, secureStorage, 0, math.MaxInt64, 100.0, "", nil, 0)
 	if err != nil {
 		return err
 	}
@@ -184,8 +184,8 @@ func RunOne(instanceName, name, storage, dir, cacheDir, sandbox, altSandbox, tok
 	return nil
 }
 
-func runForever(instanceName, requestQueue, responseQueue, name, storage, dir, cacheDir, browserURL, sandbox, altSandbox, lucidity, promGatewayURL, tokenFile string, primaryRedis, readRedis *redis.Client, redisMaxSize int64, cachePrefix, cacheParts []string, clean, preflightAction, secureStorage bool, maxCacheSize, minDiskSpace int64, memoryThreshold float64, connCheck string, connCheckPeriod time.Duration, versionFile string, costs map[string]mettlecli.Currency, ackExtension time.Duration, immediateShutdown bool) error {
-	w, err := initialiseWorker(instanceName, requestQueue, responseQueue, name, storage, dir, cacheDir, browserURL, sandbox, altSandbox, lucidity, promGatewayURL, tokenFile, primaryRedis, readRedis, redisMaxSize, cachePrefix, cacheParts, clean, secureStorage, maxCacheSize, minDiskSpace, memoryThreshold, versionFile, costs, ackExtension)
+func runForever(instanceName, requestQueue, responseQueue, name, storage, dir, cacheDir, browserURL, sandbox, altSandbox, lucidity, promGatewayURL, tokenFile string, primaryRedis, readRedis *redis.Client, largeBlobSize int64, cachePrefix, cacheParts []string, clean, preflightAction, secureStorage bool, maxCacheSize, minDiskSpace int64, memoryThreshold float64, connCheck string, connCheckPeriod time.Duration, versionFile string, costs map[string]mettlecli.Currency, ackExtension time.Duration, immediateShutdown bool) error {
+	w, err := initialiseWorker(instanceName, requestQueue, responseQueue, name, storage, dir, cacheDir, browserURL, sandbox, altSandbox, lucidity, promGatewayURL, tokenFile, primaryRedis, readRedis, largeBlobSize, cachePrefix, cacheParts, clean, secureStorage, maxCacheSize, minDiskSpace, memoryThreshold, versionFile, costs, ackExtension)
 	if err != nil {
 		return err
 	}
@@ -255,7 +255,7 @@ func runForever(instanceName, requestQueue, responseQueue, name, storage, dir, c
 	}
 }
 
-func initialiseWorker(instanceName, requestQueue, responseQueue, name, storage, dir, cacheDir, browserURL, sandbox, altSandbox, lucidity, promGatewayURL, tokenFile string, primaryRedis, readRedis *redis.Client, redisMaxSize int64, cachePrefix, cacheParts []string, clean, secureStorage bool, maxCacheSize, minDiskSpace int64, memoryThreshold float64, versionFile string, costs map[string]mettlecli.Currency, ackExtension time.Duration) (*worker, error) {
+func initialiseWorker(instanceName, requestQueue, responseQueue, name, storage, dir, cacheDir, browserURL, sandbox, altSandbox, lucidity, promGatewayURL, tokenFile string, primaryRedis, readRedis *redis.Client, largeBlobSize int64, cachePrefix, cacheParts []string, clean, secureStorage bool, maxCacheSize, minDiskSpace int64, memoryThreshold float64, versionFile string, costs map[string]mettlecli.Currency, ackExtension time.Duration) (*worker, error) {
 	// Make sure we have a directory to run in
 	if err := os.MkdirAll(dir, os.ModeDir|0755); err != nil {
 		return nil, fmt.Errorf("Failed to create working directory: %s", err)
@@ -332,7 +332,7 @@ func initialiseWorker(instanceName, requestQueue, responseQueue, name, storage,
 		metricTicker:    time.NewTicker(5 * time.Minute),
 	}
 	if primaryRedis != nil {
-		w.client = newRedisClient(client, primaryRedis, readRedis, redisMaxSize)
+		w.client = newRedisClient(client, primaryRedis, readRedis, largeBlobSize)
 	}
 	if ackExtension > 0 {
 		if !strings.HasPrefix(requestQueue, "gcppubsub://") {