[coordinator] Allow clients to pass in augmentM3Tags as a config (#3014)

AugmentM3Tags will augment the metric type to aggregated metrics
to be used within the filter for rules. If enabled, for example,
your filter can specify '__m3_type__:gauge' to filter by gauges.
This is particularly useful for Graphite metrics today.
Furthermore, the option is automatically enabled if static rules are
used and any filter contain an __m3_type__ tag.

src/cmd/services/m3coordinator/downsample/downsampler.go

@@ -135,7 +135,7 @@ func defaultMetricsAppenderOptions(opts DownsamplerOptions, agg agg) metricsAppe
 		metricTagsIteratorPool: agg.pools.metricTagsIteratorPool,
 		debugLogging:           debugLogging,
 		logger:                 logger,
-		augmentM3Tags:          agg.m3PrefixFilter,
+		augmentM3Tags:          agg.augmentM3Tags,
 	}
 }
 

src/cmd/services/m3coordinator/downsample/options.go

@@ -225,10 +225,10 @@ type agg struct {
 	aggregator   aggregator.Aggregator
 	clientRemote client.Client
 
-	clockOpts      clock.Options
-	matcher        matcher.Matcher
-	pools          aggPools
-	m3PrefixFilter bool
+	clockOpts     clock.Options
+	matcher       matcher.Matcher
+	pools         aggPools
+	augmentM3Tags bool
 }
 
 // Configuration configurates a downsampler.
@@ -265,6 +265,14 @@ type Configuration struct {
 
 	// DisableAutoMappingRules disables auto mapping rules.
 	DisableAutoMappingRules bool `yaml:"disableAutoMappingRules"`
+
+	// AugmentM3Tags will augment the metric type to aggregated metrics
+	// to be used within the filter for rules. If enabled, for example,
+	// your filter can specify '__m3_type__:gauge' to filter by gauges.
+	// This is particularly useful for Graphite metrics today.
+	// Furthermore, the option is automatically enabled if static rules are
+	// used and any filter contain an __m3_type__ tag.
+	AugmentM3Tags bool `yaml:"augmentM3Tags"`
 }
 
 // MatcherConfiguration is the configuration for the rule matcher.
@@ -653,7 +661,7 @@ func (cfg Configuration) newAggregator(o DownsamplerOptions) (agg, error) {
 		scope                   = instrumentOpts.MetricsScope()
 		logger                  = instrumentOpts.Logger()
 		openTimeout             = defaultOpenTimeout
-		m3PrefixFilter          = false
+		augmentM3Tags           = cfg.AugmentM3Tags
 		namespaceTag            = defaultNamespaceTag
 	)
 	if o.StorageFlushConcurrency > 0 {
@@ -713,7 +721,7 @@ func (cfg Configuration) newAggregator(o DownsamplerOptions) (agg, error) {
 			updateMetadata)
 		for _, mappingRule := range cfg.Rules.MappingRules {
 			if strings.Contains(mappingRule.Filter, metric.M3MetricsPrefixString) {
-				m3PrefixFilter = true
+				augmentM3Tags = true
 			}
 			rule, err := mappingRule.Rule()
 			if err != nil {
@@ -728,7 +736,7 @@ func (cfg Configuration) newAggregator(o DownsamplerOptions) (agg, error) {
 
 		for _, rollupRule := range cfg.Rules.RollupRules {
 			if strings.Contains(rollupRule.Filter, metric.M3MetricsPrefixString) {
-				m3PrefixFilter = true
+				augmentM3Tags = true
 			}
 			rule, err := rollupRule.Rule()
 			if err != nil {
@@ -783,10 +791,10 @@ func (cfg Configuration) newAggregator(o DownsamplerOptions) (agg, error) {
 		}
 
 		return agg{
-			clientRemote:   client,
-			matcher:        matcher,
-			pools:          pools,
-			m3PrefixFilter: m3PrefixFilter,
+			clientRemote:  client,
+			matcher:       matcher,
+			pools:         pools,
+			augmentM3Tags: augmentM3Tags,
 		}, nil
 	}
 
@@ -948,10 +956,10 @@ func (cfg Configuration) newAggregator(o DownsamplerOptions) (agg, error) {
 	}
 
 	return agg{
-		aggregator:     aggregatorInstance,
-		matcher:        matcher,
-		pools:          pools,
-		m3PrefixFilter: m3PrefixFilter,
+		aggregator:    aggregatorInstance,
+		matcher:       matcher,
+		pools:         pools,
+		augmentM3Tags: augmentM3Tags,
 	}, nil
 }