feat: add cache isolation (#1213)

Signed-off-by: Binbin Li <[email protected]>
Co-authored-by: Susan Shi <[email protected]>

httpserver/handlers.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/deislabs/ratify/errors"
+	ctxUtils "github.com/deislabs/ratify/internal/context"
 	"github.com/deislabs/ratify/internal/logger"
 	"github.com/deislabs/ratify/pkg/cache"
 	"github.com/deislabs/ratify/pkg/executor"
@@ -69,7 +70,7 @@ func (server *Server) verify(ctx context.Context, w http.ResponseWriter, r *http
 	// iterate over all keys
 	for _, key := range providerRequest.Request.Keys {
 		wg.Add(1)
-		go func(key string) {
+		go func(key string, ctx context.Context) {
 			defer wg.Done()
 			routineStartTime := time.Now()
 			returnItem := externaldata.Item{
@@ -90,6 +91,8 @@ func (server *Server) verify(ctx context.Context, w http.ResponseWriter, r *http
 				returnItem.Error = err.Error()
 				return
 			}
+			ctx = ctxUtils.SetContextWithNamespace(ctx, requestKey.Namespace)
+
 			if subjectReference.Digest.String() == "" {
 				logger.GetLogger(ctx, server.LogOption).Warn("Digest should be used instead of tagged reference. The resolved digest may not point to the same signed artifact, since tags are mutable.")
 			}
@@ -139,7 +142,7 @@ func (server *Server) verify(ctx context.Context, w http.ResponseWriter, r *http
 
 			returnItem.Value = fromVerifyResult(result, server.GetExecutor().PolicyEnforcer.GetPolicyType(ctx))
 			logger.GetLogger(ctx, server.LogOption).Debugf("verification: execution time for image %s: %dms", resolvedSubjectReference, time.Since(routineStartTime).Milliseconds())
-		}(utils.SanitizeString(key))
+		}(utils.SanitizeString(key), ctx)
 	}
 	wg.Wait()
 	elapsedTime := time.Since(startTime).Milliseconds()

internal/context/utils.go

@@ -0,0 +1,44 @@
+/*
+Copyright The Ratify 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 context
+
+import (
+	"context"
+	"fmt"
+)
+
+type contextKey string
+
+const contextKeyNamespace = contextKey("namespace")
+
+// SetContextWithNamespace embeds namespace to the context.
+func SetContextWithNamespace(ctx context.Context, namespace string) context.Context {
+	return context.WithValue(ctx, contextKeyNamespace, namespace)
+}
+
+// CreateCacheKey creates a new cache key prefixed with embedded namespace.
+func CreateCacheKey(ctx context.Context, key string) string {
+	namespace := ctx.Value(contextKeyNamespace)
+	if namespace == nil {
+		return key
+	}
+
+	namespaceStr := namespace.(string)
+	if namespaceStr == "" {
+		return key
+	}
+	return fmt.Sprintf("%s:%s", namespaceStr, key)
+}

internal/context/utils_test.go

@@ -0,0 +1,80 @@
+/*
+Copyright The Ratify 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 context
+
+import (
+	"context"
+	"testing"
+)
+
+const (
+	testNamespace = "testNamespace"
+	testKey       = "testKey"
+)
+
+func TestSetContext(t *testing.T) {
+	ctx := context.Background()
+	ctx = SetContextWithNamespace(ctx, testNamespace)
+	namespace := ctx.Value(contextKeyNamespace).(string)
+	if namespace != testNamespace {
+		t.Fatalf("expected namespace %s, got %s", testNamespace, namespace)
+	}
+}
+
+func TestCreateCacheKey(t *testing.T) {
+	testCases := []struct {
+		name         string
+		key          string
+		namespaceSet bool
+		namespace    string
+		expectedKey  string
+	}{
+		{
+			name:         "no namespace",
+			key:          testKey,
+			namespaceSet: false,
+			expectedKey:  testKey,
+		},
+		{
+			name:         "empty namespace",
+			key:          testKey,
+			namespaceSet: true,
+			namespace:    "",
+			expectedKey:  testKey,
+		},
+		{
+			name:         "with namespace",
+			key:          testKey,
+			namespaceSet: true,
+			namespace:    testNamespace,
+			expectedKey:  "testNamespace:testKey",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			if tc.namespaceSet {
+				ctx = SetContextWithNamespace(ctx, tc.namespace)
+			}
+
+			key := CreateCacheKey(ctx, tc.key)
+			if key != tc.expectedKey {
+				t.Fatalf("expected key %s, got %s", tc.expectedKey, key)
+			}
+		})
+	}
+}

pkg/cache/dapr/dapr.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/dapr/go-sdk/client"
+	ctxUtils "github.com/deislabs/ratify/internal/context"
 	"github.com/deislabs/ratify/internal/logger"
 	"github.com/deislabs/ratify/pkg/cache"
 	"github.com/deislabs/ratify/pkg/featureflag"
@@ -61,7 +62,7 @@ func (factory *factory) Create(_ context.Context, cacheName string, _ int) (cach
 }
 
 func (d *daprCache) Get(ctx context.Context, key string) (string, bool) {
-	item, err := d.daprClient.GetState(ctx, d.cacheName, key, nil)
+	item, err := d.daprClient.GetState(ctx, d.cacheName, ctxUtils.CreateCacheKey(ctx, key), nil)
 	if err != nil {
 		return "", false
 	}
@@ -74,7 +75,7 @@ func (d *daprCache) Set(ctx context.Context, key string, value interface{}) bool
 		logger.GetLogger(ctx, logOpt).Error("Error marshalling value for redis: ", err)
 		return false
 	}
-	if err := d.daprClient.SaveState(ctx, d.cacheName, key, bytes, nil); err != nil {
+	if err := d.daprClient.SaveState(ctx, d.cacheName, ctxUtils.CreateCacheKey(ctx, key), bytes, nil); err != nil {
 		logger.GetLogger(ctx, logOpt).Error("Error saving value to redis: ", err)
 		return false
 	}
@@ -93,15 +94,15 @@ func (d *daprCache) SetWithTTL(ctx context.Context, key string, value interface{
 	}
 	ttlString := strconv.Itoa(int(ttl.Seconds()))
 	md := map[string]string{"ttlInSeconds": ttlString}
-	if err := d.daprClient.SaveState(ctx, d.cacheName, key, bytes, md); err != nil {
+	if err := d.daprClient.SaveState(ctx, d.cacheName, ctxUtils.CreateCacheKey(ctx, key), bytes, md); err != nil {
 		logger.GetLogger(ctx, logOpt).Error("Error saving value to redis: ", err)
 		return false
 	}
 	return true
 }
 
 func (d *daprCache) Delete(ctx context.Context, key string) bool {
-	if err := d.daprClient.DeleteState(ctx, d.cacheName, key, nil); err != nil {
+	if err := d.daprClient.DeleteState(ctx, d.cacheName, ctxUtils.CreateCacheKey(ctx, key), nil); err != nil {
 		logger.GetLogger(ctx, logOpt).Error("Error deleting value from redis: ", err)
 		return false
 	}

pkg/cache/ristretto/ristretto.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/cespare/xxhash/v2"
+	ctxUtils "github.com/deislabs/ratify/internal/context"
 	"github.com/deislabs/ratify/internal/logger"
 	"github.com/deislabs/ratify/pkg/cache"
 	"github.com/dgraph-io/ristretto"
@@ -67,8 +68,8 @@ func (f *factory) Create(ctx context.Context, _ string, cacheSize int) (cache.Ca
 	}, nil
 }
 
-func (r *ristrettoCache) Get(_ context.Context, key string) (string, bool) {
-	cacheValue, found := r.memoryCache.Get(key)
+func (r *ristrettoCache) Get(ctx context.Context, key string) (string, bool) {
+	cacheValue, found := r.memoryCache.Get(ctxUtils.CreateCacheKey(ctx, key))
 	if !found {
 		return "", false
 	}
@@ -82,7 +83,7 @@ func (r *ristrettoCache) Set(ctx context.Context, key string, value interface{})
 		logger.GetLogger(ctx, logOpt).Error("Error marshalling value for ristretto: ", err)
 		return false
 	}
-	return r.memoryCache.Set(key, string(bytes), 1)
+	return r.memoryCache.Set(ctxUtils.CreateCacheKey(ctx, key), string(bytes), 1)
 }
 
 func (r *ristrettoCache) SetWithTTL(ctx context.Context, key string, value interface{}, ttl time.Duration) bool {
@@ -95,11 +96,11 @@ func (r *ristrettoCache) SetWithTTL(ctx context.Context, key string, value inter
 		logger.GetLogger(ctx, logOpt).Error("Error marshalling value for ristretto: ", err)
 		return false
 	}
-	return r.memoryCache.SetWithTTL(key, string(bytes), 1, ttl)
+	return r.memoryCache.SetWithTTL(ctxUtils.CreateCacheKey(ctx, key), string(bytes), 1, ttl)
 }
 
-func (r *ristrettoCache) Delete(_ context.Context, key string) bool {
-	r.memoryCache.Del(key)
+func (r *ristrettoCache) Delete(ctx context.Context, key string) bool {
+	r.memoryCache.Del(ctxUtils.CreateCacheKey(ctx, key))
 	// Note: ristretto does not return a bool for delete.
 	// Delete ops are eventually consistent and we don't want to block on them.
 	return true