Merge pull request #1523 from grafana/rewrite_enricher

Rewrite enricher to use map lookup

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## breaking changes
 
+* as of v0.13.0-160-gd2703083 the default setting for `memory-idx.tag-query-workers` is `5` instead of `50`.
+  If a user still has the value `50` in their config file we recommend decreasing that, because due to how
+  meta tag queries get processed MT may now create multiple pools of workers concurrently to process a single
+  query, where each pool consists of `tag-query-workers` threads.
 * as of v0.13.0-75-geaac736a Metrictank requires two new Cassandra tables if the meta tag feature is enabled and the Cassandra index is used. It only creates them automatically if `cassandra-idx-create-keyspace` is set to true.
 * as of v0.12.0-404-gc7715cb2 we clean up poorly formatted graphite metrics better. To the extent that they have previously worked, queries may need some adjusting
   #1435

docker/docker-chaos/metrictank.ini

@@ -452,8 +452,12 @@ meta-tag-support = false
 tag-query-workers = 5
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
-# size of the meta tag enrichment cache
-enrichment-cache-size = 10000
+# size of event queue in the meta tag enricher
+meta-tag-enricher-queue-size = 100
+# size of add metric event buffer in enricher
+meta-tag-enricher-buffer-size = 10000
+# how long to buffer enricher events before they must be processed
+meta-tag-enricher-buffer-time = 5s
 # path to index-rules.conf file
 rules-file = /etc/metrictank/index-rules.conf
 # maximum duration each second a prune job can lock the index.

docker/docker-cluster-query/metrictank.ini

@@ -452,8 +452,12 @@ meta-tag-support = false
 tag-query-workers = 5
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
-# size of the meta tag enrichment cache
-enrichment-cache-size = 10000
+# size of event queue in the meta tag enricher
+meta-tag-enricher-queue-size = 100
+# size of add metric event buffer in enricher
+meta-tag-enricher-buffer-size = 10000
+# how long to buffer enricher events before they must be processed
+meta-tag-enricher-buffer-time = 5s
 # path to index-rules.conf file
 rules-file = /etc/metrictank/index-rules.conf
 # maximum duration each second a prune job can lock the index.

docker/docker-cluster/metrictank.ini

@@ -452,8 +452,12 @@ meta-tag-support = false
 tag-query-workers = 5
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
-# size of the meta tag enrichment cache
-enrichment-cache-size = 10000
+# size of event queue in the meta tag enricher
+meta-tag-enricher-queue-size = 100
+# size of add metric event buffer in enricher
+meta-tag-enricher-buffer-size = 10000
+# how long to buffer enricher events before they must be processed
+meta-tag-enricher-buffer-time = 5s
 # path to index-rules.conf file
 rules-file = /etc/metrictank/index-rules.conf
 # maximum duration each second a prune job can lock the index.

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

@@ -452,8 +452,12 @@ meta-tag-support = false
 tag-query-workers = 5
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
-# size of the meta tag enrichment cache
-enrichment-cache-size = 10000
+# size of event queue in the meta tag enricher
+meta-tag-enricher-queue-size = 100
+# size of add metric event buffer in enricher
+meta-tag-enricher-buffer-size = 10000
+# how long to buffer enricher events before they must be processed
+meta-tag-enricher-buffer-time = 5s
 # path to index-rules.conf file
 rules-file = /etc/metrictank/index-rules.conf
 # maximum duration each second a prune job can lock the index.

docs/config.md

@@ -526,8 +526,12 @@ meta-tag-support = false
 tag-query-workers = 5
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
-# size of the meta tag enrichment cache
-enrichment-cache-size = 10000
+# size of event queue in the meta tag enricher
+meta-tag-enricher-queue-size = 100
+# size of add metric event buffer in enricher
+meta-tag-enricher-buffer-size = 10000
+# how long to buffer enricher events before they must be processed
+meta-tag-enricher-buffer-time = 5s
 # path to index-rules.conf file
 rules-file = /etc/metrictank/index-rules.conf
 # maximum duration each second a prune job can lock the index.

docs/metrics.md

@@ -202,12 +202,10 @@ a counter of findCache misses
 the duration of a get of one metric in the memory idx
 * `idx.memory.list`:  
 the duration of memory idx listings
