Replace CAS with CS and CLS

This commit builds on top of our split and splice blob support to make
it a mandatory first class feature in Buildbarn. With this commit the
Content Addressable Storage (CAS) is created from two Storage
configurations that work in tandem. A Chunk Storage (CS) which is
content addressed and contains chunks of blobs, and a Chunk List Storage
(CLS) which is addressed by a blob digest and contains a manifest
describing the chunks that make up the blob.

All api calls are automatically translated to use Chunk Lists created
with RepMaxCDC. Effectively this means that large blobs no longer exists
in the storage layer, individual chunks of the large blobs are in turn
deduplicated in such a manner that the chunks are stored only once.

The automatic translation makes certain that clients that are not cdc
aware can still continue to use the storage backend without performing
any changes. Clients which support RepMaxCDC also gets a significant
reduction in the amount of blobs to transfer as they only need to
transfer modified chunks rather than the entire blob.

Author: meroton-benjamin

cmd/bb_copy/main.go

@@ -42,7 +42,7 @@ func main() {
 
 		grpcClientFactory := grpc.NewBaseClientFactory(grpc.BaseClientDialer, nil, nil, nil)
 
-		blobAccessCreator := blobstore_configuration.NewCASBlobAccessCreator(
+		blobAccessCreator := blobstore_configuration.NewCSBlobAccessCreator(
 			grpcClientFactory,
 			int(configuration.MaximumMessageSizeBytes),
 			bb_zstd.NewPoolFromConfiguration(nil),
@@ -68,7 +68,7 @@ func main() {
 			configuration.Replicator,
 			source.BlobAccess,
 			sink,
-			blobstore_configuration.NewCASBlobReplicatorCreator(grpcClientFactory),
+			blobstore_configuration.NewCSBlobReplicatorCreator(grpcClientFactory),
 		)
 		if err != nil {
 			return util.StatusWrap(err, "Failed to create replicator")

cmd/bb_replicator/main.go

@@ -33,7 +33,7 @@ func main() {
 			return util.StatusWrap(err, "Failed to apply global configuration options")
 		}
 
-		blobAccessCreator := blobstore_configuration.NewCASBlobAccessCreator(
+		blobAccessCreator := blobstore_configuration.NewCSBlobAccessCreator(
 			grpcClientFactory,
 			int(configuration.MaximumMessageSizeBytes),
 			bb_zstd.NewPoolFromConfiguration(nil),
@@ -59,7 +59,7 @@ func main() {
 			configuration.Replicator,
 			source.BlobAccess,
 			sink,
-			blobstore_configuration.NewCASBlobReplicatorCreator(grpcClientFactory),
+			blobstore_configuration.NewCSBlobReplicatorCreator(grpcClientFactory),
 		)
 		if err != nil {
 			return util.StatusWrap(err, "Failed to create replicator")

cmd/bb_storage/BUILD.bazel

@@ -10,17 +10,21 @@ go_library(
         "//pkg/auth",
         "//pkg/auth/configuration",
         "//pkg/blobstore",
+        "//pkg/blobstore/cdc",
         "//pkg/blobstore/configuration",
         "//pkg/blobstore/grpcservers",
         "//pkg/builder",
         "//pkg/capabilities",
+        "//pkg/clock",
+        "//pkg/digest",
         "//pkg/global",
         "//pkg/grpc",
         "//pkg/program",
         "//pkg/proto/configuration/bb_storage",
         "//pkg/proto/fsac",
         "//pkg/proto/icas",
         "//pkg/proto/iscc",
+        "//pkg/ttlcache",
         "//pkg/util",
         "//pkg/zstd",
         "@bazel_remote_apis//build/bazel/remote/execution/v2:remote_execution_go_proto",

cmd/bb_storage/main.go

@@ -9,6 +9,7 @@ import (
 	"github.com/buildbarn/bb-storage/pkg/auth"
 	auth_configuration "github.com/buildbarn/bb-storage/pkg/auth/configuration"
 	"github.com/buildbarn/bb-storage/pkg/blobstore"
+	"github.com/buildbarn/bb-storage/pkg/blobstore/cdc"
 	blobstore_configuration "github.com/buildbarn/bb-storage/pkg/blobstore/configuration"
 	"github.com/buildbarn/bb-storage/pkg/blobstore/grpcservers"
 	"github.com/buildbarn/bb-storage/pkg/builder"
@@ -54,55 +55,57 @@ func main() {
 		var cacheCapabilitiesAuthorizers []auth.Authorizer
 
 		// Content Addressable Storage (CAS).
-		var contentAddressableStorageInfo *blobstore_configuration.BlobAccessInfo
-		var contentAddressableStorage blobstore.BlobAccess
-		if configuration.ContentAddressableStorage != nil {
-			info, authorizedBackend, allAuthorizers, err := newScannableBlobAccess(
+		var contentAddressableStorage cdc.ContentAddressableStorage
+		var authorizedContentAddressableStorage cdc.ContentAddressableStorage
+		if configuration.ContentAddressableStorageServer != nil {
+			cas, chunkStorage, chunkListStorage, _, cdcParametersFetcher, err := blobstore_configuration.NewCASFromConfiguration(
 				dependenciesGroup,
-				configuration.ContentAddressableStorage,
-				blobstore_configuration.NewCASBlobAccessCreator(
-					grpcClientFactory,
-					int(configuration.MaximumMessageSizeBytes),
-					zstdPool,
-				),
+				configuration.ContentAddressableStorageServer.ContentAddressableStorage,
 				grpcClientFactory,
+				int(configuration.MaximumMessageSizeBytes),
+				zstdPool,
 			)
 			if err != nil {
 				return util.StatusWrap(err, "Failed to create Content Addressable Storage")
 			}
+			contentAddressableStorage = cas
+
+			// Create authorizers.
+			getAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.ContentAddressableStorageServer.GetAuthorizer, dependenciesGroup, grpcClientFactory)
+			if err != nil {
+				return util.StatusWrap(err, "Failed to create Get() authorizer for Content Addressable Storage")
+			}
+			putAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.ContentAddressableStorageServer.PutAuthorizer, dependenciesGroup, grpcClientFactory)
+			if err != nil {
+				return util.StatusWrap(err, "Failed to create Put() authorizer for Content Addressable Storage")
+			}
+			findMissingAuthorizer, err := auth_configuration.DefaultAuthorizerFactory.NewAuthorizerFromConfiguration(configuration.ContentAddressableStorageServer.FindMissingAuthorizer, dependenciesGroup, grpcClientFactory)
+			if err != nil {
+				return util.StatusWrap(err, "Failed to create FindMissing() authorizer for Content Addressable Storage")
+			}
+
+			// Create authorized versions of the backends.
+			authorizedChunkStorage := blobstore.NewAuthorizingBlobAccess(chunkStorage, getAuthorizer, putAuthorizer, findMissingAuthorizer)
+			authorizedChunkListStorage := blobstore.NewAuthorizingBlobAccess(chunkListStorage, getAuthorizer, putAuthorizer, findMissingAuthorizer)
+			authorizedChunkListFetcher := blobstore.NewBlobAccessChunkListFetcher(authorizedChunkListStorage, int(configuration.MaximumMessageSizeBytes))
+			authorizedContentAddressableStorage = cdc.NewContentAddressableStorage(
+				authorizedChunkStorage,
+				authorizedChunkListStorage,
+				authorizedChunkListFetcher,
+				cdcParametersFetcher,
+				contentAddressableStorage.GetDigestKeyFormat(),
+			)
+			// Create the Chunk Storage (CS).
 			cacheCapabilitiesProviders = append(
 				cacheCapabilitiesProviders,
-				info.BlobAccess,
+				chunkStorage,
 				capabilities.NewStaticProvider(&remoteexecution.ServerCapabilities{
 					CacheCapabilities: &remoteexecution.CacheCapabilities{
 						SupportedCompressors: configuration.SupportedCompressors,
 					},
 				}),
 			)
-			cacheCapabilitiesAuthorizers = append(cacheCapabilitiesAuthorizers, allAuthorizers...)
-			contentAddressableStorageInfo = &info
-			contentAddressableStorage = authorizedBackend
-		}
-
-		// Chunk List Storage (CLS).
-		var chunkListStorage blobstore.BlobAccess
-		if configuration.ChunkListStorage != nil {
-			info, authorizedBackend, allAuthorizers, err := newScannableBlobAccess(
-				dependenciesGroup,
-				configuration.ChunkListStorage,
-				blobstore_configuration.NewCLSBlobAccessCreator(
-					contentAddressableStorageInfo,
-					grpcClientFactory,
-					int(configuration.MaximumMessageSizeBytes),
-				),
-				grpcClientFactory,
-			)
-			if err != nil {
-				return util.StatusWrap(err, "Failed to create Chunk List Storage")
-			}
-			cacheCapabilitiesProviders = append(cacheCapabilitiesProviders, info.BlobAccess)
-			cacheCapabilitiesAuthorizers = append(cacheCapabilitiesAuthorizers, allAuthorizers...)
-			chunkListStorage = authorizedBackend
+			cacheCapabilitiesAuthorizers = append(cacheCapabilitiesAuthorizers, getAuthorizer, putAuthorizer, findMissingAuthorizer)
 		}
 
 		// Action Cache (AC).
@@ -112,7 +115,7 @@ func main() {
 				dependenciesGroup,
 				configuration.ActionCache,
 				blobstore_configuration.NewACBlobAccessCreator(
-					contentAddressableStorageInfo,
+					contentAddressableStorage,
 					grpcClientFactory,
 					int(configuration.MaximumMessageSizeBytes),
 				),
@@ -213,10 +216,9 @@ func main() {
 		if err := bb_grpc.NewServersFromConfigurationAndServe(
 			configuration.GrpcServers,
 			func(s grpc.ServiceRegistrar) {
-				if contentAddressableStorage != nil {
+				if authorizedContentAddressableStorage != nil {
 					contentAddressableStorageServer := grpcservers.NewContentAddressableStorageServer(
-						contentAddressableStorage,
-						chunkListStorage,
+						authorizedContentAddressableStorage,
 						configuration.MaximumMessageSizeBytes,
 					)
 					remoteexecution.RegisterContentAddressableStorageServer(
@@ -226,8 +228,7 @@ func main() {
 					bytestream.RegisterByteStreamServer(
 						s,
 						grpcservers.NewByteStreamServer(
-							contentAddressableStorage,
-							1<<16,
+							authorizedContentAddressableStorage,
 							zstdPool,
 						),
 					)

internal/mock/BUILD.bazel

@@ -152,6 +152,19 @@ gomock(
     package = "mock",
 )
 
+gomock(
+    name = "cdc",
+    out = "cdc.go",
+    interfaces = [
+        "ContentAddressableStorage",
+        "ParametersFetcher",
+    ],
+    library = "//pkg/blobstore/cdc",
+    mockgen_model_library = "@org_uber_go_mock//mockgen/model",
+    mockgen_tool = "@org_uber_go_mock//mockgen",
+    package = "mock",
+)
+
 gomock(
     name = "clock",
     out = "clock.go",
@@ -367,6 +380,7 @@ go_library(
         "buffer.go",
         "builder.go",
         "capabilities.go",
+        "cdc.go",
         "clock.go",
         "cloud_aws.go",
         "cloud_gcp.go",
@@ -391,6 +405,8 @@ go_library(
         "//pkg/auth",
         "//pkg/blobstore",
         "//pkg/blobstore/buffer",
+        "//pkg/blobstore/cdc",
+        "//pkg/blobstore/chunklist",
         "//pkg/blobstore/local",
         "//pkg/blobstore/sharding",
         "//pkg/blobstore/slicing",

pkg/blobstore/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "authorizing_blob_access.go",
         "blob_access.go",
         "cas_read_buffer_factory.go",
+        "chunk_list_fetcher.go",
         "cls_read_buffer_factory.go",
         "deadline_enforcing_blob_access.go",
         "demultiplexing_blob_access.go",
@@ -22,7 +23,6 @@ go_library(
         "metrics_blob_access.go",
         "read_buffer_factory.go",
         "read_canarying_blob_access.go",
-        "reference_expanding_blob_access.go",
         "validation_caching_read_buffer_factory.go",
         "visit_topologically_sorted_tree.go",
         "zip_reading_blob_access.go",
@@ -33,21 +33,18 @@ go_library(
     deps = [
         "//pkg/auth",
         "//pkg/blobstore/buffer",
+        "//pkg/blobstore/chunklist",
         "//pkg/blobstore/slicing",
         "//pkg/capabilities",
         "//pkg/clock",
-        "//pkg/cloud/aws",
-        "//pkg/cloud/gcp",
         "//pkg/digest",
         "//pkg/eviction",
+        "//pkg/proto/blobstore/chunklist",
         "//pkg/proto/fsac",
         "//pkg/proto/icas",
         "//pkg/proto/iscc",
         "//pkg/util",
-        "//pkg/zstd",
         "@bazel_remote_apis//build/bazel/remote/execution/v2:remote_execution_go_proto",
-        "@com_github_aws_aws_sdk_go_v2//aws",
-        "@com_github_aws_aws_sdk_go_v2_service_s3//:s3",
         "@com_github_prometheus_client_golang//prometheus",
         "@org_golang_google_grpc//codes",
         "@org_golang_google_grpc//status",
@@ -68,7 +65,6 @@ go_test(
         "existence_caching_blob_access_test.go",
         "hierarchical_instance_names_blob_access_test.go",
         "read_canarying_blob_access_test.go",
-        "reference_expanding_blob_access_test.go",
         "validation_caching_read_buffer_factory_test.go",
         "visit_topologically_sorted_tree_test.go",
         "zip_reading_blob_access_test.go",
@@ -80,16 +76,10 @@ go_test(
         "//pkg/blobstore/buffer",
         "//pkg/digest",
         "//pkg/eviction",
-        "//pkg/proto/icas",
         "//pkg/testutil",
         "//pkg/util",
-        "//pkg/zstd",
         "@bazel_remote_apis//build/bazel/remote/execution/v2:remote_execution_go_proto",
         "@bazel_remote_apis//build/bazel/semver:semver_go_proto",
-        "@com_github_aws_aws_sdk_go_v2//aws",
-        "@com_github_aws_aws_sdk_go_v2_service_s3//:s3",
-        "@com_github_aws_aws_sdk_go_v2_service_s3//types",
-        "@com_github_klauspost_compress//zstd",
         "@com_github_stretchr_testify//require",
         "@org_golang_google_grpc//codes",
         "@org_golang_google_grpc//status",

pkg/blobstore/authorizing_blob_access.go

@@ -5,7 +5,6 @@ import (
 
 	"github.com/buildbarn/bb-storage/pkg/auth"
 	"github.com/buildbarn/bb-storage/pkg/blobstore/buffer"
-	"github.com/buildbarn/bb-storage/pkg/blobstore/slicing"
 	"github.com/buildbarn/bb-storage/pkg/digest"
 	"github.com/buildbarn/bb-storage/pkg/util"
 )
@@ -38,13 +37,6 @@ func (ba *authorizingBlobAccess) Get(ctx context.Context, d digest.Digest) buffe
 	return ba.BlobAccess.Get(ctx, d)
 }
 
-func (ba *authorizingBlobAccess) GetFromComposite(ctx context.Context, parentDigest, childDigest digest.Digest, slicer slicing.BlobSlicer) buffer.Buffer {
-	if err := auth.AuthorizeSingleInstanceName(ctx, ba.getAuthorizer, parentDigest.GetInstanceName()); err != nil {
-		return buffer.NewBufferFromError(util.StatusWrap(err, "Authorization"))
-	}
-	return ba.BlobAccess.GetFromComposite(ctx, parentDigest, childDigest, slicer)
-}
-
 func (ba *authorizingBlobAccess) Put(ctx context.Context, d digest.Digest, b buffer.Buffer) error {
 	if err := auth.AuthorizeSingleInstanceName(ctx, ba.putAuthorizer, d.GetInstanceName()); err != nil {
 		return util.StatusWrap(err, "Authorization")

pkg/blobstore/authorizing_blob_access_test.go

@@ -54,24 +54,6 @@ func TestAuthorizingBlobAccess(t *testing.T) {
 		testutil.RequireEqualStatus(t, status.Error(codes.PermissionDenied, "Authorization: You shall not pass"), err)
 	})
 
-	t.Run("GetFromComposite-Allowed", func(t *testing.T) {
-		getAuthorizer.EXPECT().Authorize(ctx, beepSlice).Return([]error{nil})
-		blobSlicer := mock.NewMockBlobSlicer(ctrl)
-		baseBlobAccess.EXPECT().GetFromComposite(ctx, d, d2, blobSlicer).Return(wantBuf)
-
-		gotBuf, err := ba.GetFromComposite(ctx, d, d2, blobSlicer).ToByteSlice(30)
-		require.NoError(t, err)
-		require.Equal(t, wantBytes, gotBuf)
-	})
-
-	t.Run("GetFromComposite-Denied", func(t *testing.T) {
-		blobSlicer := mock.NewMockBlobSlicer(ctrl)
-		getAuthorizer.EXPECT().Authorize(ctx, beepSlice).Return([]error{status.Error(codes.PermissionDenied, "You shall not pass")})
-
-		_, err := ba.GetFromComposite(ctx, d, d2, blobSlicer).ToByteSlice(30)
-		testutil.RequireEqualStatus(t, status.Error(codes.PermissionDenied, "Authorization: You shall not pass"), err)
-	})
-
 	t.Run("Put-Allowed", func(t *testing.T) {
 		putAuthorizer.EXPECT().Authorize(ctx, beepSlice).Return([]error{nil})
 		baseBlobAccess.EXPECT().Put(ctx, d, wantBuf).Return(nil)

pkg/blobstore/blob_access.go

@@ -4,7 +4,6 @@ import (
 	"context"
 
 	"github.com/buildbarn/bb-storage/pkg/blobstore/buffer"
-	"github.com/buildbarn/bb-storage/pkg/blobstore/slicing"
 	"github.com/buildbarn/bb-storage/pkg/capabilities"
 	"github.com/buildbarn/bb-storage/pkg/digest"
 )
@@ -16,7 +15,6 @@ type BlobAccess interface {
 	capabilities.Provider
 
 	Get(ctx context.Context, digest digest.Digest) buffer.Buffer
-	GetFromComposite(ctx context.Context, parentDigest, childDigest digest.Digest, slicer slicing.BlobSlicer) buffer.Buffer
 	Put(ctx context.Context, digest digest.Digest, b buffer.Buffer) error
 	FindMissing(ctx context.Context, digests digest.Set) (digest.Set, error)
 }