test

Author: tyler-french

MODULE.bazel

@@ -14,6 +14,8 @@ bazel_dep(name = "protobuf", version = "33.0")
 bazel_dep(name = "rules_foreign_cc", version = "0.15.1")
 bazel_dep(name = "rules_go", version = "0.58.3")
 bazel_dep(name = "rules_img", version = "0.2.9")
+bazel_dep(name = "rules_oci", version = "2.2.6")
+bazel_dep(name = "rules_pkg", version = "1.1.0")
 bazel_dep(name = "rules_jsonnet", version = "0.7.2")
 bazel_dep(name = "rules_nodejs", version = "6.6.0")
 bazel_dep(name = "rules_proto", version = "7.1.0")
@@ -161,3 +163,12 @@ pull(
     registry = "gcr.io",
     repository = "distroless/static",
 )
+
+oci = use_extension("@rules_oci//oci:extensions.bzl", "oci")
+oci.pull(
+    name = "distroless_static_oci",
+    digest = "sha256:3f2b64ef97bd285e36132c684e6b2ae8f2723293d09aae046196cca64251acac",
+    image = "gcr.io/distroless/static",
+    platforms = ["linux/amd64"],
+)
+use_repo(oci, "distroless_static_oci")

MODULE.bazel.lock

@@ -1,5 +1,5 @@
 load("@rules_go//go:def.bzl", "go_binary", "go_library")