-* `idx.memory.meta-tags.enrichment-cache.entries`:  
-a the number of entries in the enrichment cache
-* `idx.memory.meta-tags.enrichment-cache.ops.hit`:  
-a counter of enrichment cache hits
-* `idx.memory.meta-tags.enrichment-cache.ops.miss`:  
-a counter of enrichment cache misses
+* `idx.memory.meta-tags.enricher.ops.known-meta-records`:  
+a counter of meta records known to the enricher
+* `idx.memory.meta-tags.enricher.ops.metrics-with-meta-records`:  
+a counter of metrics with at least one associated meta record
 * `idx.memory.ops.add`:  
 the number of additions to the memory idx
 * `idx.memory.ops.update`:  

expr/tagquery/expression.go

@@ -59,6 +59,20 @@ func (e Expressions) Sort() {
 	})
 }
 
+func (e Expressions) Equal(other Expressions) bool {
+	if len(e) != len(other) {
+		return false
+	}
+
+	for i, expression := range e {
+		if !expression.Equals(other[i]) {
+			return false
+		}
+	}
+
+	return true
+}
+
 // MarshalJSON satisfies the json.Marshaler interface
 // it is used by the api endpoint /metaTags to list the meta tag records
 func (e Expressions) MarshalJSON() ([]byte, error) {

expr/tagquery/meta_tag_record.go

@@ -54,7 +54,7 @@ func (m *MetaTagRecord) Equals(other *MetaTagRecord) bool {
 		}
 	}
 
-	return m.EqualExpressions(other)
+	return m.Expressions.Equal(other.Expressions)
 }
 
 // HashExpressions returns a hash of all expressions in this meta tag record
@@ -92,24 +92,6 @@ func (m *MetaTagRecord) HashRecord() uint64 {
 	return uint64(expressionsHash) | uint64(metaTagHash)<<32
 }
 
-// EqualExpressions compares another meta tag record's expressions to
-// this one's expressions
-// Returns true if they are equal, otherwise false
-// It is assumed that all the expressions are already sorted
-func (m *MetaTagRecord) EqualExpressions(other *MetaTagRecord) bool {
-	if len(m.Expressions) != len(other.Expressions) {
-		return false
-	}
-
-	for i, expression := range m.Expressions {
-		if !expression.Equals(other.Expressions[i]) {
-			return false
-		}
-	}
-
-	return true
-}
-
 // HasMetaTags returns true if the meta tag record has one or more
 // meta tags, otherwise it returns false
 func (m *MetaTagRecord) HasMetaTags() bool {

expr/tagquery/tag.go

@@ -81,6 +81,20 @@ func (t Tags) Sort() {
 	})
 }
 
+func (t Tags) Equal(other Tags) bool {
+	if len(t) != len(other) {
+		return false
+	}
+
+	for i := range t {
+		if t[i] != other[i] {
+			return false
+		}
+	}
+
+	return true
+}
+
 // MarshalJSON satisfies the json.Marshaler interface
 // it is used by the api endpoint /metaTags to list the meta tag records
 func (t Tags) MarshalJSON() ([]byte, error) {

idx/cassandra/cassandra.go

@@ -398,19 +398,14 @@ func (c *CasIdx) rebuildIndex() {
 		}(partition)
 	}
 
-	if memory.MetaTagSupport {
-		wg.Add(1)
-		go func() {
-			gate <- struct{}{}
-
-			c.loadMetaRecords()
+	wg.Wait()
 
-			wg.Done()
-			<-gate
-		}()
+	if memory.TagSupport && memory.MetaTagSupport {
+		// should only get called after the metric index has been initialized
+		// if metrics get added first and meta records second then startup is faster
+		c.loadMetaRecords()
 	}
 
-	wg.Wait()
 	log.Infof("cassandra-idx: Rebuilding Memory Index Complete. Imported %d. Took %s", num, time.Since(pre))
 }
 

idx/cassandra/meta_records.go

@@ -129,6 +129,7 @@ func (c *CasIdx) loadMetaRecords() {
 	}
 
 	for orgId, batchId := range toLoad {
+		log.Infof("cassandra-idx: Loading meta record batch %s of org %d", batchId.String(), orgId)
 		var expressions, metatags string
 		q = fmt.Sprintf("SELECT expressions, metatags FROM %s WHERE batchid=? AND orgid=?", c.Config.MetaRecordTable)
 		iter = c.Session.Query(q, batchId, orgId).RetryPolicy(&metaRecordRetryPolicy).Iter()