[WIP] Add optional tag slice to matching config

src/metrics/filters/tags_filter.go

@@ -118,7 +118,10 @@ type TagsFilterOptions struct {
 	// Name of the name tag.
 	NameTagKey []byte
 
-	// Function to extract name and tags from an id.
+	// Name of tags to include in rollup ID if seen in metric ID
+	IncludeTagKeys [][]byte
+
+	//Function to extract name and tags from an id.
 	NameAndTagsFn id.NameAndTagsFn
 
 	// Function to create a new sorted tag iterator from id tags.

src/metrics/matcher/config.go

@@ -46,6 +46,7 @@ type Configuration struct {
 	NamespaceTag          string                       `yaml:"namespaceTag" validate:"nonzero"`
 	DefaultNamespace      string                       `yaml:"defaultNamespace" validate:"nonzero"`
 	NameTagKey            string                       `yaml:"nameTagKey" validate:"nonzero"`
+	IncludeTagKeys        []string                     `yaml:"includeTagKeys"`
 	MatchRangePast        *time.Duration               `yaml:"matchRangePast"`
 	SortedTagIteratorPool pool.ObjectPoolConfiguration `yaml:"sortedTagIteratorPool"`
 }
@@ -114,6 +115,11 @@ func (cfg *Configuration) NewOptions(
 		NameAndTagsFn:       m3.NameAndTags,
 		SortedTagIteratorFn: sortedTagIteratorFn,
 	}
+	includeTagKeys := make([][]byte, 0, len(cfg.IncludeTagKeys))
+	for i, includeTagKey := range cfg.IncludeTagKeys {
+		includeTagKeys[i] = []byte(includeTagKey)
+	}
+	tagsFilterOptions.IncludeTagKeys = includeTagKeys
 
 	isRollupIDFn := func(name []byte, tags []byte) bool {
 		return m3.IsRollupID(name, tags, sortedTagIteratorPool)

src/metrics/rules/active_ruleset.go

@@ -522,11 +522,12 @@ func (as *activeRuleSet) matchRollupTarget(
 	}
 
 	var (
-		rollupTags    = rollupOp.Tags
-		sortedTagIter = matchOpts.SortedTagIteratorFn(sortedTagPairBytes)
-		matchTagIdx   = 0
-		nameTagName   = as.tagsFilterOpts.NameTagKey
-		nameTagValue  []byte
+		rollupTags      = rollupOp.Tags
+		sortedTagIter   = matchOpts.SortedTagIteratorFn(sortedTagPairBytes)
+		matchTagIdx     = 0
+		nameTagName     = as.tagsFilterOpts.NameTagKey
+		nameTagValue    []byte
+		includeTagNames = as.tagsFilterOpts.IncludeTagKeys
 	)
 
 	switch rollupOp.Type {
@@ -535,6 +536,7 @@ func (as *activeRuleSet) matchRollupTarget(
 		//
 		// For include rules, every rule has to have a corresponding match. This means we return
 		// early whenever there's a missing match and increment matchRuleIdx whenever there is a match.
+	TagIteration:
 		for hasMoreTags := sortedTagIter.Next(); hasMoreTags; hasMoreTags = sortedTagIter.Next() {
 			tagName, tagVal := sortedTagIter.Current()
 			// nolint:gosimple
@@ -559,6 +561,13 @@ func (as *activeRuleSet) matchRollupTarget(
 				continue
 			}
 
+			for _, includeTagName := range includeTagNames {
+				if bytes.Compare(tagName, includeTagName) == 0 && targetOpts.generateRollupID {
+					tagPairs = append(tagPairs, metricid.TagPair{Name: tagName, Value: tagVal})
+					continue TagIteration
+				}
+			}
+
 			// If one of the target tags is not found in the ID, this is considered  a non-match so return immediately.
 			if res > 0 {
 				return nil, false, nil