Cache trigger secrets for the duration of request

This commit adds a request-local cache for interceptors to leverage
during the processing of triggers. It allows interceptors to avoid doing
expensive work more than once for each request, such as fetching a
Kubernetes secret for validating webhooks.

The implementation uses the request context to provide the cache. This
was the least disruptive method of providing a cache for use with
interceptors, and is appropriate if you consider the cache should live
only for the duration of each request.

Alternative implementations might have used the client-go informers to
extend the Kubernetes client to watch for secrets in the cluster. This
would cause the work required to fetch secrets to scale with the number
of secrets in the cluster, as opposed to making a fresh request per
webhook we process. That said, building caching clients seems like more
work than is necessary for fixing this simple problem, which is why I
went with a simple cache object.

The background for this change was finding Github webhooks timing out
once we exceeded ~40 triggers on our EventListener. While the CEL
filtering was super fast, the validation of Github webhook signatures
was being computed for every trigger, even though each trigger used the
same Github secret. Pulling the secret from Kubernetes was taking about
250ms, which meant 40 triggers exceeded the 10s Github timeout.

pkg/interceptors/bitbucket/bitbucket.go

@@ -64,7 +64,7 @@ func (w *Interceptor) ExecuteTrigger(request *http.Request) (*http.Response, err
 		if header == "" {
 			return nil, errors.New("no X-Hub-Signature header set")
 		}
-		secretToken, err := interceptors.GetSecretToken(w.KubeClientSet, w.Bitbucket.SecretRef, w.EventListenerNamespace)
+		secretToken, err := interceptors.GetSecretToken(request, w.KubeClientSet, w.Bitbucket.SecretRef, w.EventListenerNamespace)
 		if err != nil {
 			return nil, err
 		}

pkg/interceptors/cel/cel.go

@@ -60,7 +60,7 @@ func NewInterceptor(cel *triggersv1.CELInterceptor, k kubernetes.Interface, ns s
 
 // ExecuteTrigger is an implementation of the Interceptor interface.
 func (w *Interceptor) ExecuteTrigger(request *http.Request) (*http.Response, error) {
-	env, err := makeCelEnv(w.EventListenerNamespace, w.KubeClientSet)
+	env, err := makeCelEnv(request, w.EventListenerNamespace, w.KubeClientSet)
 	if err != nil {
 		return nil, fmt.Errorf("error creating cel environment: %w", err)
 	}
@@ -153,10 +153,10 @@ func evaluate(expr string, env *cel.Env, data map[string]interface{}) (ref.Val,
 	return out, err
 }
 
-func makeCelEnv(ns string, k kubernetes.Interface) (*cel.Env, error) {
+func makeCelEnv(request *http.Request, ns string, k kubernetes.Interface) (*cel.Env, error) {
 	mapStrDyn := decls.NewMapType(decls.String, decls.Dyn)
 	return cel.NewEnv(
-		Triggers(ns, k),
+		Triggers(request, ns, k),
 		celext.Strings(),
 		cel.Declarations(
 			decls.NewIdent("body", mapStrDyn, nil),

pkg/interceptors/cel/cel_test.go

@@ -431,7 +431,7 @@ func TestExpressionEvaluation(t *testing.T) {
 					rt.Error(err)
 				}
 			}
-			env, err := makeCelEnv(testNS, kubeClient)
+			env, err := makeCelEnv(&http.Request{}, testNS, kubeClient)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -540,7 +540,7 @@ func TestExpressionEvaluation_Error(t *testing.T) {
 				}
 				ns = tt.secretNS
 			}
-			env, err := makeCelEnv(ns, kubeClient)
+			env, err := makeCelEnv(&http.Request{}, ns, kubeClient)
 			if err != nil {
 				t.Fatal(err)
 			}

pkg/interceptors/cel/triggers.go

@@ -139,11 +139,12 @@ import (
 //     'https://example.com/testing'.parseURL().host == 'example.com'
 
 // Triggers creates and returns a new cel.Lib with the triggers extensions.
-func Triggers(ns string, k kubernetes.Interface) cel.EnvOption {
-	return cel.Lib(triggersLib{defaultNS: ns, client: k})
+func Triggers(request *http.Request, ns string, k kubernetes.Interface) cel.EnvOption {
+	return cel.Lib(triggersLib{request: request, defaultNS: ns, client: k})
 }
 
 type triggersLib struct {
+	request   *http.Request
 	defaultNS string
 	client    kubernetes.Interface
 }
@@ -201,7 +202,7 @@ func (t triggersLib) ProgramOptions() []cel.ProgramOption {
 				Unary:    parseURLString},
 			&functions.Overload{
 				Operator: "compareSecret",
-				Function: makeCompareSecret(t.defaultNS, t.client)},
+				Function: makeCompareSecret(t.request, t.defaultNS, t.client)},
 		)}
 }
 
@@ -266,7 +267,7 @@ func decodeB64String(val ref.Val) ref.Val {
 
 // makeCompareSecret creates and returns a functions.FunctionOp that wraps the
 // ns and client in a closure with a function that can compare the string.
-func makeCompareSecret(defaultNS string, k kubernetes.Interface) functions.FunctionOp {
+func makeCompareSecret(request *http.Request, defaultNS string, k kubernetes.Interface) functions.FunctionOp {
 	return func(vals ...ref.Val) ref.Val {
 		var ok bool
 		compareString, ok := vals[0].(types.String)
@@ -300,7 +301,7 @@ func makeCompareSecret(defaultNS string, k kubernetes.Interface) functions.Funct
 			SecretName: string(secretName),
 			Namespace:  string(secretNS),
 		}
-		secretToken, err := interceptors.GetSecretToken(k, secretRef, string(secretNS))
+		secretToken, err := interceptors.GetSecretToken(request, k, secretRef, string(secretNS))
 		if err != nil {
 			return types.NewErr("failed to find secret '%#v' in compareSecret: %w", *secretRef, err)
 		}

pkg/interceptors/github/github.go

@@ -64,7 +64,7 @@ func (w *Interceptor) ExecuteTrigger(request *http.Request) (*http.Response, err
 		if header == "" {
 			return nil, errors.New("no X-Hub-Signature header set")
 		}
-		secretToken, err := interceptors.GetSecretToken(w.KubeClientSet, w.GitHub.SecretRef, w.EventListenerNamespace)
+		secretToken, err := interceptors.GetSecretToken(request, w.KubeClientSet, w.GitHub.SecretRef, w.EventListenerNamespace)
 		if err != nil {
 			return nil, err
 		}

pkg/interceptors/gitlab/gitlab.go

@@ -54,7 +54,7 @@ func (w *Interceptor) ExecuteTrigger(request *http.Request) (*http.Response, err
 			return nil, errors.New("no X-GitLab-Token header set")
 		}
 
-		secretToken, err := interceptors.GetSecretToken(w.KubeClientSet, w.GitLab.SecretRef, w.EventListenerNamespace)
+		secretToken, err := interceptors.GetSecretToken(request, w.KubeClientSet, w.GitLab.SecretRef, w.EventListenerNamespace)
 		if err != nil {
 			return nil, err
 		}

pkg/interceptors/interceptors.go

@@ -17,7 +17,9 @@ limitations under the License.
 package interceptors
 
 import (
+	"context"
 	"net/http"
+	"path"
 
 	triggersv1 "github.com/tektoncd/triggers/pkg/apis/triggers/v1alpha1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -29,7 +31,46 @@ type Interceptor interface {
 	ExecuteTrigger(req *http.Request) (*http.Response, error)
 }
 
-func GetSecretToken(cs kubernetes.Interface, sr *triggersv1.SecretRef, eventListenerNamespace string) ([]byte, error) {
+type key string
+
+const requestCacheKey key = "interceptors.RequestCache"
+
+// WithCache clones the given request and sets the request context to include a cache.
+// This allows us to cache results from expensive operations and perform them just once
+// per each trigger.
+//
+// Each request should have its own cache, and those caches should expire once the request
+// is processed. For this reason, it's appropriate to store the cache on the request
+// context.
+func WithCache(req *http.Request) *http.Request {
+	return req.WithContext(context.WithValue(req.Context(), requestCacheKey, make(map[string]interface{})))
+}
+
+func getCache(req *http.Request) map[string]interface{} {
+	if cache, ok := req.Context().Value(requestCacheKey).(map[string]interface{}); ok {
+		return cache
+	}
+
+	return make(map[string]interface{})
+}
+
+// GetSecretToken queries Kubernetes for the given secret reference. We use this function
+// to resolve secret material like Github webhook secrets, and call it once for every
+// trigger that references it.
+//
+// As we may have many triggers that all use the same secret, we cache the secret values
+// in the request cache.
+func GetSecretToken(req *http.Request, cs kubernetes.Interface, sr *triggersv1.SecretRef, eventListenerNamespace string) ([]byte, error) {
+	var cache map[string]interface{}
+
+	cacheKey := path.Join("secret", sr.Namespace, sr.SecretName, sr.SecretKey)
+	if req != nil {
+		cache = getCache(req)
+		if secretValue, ok := cache[cacheKey]; ok {
+			return secretValue.([]byte), nil
+		}
+	}
+
 	ns := sr.Namespace
 	if ns == "" {
 		ns = eventListenerNamespace
@@ -39,5 +80,10 @@ func GetSecretToken(cs kubernetes.Interface, sr *triggersv1.SecretRef, eventList
 		return nil, err
 	}
 
-	return secret.Data[sr.SecretKey], nil
+	secretValue := secret.Data[sr.SecretKey]
+	if req != nil {
+		cache[cacheKey] = secret.Data[sr.SecretKey]
+	}
+
+	return secretValue, nil
 }

pkg/interceptors/interceptors_test.go

@@ -0,0 +1,102 @@
+/*
+Copyright 2019 The Tekton 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 interceptors
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	rtesting "knative.dev/pkg/reconciler/testing"
+
+	triggersv1 "github.com/tektoncd/triggers/pkg/apis/triggers/v1alpha1"
+)
+
+const testNS = "testing-ns"
+
+func Test_GetSecretToken(t *testing.T) {
+	tests := []struct {
+		name   string
+		cache  map[string]interface{}
+		wanted []byte
+	}{
+		{
+			name:   "no matching cache entry exists",
+			cache:  make(map[string]interface{}),
+			wanted: []byte("secret from API"),
+		},
+		{
+			name: "a matching cache entry exists",
+			cache: map[string]interface{}{
+				fmt.Sprintf("secret/%s/test-secret/token", testNS): []byte("secret from cache"),
+			},
+			wanted: []byte("secret from cache"),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(rt *testing.T) {
+			req := setCache(&http.Request{}, tt.cache)
+
+			ctx, _ := rtesting.SetupFakeContext(t)
+			kubeClient := fakekubeclient.Get(ctx)
+			secretRef := makeSecretRef()
+
+			if _, err := kubeClient.CoreV1().Secrets(testNS).Create(makeSecret("secret from API")); err != nil {
+				rt.Error(err)
+			}
+
+			secret, err := GetSecretToken(req, kubeClient, &secretRef, testNS)
+			if err != nil {
+				rt.Error(err)
+			}
+
+			if diff := cmp.Diff(tt.wanted, secret); diff != "" {
+				rt.Errorf("secret value (-want, +got) = %s", diff)
+			}
+		})
+	}
+}
+
+func makeSecret(secretText string) *corev1.Secret {
+	return &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: testNS,
+			Name:      "test-secret",
+		},
+		Data: map[string][]byte{
+			"token": []byte(secretText),
+		},
+	}
+}
+
+func makeSecretRef() triggersv1.SecretRef {
+	return triggersv1.SecretRef{
+		SecretKey:  "token",
+		SecretName: "test-secret",
+		Namespace:  testNS,
+	}
+}
+
+func setCache(req *http.Request, vals map[string]interface{}) *http.Request {
+	return req.WithContext(context.WithValue(req.Context(), requestCacheKey, vals))
+}

pkg/sink/sink.go

@@ -187,6 +187,11 @@ func (r Sink) executeInterceptors(t *triggersv1.EventListenerTrigger, in *http.R
 		Header: in.Header,
 		Body:   ioutil.NopCloser(bytes.NewBuffer(event)),
 	}
+
+	// We create a cache against each request, so whenever we make network calls like
+	// fetching kubernetes secrets, we can do so only once per request.
+	request = interceptors.WithCache(request)
+
 	var resp *http.Response
 	for _, i := range t.Interceptors {
 		var interceptor interceptors.Interceptor