add findCache for memoryIdx

Cache patterns and their results in a LRU.
When new defs are added to the index, only cached patterns that match
the new series name are invalidated.  The cache is purged after every
prune task runs and after any call to delete items from the index.

Author: woodsaj

docker/docker-chaos/metrictank.ini

@@ -396,6 +396,8 @@ rules-file = /etc/metrictank/index-rules.conf
 max-prune-lock-time = 100ms
 # use separate indexes per partition
 partitioned = false
+# number of find expressions to cache
+find-cache-size = 1000
 
 ### Bigtable index
 [bigtable-idx]

docker/docker-cluster/metrictank.ini

@@ -396,6 +396,8 @@ rules-file = /etc/metrictank/index-rules.conf
 max-prune-lock-time = 100ms
 # use separate indexes per partition
 partitioned = false
+# number of find expressions to cache
+find-cache-size = 1000
 
 ### Bigtable index
 [bigtable-idx]

docker/docker-dev-custom-cfg-kafka/metrictank.ini

@@ -396,6 +396,8 @@ rules-file = /etc/metrictank/index-rules.conf
 max-prune-lock-time = 100ms
 # use separate indexes per partition
 partitioned = false
+# number of find expressions to cache
+find-cache-size = 1000
 
 ### Bigtable index
 [bigtable-idx]

docs/config.md

@@ -463,6 +463,8 @@ rules-file = /etc/metrictank/index-rules.conf
 max-prune-lock-time = 100ms
 # use separate indexes per partition
 partitioned = false
