Repository: deislabs/ratify (go)

Issue:
Add Request Cache Lock
**Describe the solution you'd like**
Ratify should implement a request cache lock for the http server so that the simultaneous requests for the same subject will not require redundant verification operations concurrently that will result in large amount of unnecessary calls to registry. The lock should be dimensioned by reference digest.

Example Scenario:

- 200 pod creation requests received at once all with a single container with the same image digest
- 200 External Data calls from Gatekeeper. This image is new so cache is cold and will currently generate 200 separate verification processes. 
- 1 of 200 requests grabs the lock and 199 wait on lock
- 1 request completes and updates entity/cache (maybe the verification result sits in the locked entity and is updated by single process on finish)
- 199 requests read from entity/cache and immediately return. 


Pull request:
feat: add request cache lock for verification
Signed-off-by: Anish Ramasekar <anish.ramasekar@gmail.com>

# Description

## What this PR does / why we need it:
<!--Please include a summary of the changes and relevant context. -->
- Adds request cache lock to enable processing verification once per subject in case of concurrent requests. The verification results are cached with a 5min TTL and refreshed if the cache entry is expired at read time. 

## Which issue(s) this PR fixes *(optional, using `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close the issue(s) when the PR gets merged)*:
Fixes #507 

## Type of change

Please delete options that are not relevant.

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

# How Has This Been Tested?

Please describe the tests that you ran to verify your changes. Please also list any relevant details for your test configuration

- [ ] Test A
- [ ] Test B

# Checklist:

- [ ] Does the affected code have corresponding tests?
- [ ] Are the changes documented, not just with inline documentation, but also with conceptual documentation such as an overview of a new feature, or task-based documentation like a tutorial? Consider if this change should be announced on your project blog.
- [ ] Does this introduce breaking changes that would require an announcement or bumping the major version?
- [ ] Do all new files have appropriate license header?


Patch:
diff --git a/charts/ratify/templates/deployment.yaml b/charts/ratify/templates/deployment.yaml
index 3420d0b..b9f0558 100644
--- a/charts/ratify/templates/deployment.yaml
+++ b/charts/ratify/templates/deployment.yaml
@@ -48,6 +48,8 @@ spec:
             {{- if (lookup "v1" "Secret" .Release.Namespace "gatekeeper-webhook-server-cert") }}
             - --ca-cert-file=usr/local/tls/client-ca/ca.crt
             {{- end }}
+            - --cache-size={{ .Values.provider.cache.size }}
+            - --cache-ttl={{ .Values.provider.cache.ttl }}
           ports:
             - containerPort: 6001
           volumeMounts:
@@ -113,11 +115,11 @@ spec:
       volumes:
         - name: notary-certs
           secret:
-            secretName: {{ include "ratify.fullname" . }}-notary-certificate         
+            secretName: {{ include "ratify.fullname" . }}-notary-certificate
         {{- if .Values.cosign.enabled }}
         - name: cosign-certs
           secret:
-            secretName: {{ include "ratify.fullname" . }}-cosign-certificate         
+            secretName: {{ include "ratify.fullname" . }}-cosign-certificate
         {{- end }}
         {{- if $dockerAuthMode }}
         - name: dockerconfig
diff --git a/charts/ratify/values.yaml b/charts/ratify/values.yaml
index 71b9846..510e7f4 100644
--- a/charts/ratify/values.yaml
+++ b/charts/ratify/values.yaml
@@ -95,6 +95,9 @@ provider:
     # timeout values must match gatekeeper webhook timeouts
     validationTimeoutSeconds: 5
     mutationTimeoutSeconds: 2
+  cache:
+    ttl: 10s  # cache ttl duration
+    size: 100 # number of items to be kept in cache. Set to 0 to disable cache.
 
 podAnnotations: {}
 podLabels: {}
@@ -102,7 +105,7 @@ enableRuntimeDefaultSeccompProfile: true
 
 rbac:
   create: true
