feat: Composite configurable mutator cache key per rule.

.schema/config.schema.json

@@ -1127,6 +1127,11 @@
             "enabled": {
               "$ref": "#/definitions/handlerSwitch"
             },
+            "key": {
+              "type": "string",
+              "title": "Cache key",
+              "description": "Define custom cache key, i.e '{{ print .Subject }}'. All properties must be part of AuthenticationSession."
+            },
             "ttl": {
               "type": "string",
               "pattern": "^[0-9]+(ns|us|ms|s|m|h)$",

pipeline/mutate/mutator_hydrator.go

@@ -27,6 +27,8 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
+	"strings"
+	"sync"
 	"time"
 
 	"github.com/dgraph-io/ristretto"
@@ -51,13 +53,14 @@ const (
 	ErrNoCredentialsProvided            = "No credentials were provided in mutator configuration"
 	contentTypeHeaderKey                = "Content-Type"
 	contentTypeJSONHeaderValue          = "application/json"
+	poolBufCapacityBytes                = 4096
 )
 
 type MutatorHydrator struct {
-	c      configuration.Provider
-	client *http.Client
-	d      mutatorHydratorDependencies
-
+	c            configuration.Provider
+	client       *http.Client
+	d            mutatorHydratorDependencies
+	p            sync.Pool
 	hydrateCache *ristretto.Cache
 	cacheTTL     *time.Duration
 }
@@ -85,8 +88,8 @@ type externalAPIConfig struct {
 type cacheConfig struct {
 	Enabled bool   `json:"enabled"`
 	TTL     string `json:"ttl"`
-
-	ttl time.Duration
+	Key     string `json:"key"`
+	ttl     time.Duration
 }
 
 type MutatorHydratorConfig struct {
@@ -107,68 +110,83 @@ func NewMutatorHydrator(c configuration.Provider, d mutatorHydratorDependencies)
 		// This is a best-practice value.
 		BufferItems: 64,
 	})
-	return &MutatorHydrator{c: c, d: d, client: httpx.NewResilientClientLatencyToleranceSmall(nil), hydrateCache: cache}
+	return &MutatorHydrator{c: c, d: d,
+		p: sync.Pool{
+			New: func() interface{} {
+				b := &bytes.Buffer{}
+				b.Grow(poolBufCapacityBytes)
+				return b
+			},
+		},
+		client: httpx.NewResilientClientLatencyToleranceSmall(nil), hydrateCache: cache}
 }
 
 func (a *MutatorHydrator) GetID() string {
 	return "hydrator"
 }
 