+# number of find expressions to cache
+find-cache-size = 1000
 ```
 
 ### Bigtable index

idx/memory/find_cache.go

@@ -0,0 +1,122 @@
+package memory
+
+import (
+	"strings"
+	"sync"
+
+	"github.com/grafana/metrictank/stats"
+	lru "github.com/hashicorp/golang-lru"
+	"github.com/raintank/schema"
+	log "github.com/sirupsen/logrus"
+)
+
+var (
+	// metric idx.memory.find-cache.hit is a counter of findCache hits
+	findCacheHit = stats.NewCounterRate32("idx.memory.find-cache.hit")
+	// metric idx.memory.find-cache.hit is a counter of findCache misses
+	findCacheMiss = stats.NewCounterRate32("idx.memory.find-cache.miss")
+)
+
+type FindCache struct {
+	cache map[uint32]*lru.Cache
+	size  int
+	sync.RWMutex
+}
+
+func NewFindCache(size int) *FindCache {
+	return &FindCache{
+		cache: make(map[uint32]*lru.Cache),
+		size:  size,
+	}
+}
+
+func (c *FindCache) Get(orgId uint32, pattern string) ([]*Node, bool) {
+	c.RLock()
+	cache, ok := c.cache[orgId]
+	c.RUnlock()
+	if !ok {
+		findCacheMiss.Inc()
+		return nil, false
+	}
+	nodes, ok := cache.Get(pattern)
+	if !ok {
+		return nil, ok
+	}
+	findCacheHit.Inc()
+	return nodes.([]*Node), ok
+}
+
+func (c *FindCache) Add(orgId uint32, pattern string, nodes []*Node) {
+	c.RLock()
+	cache, ok := c.cache[orgId]
+	c.RUnlock()
+	var err error
+	if !ok {
+		cache, err = lru.New(c.size)
+		if err != nil {
+			log.Errorf("memory-idx: findCache failed to create lru. err=%s", err)
+			return
+		}
+		c.Lock()
+		c.cache[orgId] = cache
+		c.Unlock()
+	}
+	cache.Add(pattern, nodes)
+}
+
+func (c *FindCache) Purge(orgId uint32) {
+	c.RLock()
+	cache, ok := c.cache[orgId]
+	c.RUnlock()
+	if !ok {
+		return
+	}
+	cache.Purge()
+}
+
+func (c *FindCache) InvalidateFor(orgId uint32, path string) {
+	c.RLock()
+	cache, ok := c.cache[orgId]
+	c.RUnlock()
+	if !ok || cache.Len() < 1 {
+		return
+	}
+	tree := &Tree{
+		Items: map[string]*Node{
+			"": &Node{
+				Path:     "",
+				Children: make([]string, 0),
+				Defs:     make([]schema.MKey, 0),
+			},
+		},
+	}
+	pos := strings.Index(path, ".")
+	prevPos := 0
+	for {
+		branch := path[:pos]
+		// add as child of parent branch
+		tree.Items[path[:prevPos]].Children = []string{branch}
+
+		// create this branch/leaf
+		tree.Items[branch] = &Node{
+			Path: branch,
+		}
+		if branch == path {
+			tree.Items[branch].Defs = []schema.MKey{{}}
+			break
+		}
+		prevPos = pos
+		pos = pos + strings.Index(path[pos+1:], ".")
+	}
+
+	for _, k := range cache.Keys() {
+		matches, err := find(tree, k.(string))
+		if err != nil {
+			log.Errorf("memory-idx: checking if new series matches expressions in findCache. series=%s expr=%s", path, k)
+			continue
+		}
+		if len(matches) > 0 {
+			cache.Remove(k)
+		}
+	}
+}

idx/memory/memory.go

@@ -56,6 +56,7 @@ var (
 	indexRulesFile      string
 	IndexRules          conf.IndexRules
 	Partitioned         bool
+	findCacheSize       = 1000
 )
 
 func ConfigSetup() {
@@ -65,6 +66,7 @@ func ConfigSetup() {
 	memoryIdx.BoolVar(&Partitioned, "partitioned", false, "use separate indexes per partition")
 	memoryIdx.IntVar(&TagQueryWorkers, "tag-query-workers", 50, "number of workers to spin up to evaluate tag queries")
 	memoryIdx.IntVar(&matchCacheSize, "match-cache-size", 1000, "size of regular expression cache in tag query evaluation")
+	memoryIdx.IntVar(&findCacheSize, "find-cache-size", 1000, "number of find expressions to cache")
 	memoryIdx.StringVar(&indexRulesFile, "rules-file", "/etc/metrictank/index-rules.conf", "path to index-rules.conf file")
 	memoryIdx.StringVar(&maxPruneLockTimeStr, "max-prune-lock-time", "100ms", "Maximum duration each second a prune job can lock the index.")
 	globalconf.Register("memory-idx", memoryIdx, flag.ExitOnError)
@@ -233,6 +235,8 @@ type UnpartitionedMemoryIdx struct {
 	// used by tag index
 	defByTagSet defByTagSet
 	tags        map[uint32]TagIndex // by orgId
+
+	findCache *FindCache
 }
 
 func NewUnpartitionedMemoryIdx() *UnpartitionedMemoryIdx {
@@ -241,6 +245,7 @@ func NewUnpartitionedMemoryIdx() *UnpartitionedMemoryIdx {
 		defByTagSet: make(defByTagSet),
 		tree:        make(map[uint32]*Tree),
 		tags:        make(map[uint32]TagIndex),
+		findCache:   NewFindCache(findCacheSize),
 	}
 }
 
@@ -453,6 +458,11 @@ func (m *UnpartitionedMemoryIdx) add(def *schema.MetricDefinition) idx.Archive {
 		return *archive
 	}
 
+	// invalidate any findCache items that match this series name
+	defer func() {
+		go m.findCache.InvalidateFor(def.OrgId, path)
+	}()
+
 	//first check to see if a tree has been created for this OrgId
 	tree, ok := m.tree[def.OrgId]
 	if !ok || len(tree.Items) == 0 {
@@ -943,18 +953,36 @@ func (m *UnpartitionedMemoryIdx) idsByTagQuery(orgId uint32, query TagQuery) IdS
 
 func (m *UnpartitionedMemoryIdx) Find(orgId uint32, pattern string, from int64) ([]idx.Node, error) {
 	pre := time.Now()
+	var matchedNodes []*Node
+	var err error
 	m.RLock()
 	defer m.RUnlock()
-	matchedNodes, err := m.find(orgId, pattern)
-	if err != nil {
-		return nil, err
+	tree, ok := m.tree[orgId]
+	if !ok {
+		log.Debugf("memory-idx: orgId %d has no metrics indexed.", orgId)
+	} else {
+		matchedNodes, ok = m.findCache.Get(orgId, pattern)
+		if !ok {
+			matchedNodes, err = find(tree, pattern)
+			if err != nil {
+				return nil, err
+			}
+			m.findCache.Add(orgId, pattern, matchedNodes)
+		}
 	}
 	if orgId != idx.OrgIdPublic && idx.OrgIdPublic > 0 {
-		publicNodes, err := m.find(idx.OrgIdPublic, pattern)
-		if err != nil {
-			return nil, err
+		tree, ok = m.tree[idx.OrgIdPublic]
+		if ok {
+			publicNodes, ok := m.findCache.Get(idx.OrgIdPublic, pattern)
+			if !ok {
+				publicNodes, err = find(tree, pattern)
+				if err != nil {
+					return nil, err
+				}
+				m.findCache.Add(idx.OrgIdPublic, pattern, publicNodes)
+			}
+			matchedNodes = append(matchedNodes, publicNodes...)
 		}
-		matchedNodes = append(matchedNodes, publicNodes...)
 	}
 	log.Debugf("memory-idx: %d nodes matching pattern %s found", len(matchedNodes), pattern)
 	results := make([]idx.Node, 0)
@@ -997,14 +1025,8 @@ func (m *UnpartitionedMemoryIdx) Find(orgId uint32, pattern string, from int64)
 	return results, nil
 }
 
-// find returns all Nodes matching the pattern for the given orgId
-func (m *UnpartitionedMemoryIdx) find(orgId uint32, pattern string) ([]*Node, error) {
-	tree, ok := m.tree[orgId]
-	if !ok {
-		log.Debugf("memory-idx: orgId %d has no metrics indexed.", orgId)
-		return nil, nil
-	}
-
+// find returns all Nodes matching the pattern for the given tree
+func find(tree *Tree, pattern string) ([]*Node, error) {
 	var nodes []string
 	if strings.Index(pattern, ";") == -1 {
 		nodes = strings.Split(pattern, ".")
@@ -1031,17 +1053,17 @@ func (m *UnpartitionedMemoryIdx) find(orgId uint32, pattern string) ([]*Node, er
 	if pos != 0 {
 		branch = strings.Join(nodes[:pos], ".")
 	}
-	log.Debugf("memory-idx: starting search at orgId %d, node %q", orgId, branch)
+	log.Debugf("memory-idx: starting search at node %q", branch)
 	startNode, ok := tree.Items[branch]
 
 	if !ok {
-		log.Debugf("memory-idx: branch %q does not exist in the index for orgId %d", branch, orgId)
+		log.Debugf("memory-idx: branch %q does not exist in the index", branch)
 		return nil, nil
 	}
 
 	if startNode == nil {
 		corruptIndex.Inc()
-		log.Errorf("memory-idx: startNode is nil. org=%d,patt=%q,pos=%d,branch=%q", orgId, pattern, pos, branch)
+		log.Errorf("memory-idx: startNode is nil. patt=%q,pos=%d,branch=%q", pattern, pos, branch)
 		return nil, errors.NewInternal("hit an empty path in the index")
 	}
 
@@ -1072,7 +1094,7 @@ func (m *UnpartitionedMemoryIdx) find(orgId uint32, pattern string) ([]*Node, er
 				grandChild := tree.Items[newBranch]
 				if grandChild == nil {
 					corruptIndex.Inc()
-					log.Errorf("memory-idx: grandChild is nil. org=%d,patt=%q,i=%d,pos=%d,p=%q,path=%q", orgId, pattern, i, pos, p, newBranch)
+					log.Errorf("memory-idx: grandChild is nil. patt=%q,i=%d,pos=%d,p=%q,path=%q", pattern, i, pos, p, newBranch)
 					return nil, errors.NewInternal("hit an empty path in the index")
 				}
 
@@ -1180,7 +1202,11 @@ func (m *UnpartitionedMemoryIdx) Delete(orgId uint32, pattern string) ([]idx.Arc
 	pre := time.Now()
 	m.Lock()
 	defer m.Unlock()
-	found, err := m.find(orgId, pattern)
+	tree, ok := m.tree[orgId]
+	if !ok {
+		return nil, nil
+	}
+	found, err := find(tree, pattern)
 	if err != nil {
 		return nil, err
 	}
@@ -1192,7 +1218,9 @@ func (m *UnpartitionedMemoryIdx) Delete(orgId uint32, pattern string) ([]idx.Arc
 
 	statMetricsActive.Set(len(m.defById))
 	statDeleteDuration.Value(time.Since(pre))
-
+	if len(deletedDefs) > 0 {
+		m.findCache.Purge(orgId)
+	}
 	return deletedDefs, nil
 }
 
@@ -1411,8 +1439,8 @@ ORGS:
 			m.Unlock()
 			tl.Add(time.Since(lockStart))
 			pruned = append(pruned, defs...)
-
 		}
+		m.findCache.Purge(org)
 	}
 
 	statMetricsActive.Set(len(m.defById))

metrictank-sample.ini

@@ -399,6 +399,8 @@ rules-file = /etc/metrictank/index-rules.conf
 max-prune-lock-time = 100ms
 # use separate indexes per partition
 partitioned = false
+# number of find expressions to cache
+find-cache-size = 1000
 
 ### Bigtable index
 [bigtable-idx]

scripts/config/metrictank-docker.ini

@@ -396,6 +396,8 @@ rules-file = /etc/metrictank/index-rules.conf
 max-prune-lock-time = 100ms
 # use separate indexes per partition
 partitioned = false
+# number of find expressions to cache
+find-cache-size = 1000
 
 ### Bigtable index
 [bigtable-idx]

scripts/config/metrictank-package.ini

@@ -396,6 +396,8 @@ rules-file = /etc/metrictank/index-rules.conf
 max-prune-lock-time = 100ms
 # use separate indexes per partition
 partitioned = false
+# number of find expressions to cache
+find-cache-size = 1000
 
 ### Bigtable index
 [bigtable-idx]