-  
+
 upgradeCRDs:
   enabled: true
   extraRules: []
@@ -125,4 +128,3 @@ crds:
 # See https://github.com/deislabs/ratify/blob/main/docs/reference/usage.md for a list of available feature flags
 featureFlags:
   # RATIFY_FEATURE_NAME: true
-
diff --git a/cmd/ratify/cmd/serve.go b/cmd/ratify/cmd/serve.go
index a7071b8..1de330b 100644
--- a/cmd/ratify/cmd/serve.go
+++ b/cmd/ratify/cmd/serve.go
@@ -17,6 +17,7 @@ package cmd
 
 import (
 	"context"
+	"time"
 
 	"github.com/deislabs/ratify/config"
 	"github.com/deislabs/ratify/httpserver"
@@ -35,6 +36,8 @@ type serveCmdOptions struct {
 	certDirectory     string
 	caCertFile        string
 	enableCrdManager  bool
+	cacheSize         int
+	cacheTTL          time.Duration
 }
 
 func NewCmdServe(argv ...string) *cobra.Command {
@@ -57,6 +60,8 @@ func NewCmdServe(argv ...string) *cobra.Command {
 	flags.StringVar(&opts.certDirectory, "cert-dir", "", "Path to ratify certs")
 	flags.StringVar(&opts.caCertFile, "ca-cert-file", "", "Path to CA cert file")
 	flags.BoolVar(&opts.enableCrdManager, "enable-crd-manager", false, "Start crd manager if enabled")
+	flags.IntVar(&opts.cacheSize, "cache-size", httpserver.DefaultCacheMaxSize, "Cache size for the verifier http server")
+	flags.DurationVar(&opts.cacheTTL, "cache-ttl", httpserver.DefaultCacheTTL, "Cache TTL for the verifier http server")
 	return cmd
 }
 
@@ -65,7 +70,7 @@ func serve(opts serveCmdOptions) error {
 	if opts.enableCrdManager {
 		logrus.Infof("starting crd manager")
 		go manager.StartManager()
-		manager.StartServer(opts.httpServerAddress, opts.configFilePath, opts.certDirectory, opts.caCertFile)
+		manager.StartServer(opts.httpServerAddress, opts.configFilePath, opts.certDirectory, opts.caCertFile, opts.cacheSize, opts.cacheTTL)
 
 		return nil
 	}
@@ -76,7 +81,7 @@ func serve(opts serveCmdOptions) error {
 	}
 
 	if opts.httpServerAddress != "" {
-		server, err := httpserver.NewServer(context.Background(), opts.httpServerAddress, getExecutor, opts.certDirectory, opts.caCertFile)
+		server, err := httpserver.NewServer(context.Background(), opts.httpServerAddress, getExecutor, opts.certDirectory, opts.caCertFile, opts.cacheSize, opts.cacheTTL)
 		if err != nil {
 			return err
 		}
diff --git a/httpserver/cache.go b/httpserver/cache.go
new file mode 100644
index 0000000..b936963
--- /dev/null
+++ b/httpserver/cache.go
@@ -0,0 +1,50 @@
+/*
+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 httpserver
+
+import (
+	"time"
+
+	"github.com/deislabs/ratify/pkg/executor/types"
+	utilcache "k8s.io/apimachinery/pkg/util/cache"
+)
+
+type simpleCache struct {
+	cache   *utilcache.LRUExpireCache
+	ttl     time.Duration
+	maxSize int
+}
+
+func newSimpleCache(ttl time.Duration, maxSize int) *simpleCache {
+	return &simpleCache{
+		cache:   utilcache.NewLRUExpireCache(maxSize),
+		ttl:     ttl,
+		maxSize: maxSize,
+	}
+}
+
+// given a key, return the item from the cache if it exists and has not expired
+func (c *simpleCache) get(key string) *types.VerifyResult {
+	if item, ok := c.cache.Get(key); ok {
+		return item.(*types.VerifyResult)
+	}
+	return nil
+}
+
+// given a key, set the item in the cache
+func (c *simpleCache) set(key string, item *types.VerifyResult) {
+	c.cache.Add(key, item, c.ttl)
+}
diff --git a/httpserver/handlers.go b/httpserver/handlers.go
index 293a1fa..1758743 100644
--- a/httpserver/handlers.go
+++ b/httpserver/handlers.go
@@ -24,10 +24,12 @@ import (
 	"sync"
 	"time"
 
-	e "github.com/deislabs/ratify/pkg/executor"
+	"github.com/deislabs/ratify/pkg/executor"
+	"github.com/deislabs/ratify/pkg/executor/types"
 	"github.com/deislabs/ratify/pkg/referrerstore"
 	pkgUtils "github.com/deislabs/ratify/pkg/utils"
 	"github.com/deislabs/ratify/utils"
+
 	"github.com/open-policy-agent/frameworks/constraint/pkg/externaldata"
 	"github.com/sirupsen/logrus"
 )
@@ -44,11 +46,11 @@ func (server *Server) verify(ctx context.Context, w http.ResponseWriter, r *http
 	if err != nil {
 		return fmt.Errorf("unable to read request body: %w", err)
 	}
+	defer r.Body.Close()
 
 	// parse request body
 	var providerRequest externaldata.ProviderRequest
-	err = json.Unmarshal(body, &providerRequest)
-	if err != nil {
+	if err = json.Unmarshal(body, &providerRequest); err != nil {
 		return fmt.Errorf("unable to unmarshal request body: %w", err)
 	}
 
@@ -62,25 +64,36 @@ func (server *Server) verify(ctx context.Context, w http.ResponseWriter, r *http
 		go func(subject string) {
 			defer wg.Done()
 			routineStartTime := time.Now()
-			// TODO: Enable caching:  Providers should add a caching mechanism to avoid extra calls to external data sources.
+
+			unlock := server.keyMutex.Lock(subject)
+			defer unlock()
+
 			logrus.Infof("verifying subject %v", subject)
-			returnItem := externaldata.Item{
-				Key: subject,
-			}
+			var result types.VerifyResult
+			res := server.cache.get(subject)
+			if res != nil {
+				logrus.Debugf("cache hit for subject %v", subject)
+				result = *res
+			} else {
+				logrus.Debugf("cache miss for subject %v", subject)
+				returnItem := externaldata.Item{
+					Key: subject,
+				}
 
-			verifyParameters := e.VerifyParameters{
-				Subject: subject,
-			}
+				verifyParameters := executor.VerifyParameters{
+					Subject: subject,
+				}
 
-			result, err := server.GetExecutor().VerifySubject(ctx, verifyParameters)
-			if err != nil {
-				returnItem.Error = err.Error()
-			}
+				if result, err = server.GetExecutor().VerifySubject(ctx, verifyParameters); err != nil {
+					returnItem.Error = err.Error()
+				}
+				server.cache.set(subject, &result)
 
-			res, err := json.MarshalIndent(result, "", "  ")
-			if err == nil {
-				fmt.Println(string(res))
+				if res, err := json.MarshalIndent(result, "", "  "); err == nil {
+					fmt.Println(string(res))
+				}
 			}
+
 			mu.Lock()
 			defer mu.Unlock()
 			results = append(results, externaldata.Item{
diff --git a/httpserver/server.go b/httpserver/server.go
index 9accbbf..139664e 100644
--- a/httpserver/server.go
+++ b/httpserver/server.go
@@ -25,9 +25,11 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"sync"
 	"time"
 
 	"github.com/deislabs/ratify/config"
+
 	"github.com/gorilla/mux"
 	"github.com/sirupsen/logrus"
 )
@@ -38,21 +40,43 @@ const (
 	keyName                          = "tls.key"
 	readHeaderTimeout                = 5 * time.Second
 	defaultMutationReferrerStoreName = "oras"
+
+	// DefaultCacheTTL is the default time-to-live for the cache entry.
+	DefaultCacheTTL = 10 * time.Second
+	// DefaultCacheMaxSize is the default maximum size of the cache.
+	DefaultCacheMaxSize = 100
 )
 
-type (
-	Server struct {
-		Address           string
-		Router            *mux.Router
-		GetExecutor       config.GetExecutor
-		Context           context.Context
-		CertDirectory     string
-		CaCertFile        string
-		MutationStoreName string
+type Server struct {
+	Address           string
+	Router            *mux.Router
+	GetExecutor       config.GetExecutor
+	Context           context.Context
+	CertDirectory     string
+	CaCertFile        string
+	MutationStoreName string
+
+	keyMutex keyMutex
+	// cache is a thread-safe expiring lru cache which caches external data item indexed
+	// by the subject
+	cache *simpleCache
+}
+
+// keyMutex is a thread-safe map of mutexes, indexed by key.
+type keyMutex struct {
+	locks sync.Map
+}
+
+// Lock locks the mutex for the given key, and returns a function to unlock it.
+func (m *keyMutex) Lock(key string) func() {
+	v, _ := m.locks.LoadOrStore(key, &sync.Mutex{})
+	v.(*sync.Mutex).Lock()
+	return func() {
+		v.(*sync.Mutex).Unlock()
 	}
-)
+}
 
-func NewServer(context context.Context, address string, getExecutor config.GetExecutor, certDir string, caCertFile string) (*Server, error) {
+func NewServer(context context.Context, address string, getExecutor config.GetExecutor, certDir, caCertFile string, cacheSize int, cacheTTL time.Duration) (*Server, error) {
 	if address == "" {
 		return nil, ServerAddrNotFoundError{}
 	}
@@ -65,6 +89,8 @@ func NewServer(context context.Context, address string, getExecutor config.GetEx
 		CertDirectory:     certDir,
 		CaCertFile:        caCertFile,
 		MutationStoreName: defaultMutationReferrerStoreName,
+		keyMutex:          keyMutex{},
+		cache:             newSimpleCache(cacheTTL, cacheSize),
 	}
 
 	return server, server.registerHandlers()
diff --git a/pkg/manager/manager.go b/pkg/manager/manager.go
index 7fa4bc9..c8cf1b4 100644
--- a/pkg/manager/manager.go
+++ b/pkg/manager/manager.go
@@ -19,6 +19,7 @@ import (
 	"context"
 	"flag"
 	"os"
+	"time"
 
 	_ "github.com/deislabs/ratify/pkg/policyprovider/configpolicy"
 	_ "github.com/deislabs/ratify/pkg/referrerstore/oras"
@@ -61,7 +62,7 @@ func init() {
 	//+kubebuilder:scaffold:scheme
 }
 
-func StartServer(httpServerAddress string, configFilePath string, certDirectory string, caCertFile string) {
+func StartServer(httpServerAddress, configFilePath, certDirectory, caCertFile string, cacheSize int, cacheTTL time.Duration) {
 	logrus.Info("initializing executor with config file at default config path")
 
 	cf, err := config.Load(configFilePath)
@@ -71,7 +72,6 @@ func StartServer(httpServerAddress string, configFilePath string, certDirectory
 	}
 
 	configStores, configVerifiers, policy, err := config.CreateFromConfig(cf)
-
 	if err != nil {
 		logrus.Warnf("error initializing from config %v", err)
 		os.Exit(1)
@@ -110,7 +110,7 @@ func StartServer(httpServerAddress string, configFilePath string, certDirectory
 			Config:         &cf.ExecutorConfig,
 		}
 		return &executor
-	}, certDirectory, caCertFile)
+	}, certDirectory, caCertFile, cacheSize, cacheTTL)
 
 	if err != nil {
 		os.Exit(1)


Test patch:
diff --git a/httpserver/server_test.go b/httpserver/server_test.go
index c82d4a8..d3ed1e5 100644
--- a/httpserver/server_test.go
+++ b/httpserver/server_test.go
@@ -88,6 +88,9 @@ func TestServer_Timeout_Failed(t *testing.T) {
 		server := &Server{
 			GetExecutor: getExecutor,
 			Context:     request.Context(),
+
+			keyMutex: keyMutex{},
+			cache:    newSimpleCache(DefaultCacheTTL, DefaultCacheMaxSize),
 		}
 
 		handler := contextHandler{
@@ -157,6 +160,9 @@ func TestServer_MultipleSubjects_Success(t *testing.T) {
 		server := &Server{
 			GetExecutor: getExecutor,
 			Context:     request.Context(),
+
+			keyMutex: keyMutex{},
+			cache:    newSimpleCache(DefaultCacheTTL, DefaultCacheMaxSize),
 		}
 
 		handler := contextHandler{
@@ -228,6 +234,9 @@ func TestServer_Mutation_Success(t *testing.T) {
 			GetExecutor:       getExecutor,
 			Context:           request.Context(),
 			MutationStoreName: store.Name(),
+
+			keyMutex: keyMutex{},
+			cache:    newSimpleCache(DefaultCacheTTL, DefaultCacheMaxSize),
 		}
 
 		handler := contextHandler{
@@ -250,3 +259,79 @@ func TestServer_Mutation_Success(t *testing.T) {
 		}
 	})
 }
+
+func TestServer_MultipleRequestsForSameSubject_Success(t *testing.T) {
+	testImageNames := []string{"localhost:5000/net-monitor:v1", "localhost:5000/net-monitor:v1"}
+	t.Run("server_multiple_subjects_success", func(t *testing.T) {
+		body := new(bytes.Buffer)
+
+		if err := json.NewEncoder(body).Encode(externaldata.NewProviderRequest(testImageNames)); err != nil {
+			t.Fatalf("failed to encode request body: %v", err)
+		}
+		request := httptest.NewRequest(http.MethodPost, "/ratify/gatekeeper/v1/verify", bytes.NewReader(body.Bytes()))
+		logrus.Infof("policies successfully created. %s", body.Bytes())
+
+		responseRecorder := httptest.NewRecorder()
+
+		testDigest := digest.FromString("test")
+		configPolicy := config.PolicyEnforcer{
+			ArtifactTypePolicies: map[string]types.ArtifactTypeVerifyPolicy{
+				testArtifactType: types.AnyVerifySuccess,
+			}}
+		store := &mocks.TestStore{References: []ocispecs.ReferenceDescriptor{
+			{
+				ArtifactType: testArtifactType,
+			}},
+			ResolveMap: map[string]digest.Digest{
+				"v1": testDigest,
+				"v2": testDigest,
+			},
+			ExtraSubject: testImageNames[0],
+		}
+		ver := &core.TestVerifier{
+			CanVerifyFunc: func(at string) bool {
+				return at == testArtifactType
+			},
+			VerifyResult: func(artifactType string) bool {
+				return true
+			},
+		}
+
+		ex := &core.Executor{
+			PolicyEnforcer: configPolicy,
+			ReferrerStores: []referrerstore.ReferrerStore{store},
+			Verifiers:      []verifier.ReferenceVerifier{ver},
+			Config: &exconfig.ExecutorConfig{
+				VerificationRequestTimeout: nil,
+				MutationRequestTimeout:     nil,
+			},
+		}
+
+		getExecutor := func() *core.Executor {
+			return ex
+		}
+
+		server := &Server{
+			GetExecutor: getExecutor,
+			Context:     request.Context(),
+
+			keyMutex: keyMutex{},
+			cache:    newSimpleCache(DefaultCacheTTL, DefaultCacheMaxSize),
+		}
+
+		handler := contextHandler{
+			context: server.Context,
+			handler: processTimeout(server.verify, server.GetExecutor().GetVerifyRequestTimeout(), false),
+		}
+
+		handler.ServeHTTP(responseRecorder, request)
+		var respBody externaldata.ProviderResponse
+		if err := json.NewDecoder(responseRecorder.Result().Body).Decode(&respBody); err != nil {
+			t.Fatalf("failed to decode response body: %v", err)
+		}
+		retFirstKey := respBody.Response.Items[0].Key
+		if retFirstKey != testImageNames[1] {
+			t.Fatalf("Expected first subject response to be %s but got %s", testImageNames[1], retFirstKey)
+		}
+	})
+}
diff --git a/test/bats/test.bats b/test/bats/test.bats
index 4a8e788..54ce958 100644
--- a/test/bats/test.bats
+++ b/test/bats/test.bats
@@ -63,7 +63,8 @@ SLEEP_TIME=1
     assert_failure
 
     run kubectl apply -f ./config/samples/config_v1alpha1_verifier_complete_licensechecker.yaml
-    sleep 5
+    # wait for the httpserver cache to be invalidated
+    sleep 15
     run kubectl run license-checker2 --namespace default --image=wabbitnetworks.azurecr.io/test/license-checker-image:v1
     assert_success
 }
@@ -88,6 +89,8 @@ SLEEP_TIME=1
 
     run kubectl delete verifiers.config.ratify.deislabs.io/verifier-sbom
     assert_success
+    # wait for the httpserver cache to be invalidated
+    sleep 15
     run kubectl run sbom2 --namespace default --image=wabbitnetworks.azurecr.io/test/sbom-image:signed
     assert_failure
 }
@@ -110,14 +113,16 @@ SLEEP_TIME=1
 
     run kubectl apply -f ./config/samples/config_v1alpha1_verifier_schemavalidator.yaml
     sleep 5
-    # TODO 
+    # TODO
     # It's best to use an image with individual artifact types vs an all-in-one so any failures can be isolated.
     # Replace this image reference once we have a local private registry for Ratify.
     run kubectl run schemavalidator --namespace default --image=wabbitnetworks.azurecr.io/test/all-in-one-image:signed
     assert_success
 
     run kubectl apply -f ./config/samples/config_v1alpha1_verifier_schemavalidator_bad.yaml
-    sleep 5
+    assert_success
+    # wait for the httpserver cache to be invalidated
+    sleep 15
     run kubectl run schemavalidator2 --namespace default --image=wabbitnetworks.azurecr.io/test/all-in-one-image:signed
     assert_failure
 }
@@ -145,6 +150,8 @@ SLEEP_TIME=1
     run kubectl apply -f ./config/samples/config_v1alpha1_verifier_schemavalidator.yaml
     sleep 5
 
+    # wait for the httpserver cache to be invalidated
+    sleep 15
     run kubectl run all-in-one --namespace default --image=wabbitnetworks.azurecr.io/test/all-in-one-image:signed
     assert_success
 }
@@ -160,6 +167,8 @@ SLEEP_TIME=1
     assert_success
     run kubectl delete verifiers.config.ratify.deislabs.io/verifier-notary
     assert_success
+    # wait for the httpserver cache to be invalidated
+    sleep 15
     run kubectl run crdtest --namespace default --image=wabbitnetworks.azurecr.io/test/notary-image:signed
     assert_failure
 
@@ -167,6 +176,8 @@ SLEEP_TIME=1
     run kubectl apply -f ./config/samples/config_v1alpha1_verifier_notary.yaml
     assert_success
 
+    # wait for the httpserver cache to be invalidated
+    sleep 15
     run kubectl run crdtest --namespace default --image=wabbitnetworks.azurecr.io/test/notary-image:signed
     assert_success
 }