-func (a *MutatorHydrator) cacheKey(config *MutatorHydratorConfig, session string) string {
-	return fmt.Sprintf("%s|%x", config.Api.URL, md5.Sum([]byte(session)))
+// cacheKey creates a (composite) cache key.
+func (a *MutatorHydrator) cacheKey(keys ...string) string {
+	return fmt.Sprintf("%x", md5.Sum([]byte(strings.Join(keys, ""))))
 }
 
-func (a *MutatorHydrator) hydrateFromCache(config *MutatorHydratorConfig, session string) (*authn.AuthenticationSession, bool) {
-	if !config.Cache.Enabled {
-		return nil, false
-	}
-
-	item, found := a.hydrateCache.Get(a.cacheKey(config, session))
+func (a *MutatorHydrator) hydrateFromCache(key string) (*authn.AuthenticationSession, bool) {
+	item, found := a.hydrateCache.Get(key)
 	if !found {
 		return nil, false
 	}
-
 	return item.(*authn.AuthenticationSession).Copy(), true
 }
 
 func (a *MutatorHydrator) hydrateToCache(config *MutatorHydratorConfig, key string, session *authn.AuthenticationSession) {
-	if !config.Cache.Enabled {
-		return
-	}
-
-	if a.hydrateCache.SetWithTTL(a.cacheKey(config, key), session.Copy(), 0, config.Cache.ttl) {
+	if a.hydrateCache.SetWithTTL(key, session.Copy(), 0, config.Cache.ttl) {
 		a.d.Logger().Debug("Cache reject item")
 	}
 }
 
-func (a *MutatorHydrator) Mutate(r *http.Request, session *authn.AuthenticationSession, config json.RawMessage, _ pipeline.Rule) error {
+func (a *MutatorHydrator) Mutate(r *http.Request, session *authn.AuthenticationSession, config json.RawMessage, p pipeline.Rule) error {
 	cfg, err := a.Config(config)
 	if err != nil {
 		return err
 	}
 
-	var b bytes.Buffer
-	if err := json.NewEncoder(&b).Encode(session); err != nil {
-		return errors.WithStack(err)
-	}
+	sessionCacheKey := a.p.Get().(*bytes.Buffer)
+	defer func() {
+		sessionCacheKey.Reset()
+		a.p.Put(sessionCacheKey)
+	}()
 
-	encodedSession := b.String()
-	if cacheSession, ok := a.hydrateFromCache(cfg, encodedSession); ok {
-		*session = *cacheSession
-		return nil
+	err = json.NewEncoder(sessionCacheKey).Encode(session)
+	switch {
+	case err != nil:
+		return errors.WithStack(err)
+	case !cfg.Cache.Enabled:
+		break
+	case len(cfg.Cache.Key) > 0:
+		// Build a composite cache key with property from configuration.
+		if cacheSession, ok := a.hydrateFromCache(a.cacheKey(
+			cfg.Api.URL, cfg.Cache.Key, p.GetID(), session.Subject)); ok {
+			*session = *cacheSession
+			return nil
+		}
+		a.d.Logger().Debugf("Cache key %s in rule %s was not found. Falling back on default.",
+			cfg.Cache.Key, p.GetID())
+		fallthrough
+	default:
+		if cacheSession, ok := a.hydrateFromCache(a.cacheKey(sessionCacheKey.String())); ok {
+			*session = *cacheSession
+			return nil
+		}
 	}
 
-	if cfg.Api.URL == "" {
-		return errors.New(ErrMissingAPIURL)
-	} else if _, err := url.ParseRequestURI(cfg.Api.URL); err != nil {
+	if _, err = url.ParseRequestURI(cfg.Api.URL); err != nil {
 		return errors.New(ErrInvalidAPIURL)
 	}
-	req, err := http.NewRequest("POST", cfg.Api.URL, &b)
+
+	req, err := http.NewRequest("POST", cfg.Api.URL, sessionCacheKey)
 	if err != nil {
 		return errors.WithStack(err)
-	}
-
-	if r.URL != nil {
+	} else if r.URL != nil {
 		q := r.URL.Query()
 		req.URL.RawQuery = q.Encode()
 	}
@@ -190,19 +208,20 @@ func (a *MutatorHydrator) Mutate(r *http.Request, session *authn.AuthenticationS
 		giveUpAfter := time.Millisecond * 50
 		if len(cfg.Api.Retry.MaxDelay) > 0 {
 			if d, err := time.ParseDuration(cfg.Api.Retry.MaxDelay); err != nil {
-				a.d.Logger().WithError(err).Warn("Unable to parse max_delay in the Hydrator Mutator, falling pack to default.")
+				a.d.Logger().WithError(err).Warn("Unable to parse max_delay in the Hydrator Mutator, " +
+					"falling pack to default.")
 			} else {
 				maxRetryDelay = d
 			}
 		}
 		if len(cfg.Api.Retry.GiveUpAfter) > 0 {
 			if d, err := time.ParseDuration(cfg.Api.Retry.GiveUpAfter); err != nil {
-				a.d.Logger().WithError(err).Warn("Unable to parse max_delay in the Hydrator Mutator, falling pack to default.")
+				a.d.Logger().WithError(err).Warn("Unable to parse max_delay in the Hydrator Mutator, " +
+					"falling pack to default.")
 			} else {
 				giveUpAfter = d
 			}
 		}
-
 		client.Transport = httpx.NewResilientRoundTripper(a.client.Transport, maxRetryDelay, giveUpAfter)
 	}
 
@@ -217,25 +236,29 @@ func (a *MutatorHydrator) Mutate(r *http.Request, session *authn.AuthenticationS
 	case http.StatusUnauthorized:
 		if cfg.Api.Auth != nil {
 			return errors.New(ErrInvalidCredentials)
-		} else {
-			return errors.New(ErrNoCredentialsProvided)
 		}
+		return errors.New(ErrNoCredentialsProvided)
 	default:
 		return errors.New(ErrNon200ResponseFromAPI)
 	}
-
 	sessionFromUpstream := authn.AuthenticationSession{}
-	err = json.NewDecoder(res.Body).Decode(&sessionFromUpstream)
-	if err != nil {
+
+	if err := json.NewDecoder(res.Body).Decode(&sessionFromUpstream); err != nil {
 		return errors.WithStack(err)
-	}
-	if sessionFromUpstream.Subject != session.Subject {
+	} else if sessionFromUpstream.Subject != session.Subject {
 		return errors.New(ErrMalformedResponseFromUpstreamAPI)
 	}
 	*session = sessionFromUpstream
 
-	a.hydrateToCache(cfg, encodedSession, session)
-
+	switch {
+	case !cfg.Cache.Enabled:
+		break
+	case len(cfg.Cache.Key) > 0:
+		a.hydrateToCache(cfg, a.cacheKey(
+			cfg.Api.URL, cfg.Cache.Key, p.GetID(), session.Subject), session)
+	default:
+		a.hydrateToCache(cfg, sessionCacheKey.String(), session)
+	}
 	return nil
 }
 
@@ -258,14 +281,12 @@ func (a *MutatorHydrator) Config(config json.RawMessage) (*MutatorHydratorConfig
 		var err error
 		c.Cache.ttl, err = time.ParseDuration(c.Cache.TTL)
 		if err != nil {
-			a.d.Logger().WithError(err).WithField("ttl", c.Cache.TTL).Error("Unable to parse cache ttl in the Hydrator Mutator.")
+			a.d.Logger().WithError(err).WithField("ttl",
+				c.Cache.TTL).Error("Unable to parse cache ttl in the Hydrator Mutator.")
 			return nil, NewErrMutatorMisconfigured(a, err)
-		}
-
-		if c.Cache.ttl == 0 {
+		} else if c.Cache.ttl == 0 {
 			c.Cache.ttl = time.Minute
 		}
 	}
-
 	return &c, nil
 }

pipeline/mutate/mutator_hydrator_test.go

@@ -9,6 +9,7 @@ import (
 	"net/http/httptest"
 	"net/url"
 	"testing"
+	"time"
 
 	"github.com/julienschmidt/httprouter"
 
@@ -127,12 +128,17 @@ func configWithRetriesForMutator(giveUpAfter, retryDelay string) func(*httptest.
 	}
 }
 
+func configWithSpecialCacheKey(key string) func(*httptest.Server) json.RawMessage {
+	return func(s *httptest.Server) json.RawMessage {
+		return []byte(fmt.Sprintf(`{"api": {"url": "%s"}, "cache": {"enabled": true, "ttl": "30s", "key": "%s"}}`, s.URL, key))
+	}
+}
+
 func TestMutatorHydrator(t *testing.T) {
 	conf := internal.NewConfigurationWithDefaults()
 	reg := internal.NewRegistry(conf)
 
-	a, err := reg.PipelineMutator("hydrator")
-	require.NoError(t, err)
+	a := mutate.NewMutatorHydrator(conf, reg)
 	assert.Equal(t, "hydrator", a.GetID())
 
 	t.Run("method=mutate", func(t *testing.T) {
@@ -148,6 +154,7 @@ func TestMutatorHydrator(t *testing.T) {
 		sampleUserId := "user"
 		sampleValidPassword := "passwd1"
 		sampleNotValidPassword := "passwd7"
+
 		var testMap = map[string]struct {
 			Setup   func(*testing.T) http.Handler
 			Session *authn.AuthenticationSession
@@ -207,7 +214,7 @@ func TestMutatorHydrator(t *testing.T) {
 					router := httprouter.New()
 					router.POST("/", func(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
 						w.WriteHeader(http.StatusOK)
-						_, err = w.Write([]byte(`{}`))
+						_, err := w.Write([]byte(`{}`))
 						require.NoError(t, err)
 					})
 					return router
@@ -267,6 +274,17 @@ func TestMutatorHydrator(t *testing.T) {
 				Match:   newAuthenticationSession(),
 				Err:     errors.New("mutator matching this route is misconfigured or disabled"),
 			},
+			"Empty API URL": {
+				Setup:   defaultRouterSetup(),
+				Session: newAuthenticationSession(),
+				Rule:    &rule.Rule{ID: "test-rule"},
+				Config: func(s *httptest.Server) json.RawMessage {
+					return []byte(`{"api": {"url": ""}}`)
+				},
+				Request: &http.Request{},
+				Match:   newAuthenticationSession(),
+				Err:     errors.New("mutator matching this route is misconfigured or disabled"),
+			},
 			"Successful Basic Authentication": {
 				Setup:   withBasicAuth(defaultRouterSetup(setExtra(sampleKey, sampleValue)), sampleUserId, sampleValidPassword),
 				Session: newAuthenticationSession(),
@@ -334,7 +352,7 @@ func TestMutatorHydrator(t *testing.T) {
 						assert.Equal(t, q["a"], []string{"b"})
 						assert.Equal(t, q["c"], []string{"&12"})
 
-						_, err = w.Write([]byte(`{}`))
+						_, err := w.Write([]byte(`{}`))
 						require.NoError(t, err)
 					})
 					return router
@@ -346,27 +364,70 @@ func TestMutatorHydrator(t *testing.T) {
 				Match:   newAuthenticationSession(),
 				Err:     nil,
 			},
+			"Custom Cache Key No Cache Hit": {
+				Setup:   defaultRouterSetup(),
+				Session: newAuthenticationSession(setSubject(sampleSubject)),
+				Rule:    &rule.Rule{ID: "test-rule"},
+				Config:  configWithSpecialCacheKey(sampleSubject),
+				Request: &http.Request{},
+				Match:   newAuthenticationSession(setSubject(sampleSubject)),
+				Err:     nil,
+			},
+			"Custom Cache Key Cache Hit": {
+				Setup:   defaultRouterSetup(),
+				Session: newAuthenticationSession(setSubject(sampleSubject)),
+				Rule:    &rule.Rule{ID: "test-rule"},
+				Config:  configWithSpecialCacheKey(sampleSubject),
+				Request: &http.Request{},
+				Match:   newAuthenticationSession(setSubject(sampleSubject)),
+				Err:     nil,
+			},
 		}
 
 		for testName, specs := range testMap {
 			t.Run(testName, func(t *testing.T) {
 				var router http.Handler
 				var ts *httptest.Server
+
 				if specs.Setup != nil {
 					router = specs.Setup(t)
 				}
 				ts = httptest.NewServer(router)
 				defer ts.Close()
 
-				err := a.Mutate(specs.Request, specs.Session, specs.Config(ts), specs.Rule)
-				if specs.Err == nil {
+				const (
+					ck = "cache"
+					ct = "Custom Cache Key Cache Hit"
+				)
+
+				switch {
+				case testName == ct:
+					specs.Session.Extra = make(map[string]interface{})
+					specs.Session.Extra[ck] = struct{}{}
+					_ = a.Mutate(specs.Request, specs.Session, specs.Config(ts), specs.Rule)
+					// Delete the K/V-combination above, needs to be served from the cache,
+					// knowing same K/V-combination asserts session is originating from cache.
+					delete(specs.Session.Extra, ck)
+					// Cache entry is being written asynchronously. Obviously this here is not
+					// a good strategy, however, the alternative would be to replace the cache.
+					time.Sleep(50 * time.Millisecond)
+				}
+
+				if err := a.Mutate(specs.Request, specs.Session, specs.Config(ts), specs.Rule); specs.Err == nil {
 					// Issuer must run without error
 					require.NoError(t, err)
 				} else {
 					assert.EqualError(t, err, specs.Err.Error())
 				}
 
-				assert.Equal(t, specs.Match, specs.Session)
+				switch {
+				case testName == ct:
+					// As specs.Session is served from cache we can't perform
+					// full equality assertion but assert if cache key is set.
+					assert.Contains(t, specs.Session.Extra, ck)
+				default:
+					assert.Equal(t, specs.Match, specs.Session)
+				}
 			})
 		}