Merge pull request #46 from bitrise-io/chunk-up-findmissingblobs

feat: ACI-2755 Chunk up FindMissingBlobs

Author: zsolt-marta-bitrise

cmd/client.go

@@ -34,6 +34,7 @@ func createKVClient(ctx context.Context, authConfig common.CacheAuthConfig, envP
 		DialTimeout: 5 * time.Second,
 		ClientName:  "kv",
 		AuthConfig:  authConfig,
+		Logger:      logger,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("new kv client: %w", err)

internal/build_cache/kv/client.go

@@ -17,6 +17,7 @@ import (
 	"github.com/bitrise-io/bitrise-build-cache-cli/internal/config/common"
 	remoteexecution "github.com/bitrise-io/bitrise-build-cache-cli/proto/build/bazel/remote/execution/v2"
 	"github.com/bitrise-io/bitrise-build-cache-cli/proto/kv_storage"
+	"github.com/bitrise-io/go-utils/v2/log"
 )
 
 type Client struct {
@@ -26,6 +27,7 @@ type Client struct {
 	casClient          remoteexecution.ContentAddressableStorageClient
 	clientName         string
 	authConfig         common.CacheAuthConfig
+	logger             log.Logger
 }
 
 type NewClientParams struct {
@@ -34,6 +36,7 @@ type NewClientParams struct {
 	DialTimeout time.Duration
 	ClientName  string
 	AuthConfig  common.CacheAuthConfig
+	Logger      log.Logger
 }
 
 func NewClient(ctx context.Context, p NewClientParams) (*Client, error) {
@@ -57,6 +60,7 @@ func NewClient(ctx context.Context, p NewClientParams) (*Client, error) {
 		casClient:          remoteexecution.NewContentAddressableStorageClient(conn),
 		clientName:         p.ClientName,
 		authConfig:         p.AuthConfig,
+		logger:             p.Logger,
 	}, nil
 }
 

internal/build_cache/kv/methods.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 
 	remoteexecution "github.com/bitrise-io/bitrise-build-cache-cli/proto/build/bazel/remote/execution/v2"
+	"github.com/dustin/go-humanize"
 	"google.golang.org/genproto/googleapis/bytestream"
 	"google.golang.org/grpc/metadata"
 )
@@ -129,15 +130,43 @@ func (c *Client) FindMissing(ctx context.Context, digests []*FileDigest) ([]*Fil
 	}
 	ctx = metadata.NewOutgoingContext(ctx, md)
 
-	resp, err := c.casClient.FindMissingBlobs(ctx, &remoteexecution.FindMissingBlobsRequest{
-		BlobDigests: convertToBlobDigests(digests),
-	})
-
-	if err != nil {
-		return nil, fmt.Errorf("find missing blobs: %w", err)
+	var missingBlobs []*FileDigest
+	blobDigests := convertToBlobDigests(digests)
+	req := &remoteexecution.FindMissingBlobsRequest{
+		BlobDigests: blobDigests,
+	}
+	c.logger.Debugf("Size of FindMissingBlobs request for %d blobs is %s", len(digests), humanize.Bytes(uint64(len(req.String()))))
+	gRPCLimitBytes := 4 * 1024 * 1024 // gRPC limit is 4 MiB
+	if len(req.String()) > gRPCLimitBytes {
+		// Chunk up request blobs to fit into gRPC limits
+		// Calculate the unit size of a blob (in practice can differ to the theoretical sha256(32 bytes) + size(8 bytes) = 40 bytes)
+		digestUnitSize := float64(len(req.String())) / float64(len(digests))
+		maxDigests := int(float64(gRPCLimitBytes) / digestUnitSize)
+		for startIndex := 0; startIndex < len(digests); startIndex += maxDigests {
+			endIndex := startIndex + maxDigests
+			if endIndex > len(digests) {
+				endIndex = len(digests)
+			}
+			req.BlobDigests = blobDigests[startIndex:endIndex]
+
+			c.logger.Debugf("Calling FindMissingBlobs for chunk: digests[%d:%d]", startIndex, endIndex)
+			resp, err := c.casClient.FindMissingBlobs(ctx, req)
+
+			if err != nil {
+				return nil, fmt.Errorf("find missing blobs[%d:%d]: %w", startIndex, endIndex, err)
+			}
+			missingBlobs = append(missingBlobs, convertToFileDigests(resp.GetMissingBlobDigests())...)
+		}
+	} else {
+		resp, err := c.casClient.FindMissingBlobs(ctx, req)
+
+		if err != nil {
+			return nil, fmt.Errorf("find missing blobs: %w", err)
+		}
+		missingBlobs = convertToFileDigests(resp.GetMissingBlobDigests())
 	}
 
-	return convertToFileDigests(resp.GetMissingBlobDigests()), nil
+	return missingBlobs, nil
 }
 
 func convertToBlobDigests(digests []*FileDigest) []*remoteexecution.Digest {