Remove core restriction in cache and turn it into an active/standby restriction instead

physical/cache.go

@@ -2,29 +2,27 @@ package physical
 
 import (
 	"context"
-	"strings"
 
-	iradix "github.com/hashicorp/go-immutable-radix"
 	"github.com/hashicorp/golang-lru"
 	"github.com/hashicorp/vault/helper/locksutil"
 	log "github.com/mgutz/logxi/v1"
 )
 
 const (
 	// DefaultCacheSize is used if no cache size is specified for NewCache
-	DefaultCacheSize = 32 * 1024
+	DefaultCacheSize = 128 * 1024
 )
 
 // Cache is used to wrap an underlying physical backend
 // and provide an LRU cache layer on top. Most of the reads done by
 // Vault are for policy objects so there is a large read reduction
 // by using a simple write-through cache.
 type Cache struct {
-	backend    Backend
-	lru        *lru.TwoQueueCache
-	locks      []*locksutil.LockEntry
-	exceptions *iradix.Tree
-	logger     log.Logger
+	backend Backend
+	lru     *lru.TwoQueueCache
+	locks   []*locksutil.LockEntry
+	logger  log.Logger
+	enabled bool
 }
 
 // TransactionalCache is a Cache that wraps the physical that is transactional
@@ -34,49 +32,43 @@ type TransactionalCache struct {
 }
 
 // Verify Cache satisfies the correct interfaces
-var _ Purgable = (*Cache)(nil)
+var _ ToggleablePurgemonster = (*Cache)(nil)
+var _ ToggleablePurgemonster = (*TransactionalCache)(nil)
 var _ Backend = (*Cache)(nil)
 var _ Transactional = (*TransactionalCache)(nil)
-var _ Purgable = (*TransactionalCache)(nil)
 
 // NewCache returns a physical cache of the given size.
 // If no size is provided, the default size is used.
-func NewCache(b Backend, size int, coreExceptions []string, logger log.Logger) *Cache {
+func NewCache(b Backend, size int, logger log.Logger) *Cache {
 	if logger.IsTrace() {
 		logger.Trace("physical/cache: creating LRU cache", "size", size)
 	}
 	if size <= 0 {
 		size = DefaultCacheSize
 	}
-	cacheExceptions := iradix.New()
-	for _, key := range coreExceptions {
-		cacheValue := true
-		if strings.HasPrefix(key, "!") {
-			key = strings.TrimPrefix(key, "!")
-			cacheValue = false
-		}
-		cacheExceptions, _, _ = cacheExceptions.Insert([]byte(key), cacheValue)
-	}
 
 	cache, _ := lru.New2Q(size)
 	c := &Cache{
-		backend:    b,
-		lru:        cache,
-		locks:      locksutil.CreateLocks(),
-		exceptions: cacheExceptions,
-		logger:     logger,
+		backend: b,
+		lru:     cache,
+		locks:   locksutil.CreateLocks(),
+		logger:  logger,
 	}
 	return c
 }
 
-func NewTransactionalCache(b Backend, size int, coreExceptions []string, logger log.Logger) *TransactionalCache {
+func NewTransactionalCache(b Backend, size int, logger log.Logger) *TransactionalCache {
 	c := &TransactionalCache{
-		Cache:         NewCache(b, size, coreExceptions, logger),
+		Cache:         NewCache(b, size, logger),
 		Transactional: b.(Transactional),
 	}
 	return c
 }
 
+func (c *Cache) SetEnabled(enabled bool) {
+	c.enabled = enabled
+}
+
 // Purge is used to clear the cache
 func (c *Cache) Purge(ctx context.Context) {
 	// Lock the world
@@ -89,30 +81,30 @@ func (c *Cache) Purge(ctx context.Context) {
 }
 
 func (c *Cache) Put(ctx context.Context, entry *Entry) error {
+	if !c.enabled {
+		return c.backend.Put(ctx, entry)
+	}
+
 	lock := locksutil.LockForKey(c.locks, entry.Key)
 	lock.Lock()
 	defer lock.Unlock()
 
 	err := c.backend.Put(ctx, entry)
-	if err == nil && c.shouldCache(entry.Key) {
+	if err == nil {
 		c.lru.Add(entry.Key, entry)
 	}
 	return err
 }
 
 func (c *Cache) Get(ctx context.Context, key string) (*Entry, error) {
+	if !c.enabled {
+		return c.backend.Get(ctx, key)
+	}
+
 	lock := locksutil.LockForKey(c.locks, key)
 	lock.RLock()
 	defer lock.RUnlock()
 
-	// We do NOT cache negative results for keys in the 'core/' prefix
-	// otherwise we risk certain race conditions upstream. The primary issue is
-	// with the HA mode, we could potentially negatively cache the leader entry
-	// and cause leader discovery to fail.
-	if !c.shouldCache(key) {
-		return c.backend.Get(ctx, key)
-	}
-
 	// Check the LRU first
 	if raw, ok := c.lru.Get(key); ok {
 		if raw == nil {
@@ -136,12 +128,16 @@ func (c *Cache) Get(ctx context.Context, key string) (*Entry, error) {
 }
 
 func (c *Cache) Delete(ctx context.Context, key string) error {
+	if !c.enabled {
+		return c.backend.Delete(ctx, key)
+	}
+
 	lock := locksutil.LockForKey(c.locks, key)
 	lock.Lock()
 	defer lock.Unlock()
 
 	err := c.backend.Delete(ctx, key)
-	if err == nil && c.shouldCache(key) {
+	if err == nil {
 		c.lru.Remove(key)
 	}
 	return err
@@ -170,8 +166,8 @@ func (c *TransactionalCache) Transaction(ctx context.Context, txns []*TxnEntry)
 		return err
 	}
 
-	for _, txn := range txns {
-		if c.shouldCache(txn.Entry.Key) {
+	if c.enabled {
+		for _, txn := range txns {
 			switch txn.Operation {
 			case PutOperation:
 				c.lru.Add(txn.Entry.Key, txn.Entry)
@@ -183,20 +179,3 @@ func (c *TransactionalCache) Transaction(ctx context.Context, txns []*TxnEntry)
 
 	return nil
 }
-
-// shouldCache checks for any cache exceptions
-func (c *Cache) shouldCache(key string) bool {
-	// prefix match if nested under core/
-	if strings.HasPrefix(key, "core/") {
-		if prefix, val, found := c.exceptions.Root().LongestPrefix([]byte(key)); found {
-			strPrefix := string(prefix)
-			if strings.HasSuffix(strPrefix, "/") || strPrefix == key {
-				return val.(bool)
-			}
-		}
-		// default for core/ values is false
-		return false
-	}
-	// default is true
-	return true
-}