-load("//tools:container.bzl", "container_push_official", "multiarch_go_image")
+load("//tools:container.bzl", "container_load", "container_push_official", "multiarch_go_image")
 
 go_library(
     name = "bb_storage_lib",
@@ -48,3 +48,10 @@ container_push_official(
     component = "bb-storage",
     image = ":bb_storage_container",
 )
+
+container_load(
+    name = "bb_storage_tarball",
+    binary = ":bb_storage",
+    repo_tags = ["bb_storage_tfrench_test:latest"],
+)
+

cmd/bb_storage/BUILD.bazel

@@ -93,14 +93,10 @@ func (bac *casBlobAccessCreator) NewCustomBlobAccess(terminationGroup program.Gr
 		if err != nil {
 			return BlobAccessInfo{}, "", err
 		}
-		compressionThresholdBytes := int64(0)
-		if backend.Grpc.EnableCompression {
-			compressionThresholdBytes = 100
-		}
 		// TODO: Should we provide a configuration option, so
 		// that digest.KeyWithoutInstance can be used?
 		return BlobAccessInfo{
-			BlobAccess:      grpcclients.NewCASBlobAccess(client, uuid.NewRandom, 64<<10, compressionThresholdBytes),
+			BlobAccess:      grpcclients.NewCASBlobAccess(client, uuid.NewRandom, 64<<10, backend.Grpc.EnableBytestreamZstdCompression),
 			DigestKeyFormat: digest.KeyWithInstance,
 		}, "grpc", nil
 	case *pb.BlobAccessConfiguration_ReferenceExpanding:

pkg/blobstore/configuration/cas_blob_access_creator.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"slices"
 	"sync"
+	"sync/atomic"
 
 	remoteexecution "github.com/bazelbuild/remote-apis/build/bazel/remote/execution/v2"
 	"github.com/buildbarn/bb-storage/pkg/blobstore"
@@ -28,29 +29,26 @@ type casBlobAccess struct {
 	capabilitiesClient              remoteexecution.CapabilitiesClient
 	uuidGenerator                   util.UUIDGenerator
 	readChunkSize                   int
-	compressionThresholdBytes       int64
-	supportedCompressors            []remoteexecution.Compressor_Value
-	supportedCompressorsMutex       sync.RWMutex
-	capabilitiesCalled              bool
+	enableZSTDCompression           bool
+	supportedCompressors            atomic.Pointer[[]remoteexecution.Compressor_Value]
 }
 
 // NewCASBlobAccess creates a BlobAccess handle that relays any requests
-// to a GRPC service that implements the bytestream.ByteStream and
+// to a gRPC service that implements the bytestream.ByteStream and
 // remoteexecution.ContentAddressableStorage services. Those are the
 // services that Bazel uses to access blobs stored in the Content
 // Addressable Storage.
 //
-// If compressionThresholdBytes is > 0, the client will attempt to use
-// ZSTD compression for blobs larger than this threshold. The server's
-// supported compressors will be checked via GetCapabilities().
-func NewCASBlobAccess(client grpc.ClientConnInterface, uuidGenerator util.UUIDGenerator, readChunkSize int, compressionThresholdBytes int64) blobstore.BlobAccess {
+// If enableZSTDCompression is true, the client will use ZSTD compression
+// for ByteStream operations if the server supports it.
+func NewCASBlobAccess(client grpc.ClientConnInterface, uuidGenerator util.UUIDGenerator, readChunkSize int, enableZSTDCompression bool) blobstore.BlobAccess {
 	return &casBlobAccess{
 		byteStreamClient:                bytestream.NewByteStreamClient(client),
 		contentAddressableStorageClient: remoteexecution.NewContentAddressableStorageClient(client),
 		capabilitiesClient:              remoteexecution.NewCapabilitiesClient(client),
 		uuidGenerator:                   uuidGenerator,
 		readChunkSize:                   readChunkSize,
-		compressionThresholdBytes:       compressionThresholdBytes,
+		enableZSTDCompression:           enableZSTDCompression,
 	}
 }
 
@@ -178,34 +176,27 @@ func (w *zstdByteStreamWriter) Close() error {
 
 const resourceNameHeader = "build.bazel.remote.execution.v2.resource-name"
 
-// shouldUseCompression checks if compression should be used for a blob of the given size.
-// It also ensures GetCapabilities has been called to negotiate compression support.
-func (ba *casBlobAccess) shouldUseCompression(ctx context.Context, digest digest.Digest) (bool, error) {
-	if ba.compressionThresholdBytes <= 0 || digest.GetSizeBytes() < ba.compressionThresholdBytes {
+// shouldUseZSTDCompression checks if ZSTD compression should be used.
+// It ensures GetCapabilities has been called to negotiate compression support.
+func (ba *casBlobAccess) shouldUseZSTDCompression(ctx context.Context, digest digest.Digest) (bool, error) {
+	if !ba.enableZSTDCompression {
 		return false, nil
 	}
 
-	ba.supportedCompressorsMutex.RLock()
-	capabilitiesCalled := ba.capabilitiesCalled
-	supportedCompressors := ba.supportedCompressors
-	ba.supportedCompressorsMutex.RUnlock()
-
-	if !capabilitiesCalled {
-		// Call GetCapabilities to check server support
-		_, err := ba.GetCapabilities(ctx, digest.GetDigestFunction().GetInstanceName())
-		if err != nil {
+	supportedCompressors := ba.supportedCompressors.Load()
+	if supportedCompressors == nil {
+		// Call GetCapabilities to check server support.
+		if _, err := ba.GetCapabilities(ctx, digest.GetDigestFunction().GetInstanceName()); err != nil {
 			return false, err
 		}
-		ba.supportedCompressorsMutex.RLock()
-		supportedCompressors = ba.supportedCompressors
-		ba.supportedCompressorsMutex.RUnlock()
+		supportedCompressors = ba.supportedCompressors.Load()
 	}
 
-	return slices.Contains(supportedCompressors, remoteexecution.Compressor_ZSTD), nil
+	return slices.Contains(*supportedCompressors, remoteexecution.Compressor_ZSTD), nil
 }
 
 func (ba *casBlobAccess) Get(ctx context.Context, digest digest.Digest) buffer.Buffer {
-	useCompression, err := ba.shouldUseCompression(ctx, digest)
+	useCompression, err := ba.shouldUseZSTDCompression(ctx, digest)
 	if err != nil {
 		return buffer.NewBufferFromError(err)
 	}
@@ -248,7 +239,7 @@ func (ba *casBlobAccess) GetFromComposite(ctx context.Context, parentDigest, chi
 }
 
 func (ba *casBlobAccess) Put(ctx context.Context, digest digest.Digest, b buffer.Buffer) error {
-	useCompression, err := ba.shouldUseCompression(ctx, digest)
+	useCompression, err := ba.shouldUseZSTDCompression(ctx, digest)
 	if err != nil {
 		b.Discard()
 		return err
@@ -387,11 +378,8 @@ func (ba *casBlobAccess) GetCapabilities(ctx context.Context, instanceName diges
 
 	cacheCapabilities := serverCapabilities.CacheCapabilities
 
-	// Store supported compressors for compression negotiation
-	ba.supportedCompressorsMutex.Lock()
-	ba.supportedCompressors = cacheCapabilities.SupportedCompressors
-	ba.capabilitiesCalled = true
-	ba.supportedCompressorsMutex.Unlock()
+	// Store supported compressors for compression negotiation.
+	ba.supportedCompressors.Store(&cacheCapabilities.SupportedCompressors)
 
 	// Only return fields that pertain to the Content Addressable
 	// Storage. Don't set 'max_batch_total_size_bytes', as we don't