diff --git a/app/artifact-cas/cmd/wire.go b/app/artifact-cas/cmd/wire.go index a3866888e..409723ca0 100644 --- a/app/artifact-cas/cmd/wire.go +++ b/app/artifact-cas/cmd/wire.go @@ -21,11 +21,15 @@ package main import ( + "context" + "time" + "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/server" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/service" "github.com/chainloop-dev/chainloop/app/controlplane/pkg/auditor" "github.com/chainloop-dev/chainloop/pkg/blobmanager/loader" + "github.com/chainloop-dev/chainloop/pkg/cache/casexistence" "github.com/chainloop-dev/chainloop/pkg/credentials" "github.com/chainloop-dev/chainloop/pkg/natsconn" "github.com/go-kratos/kratos/v2/log" @@ -47,16 +51,35 @@ func wireApp(*conf.Bootstrap, *conf.Server, *conf.Auth, credentials.Reader, log. // publish-only: the control plane owns the chainloop-audit stream configuration auditor.NewPublishOnlyAuditLogPublisher, service.NewAuditDispatcher, + newExistenceCache, ), ) } -func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap) []service.NewOpt { +// backendClientTTL is how long a loaded backend client, and its credentials, is reused for uploads +const backendClientTTL = 5 * time.Minute + +func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap, existence *casexistence.Cache) []service.NewOpt { return []service.NewOpt{ service.WithLogger(l), service.WithAuditDispatcher(audit), service.WithStagingDir(bc.GetStagingDir()), + service.WithBackendClientCache(backendClientTTL), + service.WithExistenceCache(existence), + } +} + +// newExistenceCache returns the cache of blobs known to exist in a backend. +// It is shared through NATS KV when the connection is set, in memory otherwise. +func newExistenceCache(rc *natsconn.ReloadableConnection, logger log.Logger) (*casexistence.Cache, func(), error) { + ctx, cancel := context.WithCancel(context.Background()) + c, err := casexistence.New(ctx, rc, logger) + if err != nil { + cancel() + return nil, nil, err } + + return c, cancel, nil } // newNatsConfig converts the proto config to a plain natsconn.Config, nil when unset diff --git a/app/artifact-cas/cmd/wire_gen.go b/app/artifact-cas/cmd/wire_gen.go index 4b48eef04..aa28bcf9a 100644 --- a/app/artifact-cas/cmd/wire_gen.go +++ b/app/artifact-cas/cmd/wire_gen.go @@ -7,14 +7,17 @@ package main import ( + "context" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/conf" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/server" "github.com/chainloop-dev/chainloop/app/artifact-cas/internal/service" "github.com/chainloop-dev/chainloop/app/controlplane/pkg/auditor" "github.com/chainloop-dev/chainloop/pkg/blobmanager/loader" + "github.com/chainloop-dev/chainloop/pkg/cache/casexistence" "github.com/chainloop-dev/chainloop/pkg/credentials" "github.com/chainloop-dev/chainloop/pkg/natsconn" "github.com/go-kratos/kratos/v2/log" + "time" ) import ( @@ -37,37 +40,48 @@ func wireApp(bootstrap *conf.Bootstrap, confServer *conf.Server, auth *conf.Auth return nil, nil, err } auditDispatcher := service.NewAuditDispatcher(auditLogPublisher, logger) - v := serviceOpts(logger, auditDispatcher, bootstrap) + cache, cleanup2, err := newExistenceCache(reloadableConnection, logger) + if err != nil { + cleanup() + return nil, nil, err + } + v := serviceOpts(logger, auditDispatcher, bootstrap, cache) byteStreamService := service.NewByteStreamService(providers, v...) resourceService := service.NewResourceService(providers, v...) validator, err := newProtoValidator() if err != nil { + cleanup2() cleanup() return nil, nil, err } grpcServer, err := server.NewGRPCServer(confServer, auth, byteStreamService, resourceService, providers, validator, logger) if err != nil { + cleanup2() cleanup() return nil, nil, err } downloadService := service.NewDownloadService(providers, v...) httpServer, err := server.NewHTTPServer(confServer, auth, downloadService, providers, logger) if err != nil { + cleanup2() cleanup() return nil, nil, err } httpMetricsServer, err := server.NewHTTPMetricsServer(confServer) if err != nil { + cleanup2() cleanup() return nil, nil, err } - tracerProvider, cleanup2, err := server.NewTracerProvider(bootstrap, logger) + tracerProvider, cleanup3, err := server.NewTracerProvider(bootstrap, logger) if err != nil { + cleanup2() cleanup() return nil, nil, err } mainApp := newApp(logger, grpcServer, httpServer, httpMetricsServer, providers, tracerProvider) return mainApp, func() { + cleanup3() cleanup2() cleanup() }, nil @@ -75,8 +89,24 @@ func wireApp(bootstrap *conf.Bootstrap, confServer *conf.Server, auth *conf.Auth // wire.go: -func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap) []service.NewOpt { - return []service.NewOpt{service.WithLogger(l), service.WithAuditDispatcher(audit), service.WithStagingDir(bc.GetStagingDir())} +// backendClientTTL is how long a loaded backend client, and its credentials, is reused for uploads +const backendClientTTL = 5 * time.Minute + +func serviceOpts(l log.Logger, audit *service.AuditDispatcher, bc *conf.Bootstrap, existence *casexistence.Cache) []service.NewOpt { + return []service.NewOpt{service.WithLogger(l), service.WithAuditDispatcher(audit), service.WithStagingDir(bc.GetStagingDir()), service.WithBackendClientCache(backendClientTTL), service.WithExistenceCache(existence)} +} + +// newExistenceCache returns the cache of blobs known to exist in a backend. +// It is shared through NATS KV when the connection is set, in memory otherwise. +func newExistenceCache(rc *natsconn.ReloadableConnection, logger log.Logger) (*casexistence.Cache, func(), error) { + ctx, cancel := context.WithCancel(context.Background()) + c, err := casexistence.New(ctx, rc, logger) + if err != nil { + cancel() + return nil, nil, err + } + + return c, cancel, nil } // newNatsConfig converts the proto config to a plain natsconn.Config, nil when unset diff --git a/app/artifact-cas/internal/service/bytestream.go b/app/artifact-cas/internal/service/bytestream.go index 6a4cbe5f7..f2d40576d 100644 --- a/app/artifact-cas/internal/service/bytestream.go +++ b/app/artifact-cas/internal/service/bytestream.go @@ -79,34 +79,33 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro return kerrors.BadRequest("resource name", err.Error()) } - storageBackend, err := s.loadBackend(ctx, info.BackendType, info.StoredSecretID) - if err != nil && kerrors.IsNotFound(err) { - return err - } else if err != nil { - return sl.LogAndMaskErr(err, s.log) + // A blob cached as present skips the secrets manager and the backend. The + // key is namespaced by the organization and backend of the signed token. + existenceKey := s.existenceKey(info, req.resource.Digest) + if entry, ok := s.lookupExistence(ctx, existenceKey); ok { + s.log.Infow("msg", "artifact already exists (cached)", "digest", req.resource.Digest) + s.dispatchSkippedUpload(info, req.resource, entry.Size) + return stream.SendAndClose(&bytestream.WriteResponse{}) } // We check if the file already exists even before we wait for the whole buffer to be filled - if exists, err := storageBackend.Exists(ctx, req.resource.Digest); err != nil { - return sl.LogAndMaskErr(err, s.log) - } else if exists { + storageBackend, exists, err := s.backendExists(ctx, info, req.resource.Digest) + if err != nil { + return err + } + + if exists { s.log.Infow("msg", "artifact already exists", "digest", req.resource.Digest) - if s.audit.shouldEmit(info) { - // the stored size is not known at the dedup point, look it up best-effort + if existenceKey != "" || s.audit.shouldEmit(info) { + // the stored size is not known at the dedup point, look it up best-effort. + // Only a known size is cached, so cached hits keep complete audit events. var size int64 if r, err := storageBackend.Describe(ctx, req.resource.Digest); err == nil { size = r.Size + s.storeExistence(ctx, existenceKey, size) } - s.audit.Dispatch(&events.CASArtifactUploaded{ - CASArtifactBase: &events.CASArtifactBase{ - Digest: req.resource.Digest, - SizeBytes: size, - FileName: req.resource.FileName, - BackendType: info.BackendType, - }, - Skipped: true, - }, info) + s.dispatchSkippedUpload(info, req.resource, size) } return stream.SendAndClose(&bytestream.WriteResponse{}) } @@ -137,6 +136,8 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro return status.Error(codes.InvalidArgument, err.Error()) } if backendErr, ok := errors.AsType[*backendUploadError](err); ok { + // the reused client may hold stale credentials + s.dropUploadBackend(info) return sl.LogAndMaskErr(backendErr.err, s.log) } if isClientDisconnect(err) { @@ -150,6 +151,7 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro } s.log.Infow("msg", "upload finished", "name", req.resource.FileName, "digest", req.resource.Digest, "size", committedSize) + s.storeExistence(ctx, existenceKey, committedSize) s.audit.Dispatch(&events.CASArtifactUploaded{ CASArtifactBase: &events.CASArtifactBase{ Digest: req.resource.Digest, @@ -162,6 +164,46 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro return stream.SendAndClose(&bytestream.WriteResponse{CommittedSize: committedSize}) } +// backendExists loads the backend of the token and checks if digest exists in +// it. A reused client that fails is dropped and loaded again one time, since +// its credentials may have been rotated. Returned errors are ready to be sent +// to the client. +func (s *ByteStreamService) backendExists(ctx context.Context, info *casJWT.Claims, digest string) (backend.UploaderDownloader, bool, error) { + for attempt := 0; ; attempt++ { + storageBackend, cached, err := s.loadUploadBackend(ctx, info) + if err != nil && kerrors.IsNotFound(err) { + return nil, false, err + } else if err != nil { + return nil, false, sl.LogAndMaskErr(err, s.log) + } + + exists, err := storageBackend.Exists(ctx, digest) + if err == nil { + return storageBackend, exists, nil + } + + s.dropUploadBackend(info) + if !cached || attempt > 0 { + return nil, false, sl.LogAndMaskErr(err, s.log) + } + s.log.Warnw("msg", "reused backend client failed, loading it again", "digest", digest, "error", err) + } +} + +// dispatchSkippedUpload emits the audit event of an upload skipped because the +// blob already exists +func (s *ByteStreamService) dispatchSkippedUpload(info *casJWT.Claims, resource *v1.CASResource, size int64) { + s.audit.Dispatch(&events.CASArtifactUploaded{ + CASArtifactBase: &events.CASArtifactBase{ + Digest: resource.Digest, + SizeBytes: size, + FileName: resource.FileName, + BackendType: info.BackendType, + }, + Skipped: true, + }, info) +} + // spillVerifyUpload stages an upload on local disk, verifies its digest, and // stores it. The client's bytes are streamed into a temporary file and hashed in // the same pass; the file is handed to the backend only once the hash matches the diff --git a/app/artifact-cas/internal/service/bytestream_cache_test.go b/app/artifact-cas/internal/service/bytestream_cache_test.go new file mode 100644 index 000000000..075d4d21e --- /dev/null +++ b/app/artifact-cas/internal/service/bytestream_cache_test.go @@ -0,0 +1,291 @@ +// +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service + +import ( + "context" + "errors" + "testing" + "time" + + v1 "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1" + casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" + "github.com/chainloop-dev/chainloop/pkg/blobmanager/mocks" + "github.com/chainloop-dev/chainloop/pkg/cache" + "github.com/chainloop-dev/chainloop/pkg/cache/casexistence" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + "google.golang.org/genproto/googleapis/bytestream" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" +) + +const otherOrgID = "2b7d0d6e-7d3f-4a39-9a43-7c1d2f0e9b11" + +// write uploads data as resource in a single chunk +func (s *bytestreamSuite) write(ctx context.Context, resource *v1.CASResource, data []byte) (*bytestream.WriteResponse, error) { + stream, err := s.client.Write(ctx) + s.Require().NoError(err) + s.Require().NoError(stream.Send(&bytestream.WriteRequest{ + ResourceName: encodeResource(s.T(), resource), + Data: data, + })) + + return stream.CloseAndRecv() +} + +func uploaderCtx(kv ...string) context.Context { + return metadata.NewOutgoingContext(context.Background(), metadata.Pairs(append([]string{"role", string(casJWT.Uploader)}, kv...)...)) +} + +// restart replaces the server with one that also has opts +func (s *bytestreamSuite) restart(opts ...NewOpt) { + s.TearDownTest() + s.setup(opts...) +} + +// withExistenceCache restarts the server with the existence cache only, so a +// cached hit can be told apart from a reused backend client. +func (s *bytestreamSuite) withExistenceCache() { + c, err := cache.New[casexistence.Entry](cache.WithTTL(time.Hour)) + s.Require().NoError(err) + s.restart(WithExistenceCache(c)) +} + +// R-001, R-006: a second upload of a blob that the backend reported as present +// makes no secrets manager or backend call, and its event keeps the size. +func (s *bytestreamSuite) TestExistenceCacheHitSkipsBackend() { + s.withExistenceCache() + s.ociBackend.On("Exists", mock.Anything, s.resource.Digest).Return(true, nil).Once() + s.ociBackend.On("Describe", mock.Anything, s.resource.Digest).Return(&v1.CASResource{ + FileName: s.resource.FileName, Digest: s.resource.Digest, Size: 1024, + }, nil).Once() + + for range 2 { + got, err := s.write(s.upCtx, s.resource, nil) + s.Require().NoError(err) + s.Equal(int64(0), got.CommittedSize) + } + + s.ociProvider.AssertNumberOfCalls(s.T(), "FromCredentials", 1) + s.ociBackend.AssertNumberOfCalls(s.T(), "Exists", 1) + s.ociBackend.AssertNumberOfCalls(s.T(), "Describe", 1) + + s.Require().Len(s.audit.published, 2) + for _, e := range s.audit.published { + info := decodeArtifactEvent(s.T(), e) + s.True(info.Skipped) + s.Equal(int64(1024), info.SizeBytes) + s.Equal(s.resource.FileName, info.FileName) + } +} + +// R-002: a successful upload is cached with its committed size. +func (s *bytestreamSuite) TestExistenceCacheUploadSuccessIsCached() { + s.withExistenceCache() + data := []byte("hello world") + resource := resourceWithDigest(data, "skynet.exe") + s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil).Once() + s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(nil).Once() + + got, err := s.write(s.upCtx, resource, data) + s.Require().NoError(err) + s.Equal(int64(len(data)), got.CommittedSize) + + got, err = s.write(s.upCtx, resource, data) + s.Require().NoError(err) + s.Equal(int64(0), got.CommittedSize) + + s.ociProvider.AssertNumberOfCalls(s.T(), "FromCredentials", 1) + s.Require().Len(s.audit.published, 2) + info := decodeArtifactEvent(s.T(), s.audit.published[1]) + s.True(info.Skipped) + s.Equal(int64(len(data)), info.SizeBytes) +} + +// R-002: a "not found" answer, or a found blob without a known size, is not cached. +func (s *bytestreamSuite) TestExistenceCacheNotCached() { + data := []byte("hello world") + resource := resourceWithDigest(data, "skynet.exe") + + testCases := []struct { + name string + setup func() + wantErr bool + }{ + { + name: "not found and failed upload", + setup: func() { + s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(errors.New("boom")) + }, + wantErr: true, + }, + { + name: "found but describe fails", + setup: func() { + s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(true, nil) + s.ociBackend.On("Describe", mock.Anything, resource.Digest).Return(nil, errors.New("boom")) + }, + }, + } + + for _, tc := range testCases { + s.Run(tc.name, func() { + s.withExistenceCache() + tc.setup() + for range 2 { + _, err := s.write(s.upCtx, resource, data) + if tc.wantErr { + assertGRPCError(s.T(), err, codes.Internal, "") + } else { + s.Require().NoError(err) + } + } + s.ociBackend.AssertNumberOfCalls(s.T(), "Exists", 2) + }) + } +} + +// R-003: an entry is only visible to the same organization and backend. +func (s *bytestreamSuite) TestExistenceCacheTenantNamespace() { + testCases := []struct { + name string + secondCtx context.Context + // backend checks the two uploads must make + wantOCIExists, wantStreamingExists int + }{ + { + name: "different organization on shared storage", + secondCtx: uploaderCtx("org-id", otherOrgID), + wantOCIExists: 2, + }, + { + name: "same organization, different backend", + secondCtx: uploaderCtx("backend-streaming", "true"), + wantOCIExists: 1, + wantStreamingExists: 1, + }, + { + name: "organization missing from the token", + secondCtx: uploaderCtx("org-id", ""), + wantOCIExists: 2, + }, + } + + for _, tc := range testCases { + s.Run(tc.name, func() { + s.withExistenceCache() + describe := &v1.CASResource{Digest: s.resource.Digest, Size: 1} + for _, b := range []*mocks.UploaderDownloader{s.ociBackend, s.streamingBackend} { + b.On("Exists", mock.Anything, s.resource.Digest).Maybe().Return(true, nil) + b.On("Describe", mock.Anything, s.resource.Digest).Maybe().Return(describe, nil) + } + + _, err := s.write(s.upCtx, s.resource, nil) + s.Require().NoError(err) + _, err = s.write(tc.secondCtx, s.resource, nil) + s.Require().NoError(err) + + s.ociBackend.AssertNumberOfCalls(s.T(), "Exists", tc.wantOCIExists) + s.streamingBackend.AssertNumberOfCalls(s.T(), "Exists", tc.wantStreamingExists) + }) + } +} + +// R-008: a failing cache is a miss and never fails the upload. +func (s *bytestreamSuite) TestExistenceCacheFailureIsMiss() { + s.restart(WithExistenceCache(failingCache{})) + + data := []byte("hello world") + resource := resourceWithDigest(data, "skynet.exe") + s.ociBackend.On("Exists", mock.Anything, resource.Digest).Return(false, nil) + s.ociBackend.On("Upload", mock.Anything, mock.Anything, resource).Return(nil) + + for range 2 { + got, err := s.write(s.upCtx, resource, data) + s.Require().NoError(err) + s.Equal(int64(len(data)), got.CommittedSize) + } + s.ociBackend.AssertNumberOfCalls(s.T(), "Upload", 2) +} + +// R-007: a cache miss on the existence check reuses the loaded backend client. +func (s *bytestreamSuite) TestClientCacheReusesClient() { + s.restart(WithBackendClientCache(time.Hour)) + s.ociBackend.On("Exists", mock.Anything, mock.Anything).Return(true, nil) + + for _, d := range []string{"digest-a", "digest-b"} { + _, err := s.write(uploaderCtx("source-internal", "true"), &v1.CASResource{Digest: d}, nil) + s.Require().NoError(err) + } + s.ociProvider.AssertNumberOfCalls(s.T(), "FromCredentials", 1) + + // a client is never shared between organizations + _, err := s.write(uploaderCtx("source-internal", "true", "org-id", otherOrgID), s.resource, nil) + s.Require().NoError(err) + s.ociProvider.AssertNumberOfCalls(s.T(), "FromCredentials", 2) +} + +// D-009: an error from a reused client drops it and loads it again one time. +func (s *bytestreamSuite) TestClientCacheReloadOnError() { + s.restart(WithBackendClientCache(time.Hour)) + ctx := uploaderCtx("source-internal", "true") + s.ociBackend.On("Exists", mock.Anything, mock.Anything).Return(true, nil).Once() + _, err := s.write(ctx, s.resource, nil) + s.Require().NoError(err) + + s.ociBackend.On("Exists", mock.Anything, mock.Anything).Return(false, errors.New("expired credentials")).Once() + s.ociBackend.On("Exists", mock.Anything, mock.Anything).Return(true, nil).Once() + _, err = s.write(ctx, s.resource, nil) + s.Require().NoError(err) + s.ociProvider.AssertNumberOfCalls(s.T(), "FromCredentials", 2) +} + +// a fresh client that fails is not loaded again +func (s *bytestreamSuite) TestClientCacheNoReloadOfFreshClient() { + s.restart(WithBackendClientCache(time.Hour)) + s.ociBackend.On("Exists", mock.Anything, mock.Anything).Return(false, errors.New("boom")) + + _, err := s.write(s.upCtx, s.resource, nil) + assertGRPCError(s.T(), err, codes.Internal, "") + s.ociProvider.AssertNumberOfCalls(s.T(), "FromCredentials", 1) + s.ociBackend.AssertNumberOfCalls(s.T(), "Exists", 1) +} + +type failingCache struct{} + +var errCacheDown = errors.New("cache is down") + +func (failingCache) Get(context.Context, string) (casexistence.Entry, bool, error) { + return casexistence.Entry{}, false, errCacheDown +} +func (failingCache) Set(context.Context, string, casexistence.Entry) error { return errCacheDown } +func (failingCache) Delete(context.Context, string) error { return errCacheDown } +func (failingCache) Purge(context.Context) error { return errCacheDown } + +func TestExistenceCacheKey(t *testing.T) { + base := casexistence.Key("org", "OCI", "secret", "digest") + require.NotContains(t, base, "secret", "the secret reference must be hashed") + for _, other := range []string{ + casexistence.Key("org2", "OCI", "secret", "digest"), + casexistence.Key("org", "S3", "secret", "digest"), + casexistence.Key("org", "OCI", "secret2", "digest"), + casexistence.Key("org", "OCI", "secret", "digest2"), + } { + require.NotEqual(t, base, other) + } +} diff --git a/app/artifact-cas/internal/service/bytestream_test.go b/app/artifact-cas/internal/service/bytestream_test.go index 9eda0b4db..2c10d6dc0 100644 --- a/app/artifact-cas/internal/service/bytestream_test.go +++ b/app/artifact-cas/internal/service/bytestream_test.go @@ -425,6 +425,8 @@ type bytestreamSuite struct { srv *grpc.Server client bytestream.ByteStreamClient ociBackend *mocks.UploaderDownloader + // ociProvider builds ociBackend; tests assert how often credentials are loaded + ociProvider *mocks.Provider // streamingBackend is the backend reachable via the "backend-streaming" // metadata. streamingBackend *mocks.UploaderDownloader @@ -455,6 +457,11 @@ func (s *bytestreamSuite) TearDownSubTest() { } func (s *bytestreamSuite) SetupTest() { + s.setup() +} + +// setup starts the server with the default options plus opts +func (s *bytestreamSuite) setup(opts ...NewOpt) { const backendType = "backend-type" // 1 MB buffer l := bufconn.Listen(1 << 20) @@ -477,6 +484,9 @@ func (s *bytestreamSuite) SetupTest() { if v := md.Get("backend-streaming"); len(v) > 0 { claims.BackendType = streamingBackendType } + if v := md.Get("org-id"); len(v) > 0 { + claims.OrgID = v[0] + } if v := md.Get("source-internal"); len(v) > 0 { claims.SourceInternal = true } @@ -518,7 +528,7 @@ func (s *bytestreamSuite) SetupTest() { NewByteStreamService(backend.Providers{ backendType: ociBackendProvider, streamingBackendType: streamingBackendProvider, - }, WithLogger(log.DefaultLogger), WithAuditDispatcher(newTestDispatcher(s.audit)), WithStagingDir(s.stagingDir)), + }, append([]NewOpt{WithLogger(log.DefaultLogger), WithAuditDispatcher(newTestDispatcher(s.audit)), WithStagingDir(s.stagingDir)}, opts...)...), ) go func() { _ = server.Serve(l) @@ -534,6 +544,7 @@ func (s *bytestreamSuite) SetupTest() { s.srv = server s.conn = conn s.ociBackend = ociBackend + s.ociProvider = ociBackendProvider s.streamingBackend = streamingBackend s.client = bytestream.NewByteStreamClient(conn) s.resource = &v1.CASResource{ diff --git a/app/artifact-cas/internal/service/cache.go b/app/artifact-cas/internal/service/cache.go new file mode 100644 index 000000000..2ae16dcce --- /dev/null +++ b/app/artifact-cas/internal/service/cache.go @@ -0,0 +1,147 @@ +// +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service + +import ( + "context" + "strings" + "time" + + casJWT "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" + backend "github.com/chainloop-dev/chainloop/pkg/blobmanager" + "github.com/chainloop-dev/chainloop/pkg/cache" + "github.com/chainloop-dev/chainloop/pkg/cache/casexistence" + "github.com/hashicorp/golang-lru/v2/expirable" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +// maxCachedClients bounds the number of backend clients kept in memory +const maxCachedClients = 1000 + +var ( + existenceCacheLookups = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "chainloop_cas_existence_cache_lookups_total", + Help: "Lookups in the cache of blobs known to exist in a backend, by result (hit or miss).", + }, []string{"result"}) + + clientCacheLookups = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "chainloop_cas_backend_client_cache_lookups_total", + Help: "Lookups in the cache of loaded backend clients, by result (hit or miss).", + }, []string{"result"}) +) + +func recordLookup(c *prometheus.CounterVec, hit bool) { + if hit { + c.WithLabelValues("hit").Inc() + } else { + c.WithLabelValues("miss").Inc() + } +} + +// WithExistenceCache enables the cache of blobs known to exist in a backend. +// Uploads of a cached digest end as "already exists" with no backend call. +func WithExistenceCache(c cache.Cache[casexistence.Entry]) NewOpt { + return func(s *commonService) { + s.existence = c + } +} + +// WithBackendClientCache enables the reuse of the backend clients loaded for +// uploads for ttl. Clients and their credentials are kept in process memory only. +// The services built with the same option share one cache. +func WithBackendClientCache(ttl time.Duration) NewOpt { + clients := expirable.NewLRU[string, backend.UploaderDownloader](maxCachedClients, nil, ttl) + return func(s *commonService) { + s.clients = clients + } +} + +// existenceKey returns the existence cache key of digest in the namespace of +// the organization and backend of the token, or "" when the upload can not be cached. +func (s *commonService) existenceKey(info *casJWT.Claims, digest string) string { + if s.existence == nil || info.OrgID == "" { + return "" + } + + return casexistence.Key(info.OrgID, info.BackendType, info.StoredSecretID, digest) +} + +// lookupExistence reports if key is cached as present. Cache errors are a miss. +func (s *commonService) lookupExistence(ctx context.Context, key string) (casexistence.Entry, bool) { + if key == "" { + return casexistence.Entry{}, false + } + + entry, ok, err := s.existence.Get(ctx, key) + if err != nil { + s.log.Warnw("msg", "existence cache lookup failed", "error", err) + ok = false + } + + recordLookup(existenceCacheLookups, ok) + return entry, ok +} + +// storeExistence caches that a blob of size is present. Best-effort. +func (s *commonService) storeExistence(ctx context.Context, key string, size int64) { + if key == "" { + return + } + + if err := s.existence.Set(ctx, key, casexistence.Entry{Size: size}); err != nil { + s.log.Warnw("msg", "existence cache store failed", "error", err) + } +} + +// clientKey identifies a backend client. It includes the organization because +// some backends (e.g. AWS-S3-ACCESS-POINT) keep per-organization session +// credentials inside the client. +func clientKey(info *casJWT.Claims) string { + return strings.Join([]string{info.BackendType, info.OrgID, info.StoredSecretID}, "\x00") +} + +// loadUploadBackend returns the backend of the token, reusing a cached client +// when the client cache is enabled. cached reports if the client was reused. +func (s *commonService) loadUploadBackend(ctx context.Context, info *casJWT.Claims) (b backend.UploaderDownloader, cached bool, err error) { + if s.clients == nil { + b, err = s.loadBackend(ctx, info.BackendType, info.StoredSecretID) + return b, false, err + } + + key := clientKey(info) + if b, ok := s.clients.Get(key); ok { + recordLookup(clientCacheLookups, true) + return b, true, nil + } + recordLookup(clientCacheLookups, false) + + b, err = s.loadBackend(ctx, info.BackendType, info.StoredSecretID) + if err != nil { + return nil, false, err + } + + s.clients.Add(key, b) + return b, false, nil +} + +// dropUploadBackend removes the cached client of the token, so the next upload +// loads the credentials again. +func (s *commonService) dropUploadBackend(info *casJWT.Claims) { + if s.clients != nil { + s.clients.Remove(clientKey(info)) + } +} diff --git a/app/artifact-cas/internal/service/service.go b/app/artifact-cas/internal/service/service.go index 8aff6e466..87e69f0e8 100644 --- a/app/artifact-cas/internal/service/service.go +++ b/app/artifact-cas/internal/service/service.go @@ -22,10 +22,13 @@ import ( "syscall" backend "github.com/chainloop-dev/chainloop/pkg/blobmanager" + "github.com/chainloop-dev/chainloop/pkg/cache" + "github.com/chainloop-dev/chainloop/pkg/cache/casexistence" "github.com/chainloop-dev/chainloop/pkg/servicelogger" kerrors "github.com/go-kratos/kratos/v2/errors" "github.com/go-kratos/kratos/v2/log" "github.com/google/wire" + "github.com/hashicorp/golang-lru/v2/expirable" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -44,6 +47,10 @@ type commonService struct { // set and writable: leaving it empty is a deployment error and transfers are // refused rather than staged somewhere unintended. stagingDir string + // existence caches the blobs known to exist in a backend, nil when disabled + existence cache.Cache[casexistence.Entry] + // clients caches the backend clients loaded for uploads, nil when disabled + clients *expirable.LRU[string, backend.UploaderDownloader] } func (s *commonService) loadBackend(ctx context.Context, providerType, secretID string) (backend.UploaderDownloader, error) { diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go index c30634c51..772915fb8 100644 --- a/pkg/cache/cache.go +++ b/pkg/cache/cache.go @@ -43,6 +43,7 @@ type Logger interface { type config struct { ttl time.Duration maxBytes int64 + maxEntries int replicas int logger Logger natsConn *nats.Conn @@ -78,6 +79,13 @@ func WithMaxBytes(n int64) Option { return func(c *config) { c.maxBytes = n } } +// WithMaxEntries sets the maximum number of entries of the in-memory backend. +// A value of 0 or less keeps the default of 1000. Ignored for NATS KV backend, +// which is bounded by WithMaxBytes. +func WithMaxEntries(n int) Option { + return func(c *config) { c.maxEntries = n } +} + // WithDescription sets the NATS KV bucket description. Ignored for in-memory backend. func WithDescription(desc string) Option { return func(c *config) { c.description = desc } diff --git a/pkg/cache/casexistence/casexistence.go b/pkg/cache/casexistence/casexistence.go new file mode 100644 index 000000000..6ba6bccaf --- /dev/null +++ b/pkg/cache/casexistence/casexistence.go @@ -0,0 +1,98 @@ +// +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package casexistence provides a typed cache of the blobs known to exist in +// a CAS backend, so repeated uploads of the same content skip the backend. +package casexistence + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "strings" + "time" + + "github.com/chainloop-dev/chainloop/pkg/cache" + "github.com/chainloop-dev/chainloop/pkg/natsconn" + "github.com/go-kratos/kratos/v2/log" +) + +const ( + // ttl bounds how long an entry outlives a purge of the blob outside Chainloop + ttl = 24 * time.Hour + maxBytes = 64 * 1024 * 1024 // 64 MB + maxEntries = 50_000 + bucket = "chainloop-cas-existence" + description = "Cache of blobs known to exist in a CAS backend" +) + +// Entry is a blob known to exist in a backend. +type Entry struct { + Size int64 `json:"size"` +} + +// Cache wraps cache.Cache[Entry] to provide a distinct type for wire disambiguation. +type Cache struct { + cache.Cache[Entry] +} + +// New creates an existence cache, backed by NATS KV when rc is set and by an +// in-memory LRU otherwise. A NATS bucket that can not be set up falls back to +// memory: the cache only makes uploads faster, so it must not stop its caller +// from starting. +func New(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (*Cache, error) { + c, err := newCache(ctx, rc, logger) + if err != nil && rc != nil { + if logger != nil { + log.NewHelper(logger).Warnw("msg", "existence cache: NATS KV unavailable, using memory", "error", err) + } + c, err = newCache(ctx, nil, logger) + } + + return c, err +} + +func newCache(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (*Cache, error) { + opts := []cache.Option{ + cache.WithTTL(ttl), + cache.WithMaxBytes(maxBytes), + cache.WithMaxEntries(maxEntries), + cache.WithDescription(description), + } + + if logger != nil { + opts = append(opts, cache.WithLogger(log.NewHelper(logger))) + } + + if rc != nil { + opts = append(opts, cache.WithNATS(rc.Conn, bucket)) + opts = append(opts, cache.WithReconnect(rc.Subscribe(ctx))) + opts = append(opts, cache.WithReplicas(rc.Replicas)) + } + + c, err := cache.New[Entry](opts...) + if err != nil { + return nil, err + } + return &Cache{Cache: c}, nil +} + +// Key returns the cache key of a digest in the namespace of one organization +// and one backend. The secret reference is hashed so its path never appears in +// the shared store. +func Key(orgID, backendType, secretID, digest string) string { + secretHash := sha256.Sum256([]byte(secretID)) + return strings.Join([]string{orgID, backendType, hex.EncodeToString(secretHash[:]), digest}, "/") +} diff --git a/pkg/cache/casexistence/casexistence_test.go b/pkg/cache/casexistence/casexistence_test.go new file mode 100644 index 000000000..936ff2490 --- /dev/null +++ b/pkg/cache/casexistence/casexistence_test.go @@ -0,0 +1,84 @@ +// +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package casexistence + +import ( + "context" + "testing" + "time" + + "github.com/chainloop-dev/chainloop/pkg/natsconn" + natsserver "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func startNATS(t *testing.T, jetStream bool) *nats.Conn { + t.Helper() + ns, err := natsserver.NewServer(&natsserver.Options{ + Host: "127.0.0.1", + Port: -1, + JetStream: jetStream, + StoreDir: t.TempDir(), + }) + require.NoError(t, err) + ns.Start() + t.Cleanup(ns.Shutdown) + + require.True(t, ns.ReadyForConnections(5*time.Second)) + nc, err := nats.Connect(ns.ClientURL()) + require.NoError(t, err) + t.Cleanup(nc.Close) + return nc +} + +// TestNew: the cache uses NATS KV when it can, and falls back to memory when +// the bucket can not be set up, so its caller still starts. +func TestNew(t *testing.T) { + testCases := []struct { + name string + jetStream bool + wantBucket bool + }{ + {name: "NATS KV when JetStream is available", jetStream: true, wantBucket: true}, + {name: "memory when the bucket can not be set up", jetStream: false}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + ctx := t.Context() + nc := startNATS(t, tc.jetStream) + + c, err := New(ctx, &natsconn.ReloadableConnection{Conn: nc}, nil) + require.NoError(t, err) + + require.NoError(t, c.Set(ctx, "key", Entry{Size: 42})) + got, ok, err := c.Get(ctx, "key") + require.NoError(t, err) + assert.True(t, ok) + assert.Equal(t, int64(42), got.Size) + + if tc.wantBucket { + js, err := jetstream.New(nc) + require.NoError(t, err) + _, err = js.KeyValue(context.Background(), bucket) + assert.NoError(t, err, "the cache must be stored in its own NATS bucket") + } + }) + } +} diff --git a/pkg/cache/memory.go b/pkg/cache/memory.go index ba907a6cc..ab54bf786 100644 --- a/pkg/cache/memory.go +++ b/pkg/cache/memory.go @@ -31,9 +31,14 @@ type memoryCache[T any] struct { } func newMemoryCache[T any](cfg *config) *memoryCache[T] { - cfg.logger.Infow("msg", "cache: using in-memory LRU backend", "ttl", cfg.ttl, "maxSize", defaultMaxSize) + maxSize := defaultMaxSize + if cfg.maxEntries > 0 { + maxSize = cfg.maxEntries + } + + cfg.logger.Infow("msg", "cache: using in-memory LRU backend", "ttl", cfg.ttl, "maxSize", maxSize) return &memoryCache[T]{ - lru: expirable.NewLRU[string, T](defaultMaxSize, nil, cfg.ttl), + lru: expirable.NewLRU[string, T](maxSize, nil, cfg.ttl), logger: cfg.logger, } } diff --git a/pkg/cache/memory_test.go b/pkg/cache/memory_test.go index 80abf6e74..261279cf3 100644 --- a/pkg/cache/memory_test.go +++ b/pkg/cache/memory_test.go @@ -17,6 +17,7 @@ package cache import ( "context" + "strconv" "testing" "time" @@ -105,3 +106,25 @@ func TestNew_RequiresTTL(t *testing.T) { _, err := New[string]() require.Error(t, err) } + +func TestNew_MaxEntries(t *testing.T) { + tests := []struct { + name string + opts []Option + wantStored int + }{ + {name: "default", wantStored: defaultMaxSize}, + {name: "custom", opts: []Option{WithMaxEntries(2)}, wantStored: 2}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c, err := New[int](append(tt.opts, WithTTL(time.Minute))...) + require.NoError(t, err) + for i := range defaultMaxSize + 1 { + require.NoError(t, c.Set(context.Background(), strconv.Itoa(i), i)) + } + assert.Equal(t, tt.wantStored, c.(*memoryCache[int]).lru.Len()) + }) + } +}