Merge pull request #1433 from grafana/meta_tag_enrichment

Meta tag enrichment

Author: replay

api/dataprocessor.go

@@ -255,6 +255,7 @@ func (s *Server) getTargetsRemote(ctx context.Context, ss *models.StorageStats,
 		}
 		out = append(out, resp.series...)
 	}
+
 	log.Debugf("DP getTargetsRemote: total of %d series found on peers", len(out))
 	return out, nil
 }
@@ -317,6 +318,7 @@ LOOP:
 		}
 		out = append(out, resp.series...)
 	}
+
 	ss.Trace(span)
 	log.Debugf("DP getTargetsLocal: %d series found locally", len(out))
 	return out, nil

api/graphite.go

@@ -662,6 +662,7 @@ func (s *Server) executePlan(ctx context.Context, orgId uint32, plan expr.Plan)
 	minFrom := uint32(math.MaxUint32)
 	var maxTo uint32
 	var reqs []models.Req
+	metaTagEnrichmentData := make(map[string]tagquery.Tags)
 
 	// note that different patterns to query can have different from / to, so they require different index lookups
 	// e.g. target=movingAvg(foo.*, "1h")&target=foo.*
@@ -722,6 +723,10 @@ func (s *Server) executePlan(ctx context.Context, orgId uint32, plan expr.Plan)
 						archive.Id, archive.NameWithTags(), r.Query, r.From, r.To, plan.MaxDataPoints, uint32(archive.Interval), cons, consReq, s.Node, archive.SchemaId, archive.AggId)
 					reqs = append(reqs, newReq)
 				}
+
+				if tagquery.MetaTagSupport && len(metric.Defs) > 0 && len(metric.MetaTags) > 0 {
+					metaTagEnrichmentData[metric.Defs[0].NameWithTags()] = metric.MetaTags
+				}
 			}
 		}
 	}
@@ -770,6 +775,14 @@ func (s *Server) executePlan(ctx context.Context, orgId uint32, plan expr.Plan)
 
 	out = mergeSeries(out)
 
+	if len(metaTagEnrichmentData) > 0 {
+		for i := range out {
+			if metaTags, ok := metaTagEnrichmentData[out[i].Target]; ok {
+				out[i].EnrichWithTags(metaTags)
+			}
+		}
+	}
+
 	// instead of waiting for all data to come in and then start processing everything, we could consider starting processing earlier, at the risk of doing needless work
 	// if we need to cancel the request due to a fetch error
 
@@ -787,6 +800,7 @@ func (s *Server) executePlan(ctx context.Context, orgId uint32, plan expr.Plan)
 
 	preRun := time.Now()
 	out, err = plan.Run(data)
+
 	meta.RenderStats.PlanRunDuration = time.Since(preRun)
 	planRunDuration.Value(meta.RenderStats.PlanRunDuration)
 	return out, meta, err

api/models/series.go

@@ -3,10 +3,12 @@ package models
 import (
 	"bytes"
 	"math"
+	"sort"
 	"strconv"
 	"strings"
 
 	"github.com/grafana/metrictank/consolidation"
+	"github.com/grafana/metrictank/expr/tagquery"
 	"github.com/grafana/metrictank/schema"
 	pickle "github.com/kisielk/og-rek"
 )
@@ -68,6 +70,45 @@ func (s *Series) SetTags() {
 	s.Tags["name"] = name
 }
 
+func (s *Series) EnrichWithTags(tags tagquery.Tags) {
+	// every serie has at least one tag, the name tag
+	// if there are no tags defined, this means the tags must not have been set yet
+	if len(s.Tags) == 0 {
+		s.SetTags()
+	}
+
+	for _, tag := range tags {
+		if _, ok := s.Tags[tag.Key]; !ok {
+			s.Tags[tag.Key] = tag.Value
+		}
+	}
+
+	s.buildTargetFromTags()
+}
+
+func (s *Series) buildTargetFromTags() {
+	buf := bytes.NewBufferString(s.Tags["name"])
+
+	keys := make([]string, 0, len(s.Tags)-1)
+	for key := range s.Tags {
+		if key == "name" {
+			continue
+		}
+		keys = append(keys, key)
+	}
+
+	sort.Strings(keys)
+
+	for _, key := range keys {
+		buf.WriteRune(';')
+		buf.WriteString(key)
+		buf.WriteRune('=')
+		buf.WriteString(s.Tags[key])
+	}
+
+	s.Target = buf.String()
+}
+
 func (s Series) Copy(emptyDatapoints []schema.Point) Series {
 	newSeries := Series{
 		Target:       s.Target,

docker/docker-chaos/metrictank.ini

@@ -442,6 +442,8 @@ meta-tag-support = false
 tag-query-workers = 50
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
+# size of the meta tag enrichment cache
+enrichment-cache-size = 10000
 # 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

@@ -442,6 +442,8 @@ meta-tag-support = false
 tag-query-workers = 50
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
+# size of the meta tag enrichment cache
+enrichment-cache-size = 10000
 # 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

@@ -441,7 +441,9 @@ meta-tag-support = false
 # number of workers to spin up to evaluate tag queries
 tag-query-workers = 50
 # size of regular expression cache in tag query evaluation
-tag-query-workers = 50
+match-cache-size = 1000
+# size of the meta tag enrichment cache
+enrichment-cache-size = 10000
 # 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

@@ -442,6 +442,8 @@ meta-tag-support = false
 tag-query-workers = 50
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
+# size of the meta tag enrichment cache
+enrichment-cache-size = 10000
 # 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

@@ -516,6 +516,8 @@ meta-tag-support = false
 tag-query-workers = 50
 # size of regular expression cache in tag query evaluation
 match-cache-size = 1000
+# size of the meta tag enrichment cache
+enrichment-cache-size = 10000
 # path to index-rules.conf file
 rules-file = /etc/metrictank/index-rules.conf
 # maximum duration each second a prune job can lock the index.

expr/tagquery/tag.go

@@ -9,6 +9,7 @@ import (
 	"github.com/grafana/metrictank/schema"
 )
 
+//go:generate msgp
 type Tags []Tag
 
 func ParseTags(tags []string) (Tags, error) {