[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 automatically include in the rollup metric, if seen in the original metric.
+	IncludeTagKeys map[uint64]struct{}
+
+	//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

@@ -35,6 +35,8 @@ import (
 	"github.com/m3db/m3/src/x/clock"
 	"github.com/m3db/m3/src/x/instrument"
 	"github.com/m3db/m3/src/x/pool"
+
+	murmur3 "github.com/m3db/stackmurmur3/v2"
 )
 
 // Configuration is config used to create a Matcher.
@@ -46,6 +48,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"`
 }
@@ -111,6 +114,7 @@ func (cfg *Configuration) NewOptions(
 	}
 	tagsFilterOptions := filters.TagsFilterOptions{
 		NameTagKey:          []byte(cfg.NameTagKey),
+		IncludeTagKeys:      createIncludeTagKeysMap(cfg.IncludeTagKeys),
 		NameAndTagsFn:       m3.NameAndTags,
 		SortedTagIteratorFn: sortedTagIteratorFn,
 	}
@@ -147,3 +151,12 @@ func (cfg *Configuration) NewOptions(
 
 	return opts, nil
 }
+
+func createIncludeTagKeysMap(includeTagKeys []string) map[uint64]struct{} {
+	includeTagKeysMap := make(map[uint64]struct{}, len(includeTagKeys))
+	for _, includeTagKey := range includeTagKeys {
+		hashKey := murmur3.Sum64([]byte(includeTagKey))
+		includeTagKeysMap[hashKey] = struct{}{}
+	}
+	return includeTagKeysMap
+}

src/metrics/rules/active_ruleset.go

@@ -35,6 +35,8 @@ import (
 	"github.com/m3db/m3/src/metrics/rules/view"
 	"github.com/m3db/m3/src/query/models"
 	xerrors "github.com/m3db/m3/src/x/errors"
+
+	murmur3 "github.com/m3db/stackmurmur3/v2"
 )
 
 type activeRuleSet struct {
@@ -522,11 +524,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 {
@@ -543,9 +546,15 @@ func (as *activeRuleSet) matchRollupTarget(
 				nameTagValue = tagVal
 			}
 
-			// If we've matched all tags, no need to process.
+			_, matchedIncludeTag := includeTagNames[murmur3.Sum64(tagName)]
+
+			// If we've matched all tags, no need to process the rollup rules.
 			// We don't break out of the for loop, because we may still need to find the name tag.
+			// We also still need to add any remaining include tags.
 			if matchTagIdx >= len(rollupTags) {
+				if targetOpts.generateRollupID && matchedIncludeTag {
+					tagPairs = append(tagPairs, metricid.TagPair{Name: tagName, Value: tagVal})
+				}
 				continue
 			}
 
@@ -559,6 +568,12 @@ func (as *activeRuleSet) matchRollupTarget(
 				continue
 			}
 
+			// The current tag didn't match the rollup tag, but we still want to add the include tag to the rollup id.
+			if targetOpts.generateRollupID && matchedIncludeTag {
+				tagPairs = append(tagPairs, metricid.TagPair{Name: tagName, Value: tagVal})
+				continue
+			}
+
 			// 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