Merge branch 'master' into r/drop-mapping-rule

aggregation/type.go

@@ -272,22 +272,6 @@ func NewTypesFromProto(input []aggregationpb.AggregationType) (Types, error) {
 	return res, nil
 }
 
-// UnmarshalYAML unmarshals aggregation types from a string.
-// TODO(xichen): look into whether it's possible to unmarshal it as an array to be more consistent.
-func (aggTypes *Types) UnmarshalYAML(unmarshal func(interface{}) error) error {
-	var str string
-	if err := unmarshal(&str); err != nil {
-		return err
-	}
-
-	parsed, err := ParseTypes(str)
-	if err != nil {
-		return err
-	}
-	*aggTypes = parsed
-	return nil
-}
-
 // Contains checks if the given type is contained in the aggregation types.
 func (aggTypes Types) Contains(aggType Type) bool {
 	for _, at := range aggTypes {

aggregation/type_config_test.go

@@ -31,8 +31,11 @@ import (
 
 func TestTypesConfiguration(t *testing.T) {
 	str := `
-defaultGaugeAggregationTypes: Max
-defaultTimerAggregationTypes: P50,P99,P9999
+defaultGaugeAggregationTypes: [Max]
+defaultTimerAggregationTypes:
+  - P50
+  - P99
+  - P9999
 counterTransformFnType: empty
 timerTransformFnType: suffix
 gaugeTransformFnType: empty
@@ -56,8 +59,11 @@ gaugeTransformFnType: empty
 
 func TestTypesConfigurationNoTransformFnType(t *testing.T) {
 	str := `
-defaultGaugeAggregationTypes: Max
-defaultTimerAggregationTypes: P50,P99,P9999
+defaultGaugeAggregationTypes: [Max]
+defaultTimerAggregationTypes:
+  - P50
+  - P99
+  - P9999
 `
 
 	var cfg TypesConfiguration
@@ -75,8 +81,11 @@ defaultTimerAggregationTypes: P50,P99,P9999
 
 func TestTypesConfigurationError(t *testing.T) {
 	str := `
-defaultGaugeAggregationTypes: Max
-defaultTimerAggregationTypes: P50,P99,P9999
+defaultGaugeAggregationTypes: [Max]
+defaultTimerAggregationTypes:
+  - P50
+  - P99
+  - P9999
 timerTransformFnType: bla
 `
 

aggregation/type_test.go

@@ -21,6 +21,7 @@
 package aggregation
 
 import (
+	"encoding/json"
 	"testing"
 
 	"github.com/m3db/m3x/pool"
@@ -79,26 +80,115 @@ func TestTypesIsDefault(t *testing.T) {
 	require.False(t, Types{Max}.IsDefault())
 }
 
+func TestTypesMarshalJSON(t *testing.T) {
+	inputs := []struct {
+		types       Types
+		expected    string
+		expectedErr bool
+	}{
+		{
+			types:    Types{},
+			expected: `[]`,
+		},
+		{
+			types:    Types{Min},
+			expected: `["Min"]`,
+		},
+		{
+			types:    Types{Mean, Max, P99, P9999},
+			expected: `["Mean","Max","P99","P9999"]`,
+		},
+		{
+			types:       Types{Type(1000)},
+			expectedErr: true,
+		},
+	}
+	for _, input := range inputs {
+		b, err := json.Marshal(input.types)
+		if input.expectedErr {
+			require.Error(t, err)
+			continue
+		}
+		require.NoError(t, err)
+		require.Equal(t, input.expected, string(b))
+	}
+}
+
+func TestTypesUnMarshalJSON(t *testing.T) {
+	inputs := []struct {
+		str         string
+		expected    Types
+		expectedErr bool
+	}{
+		{
+			str:      `[]`,
+			expected: Types{},
+		},
+		{
+			str:      `["Min"]`,
+			expected: Types{Min},
+		},
+		{
+			str:      `["Mean","Max","P99","P9999"]`,
+			expected: Types{Mean, Max, P99, P9999},
+		},
+		{
+			str:         `[P100]`,
+			expectedErr: true,
+		},
+	}
+	for _, input := range inputs {
+		var aggTypes Types
+		err := json.Unmarshal([]byte(input.str), &aggTypes)
+		if input.expectedErr {
+			require.Error(t, err)
+			continue
+		}
+		require.NoError(t, err)
+		require.Equal(t, input.expected, aggTypes)
+	}
+}
+
+func TestTypesMarshalJSONRoundTrip(t *testing.T) {
+	inputs := []Types{
+		{},
+		{Min},
+		{Mean, Max, P99, P9999},
+	}
+
+	for _, input := range inputs {
+		b, err := json.Marshal(input)
+		require.NoError(t, err)
+		var aggTypes Types
+		require.NoError(t, json.Unmarshal(b, &aggTypes))
+		require.Equal(t, input, aggTypes)
+	}
+}
+
 func TestTypesUnmarshalYAML(t *testing.T) {
 	inputs := []struct {
 		str         string
 		expected    Types
 		expectedErr bool
 	}{
 		{
-			str:      "Min",
+			str:      "",
+			expected: Types(nil),
+		},
+		{
+			str:      "[Min]",
 			expected: Types{Min},
 		},
 		{
-			str:      "Mean,Max,P99,P9999",
+			str:      "[Mean,Max,P99,P9999]",
 			expected: Types{Mean, Max, P99, P9999},
 		},
 		{
-			str:         "Min,Max,P99,P9999,P100",
+			str:         "[Min,Max,P99,P9999,P100]",
 			expectedErr: true,
 		},
 		{
-			str:         "Min,Max,P99,P9999,P100",
+			str:         "[Min,Max,P99,P9999,P100]",
 			expectedErr: true,
 		},
 		{

matcher/config.go

@@ -26,7 +26,6 @@ import (
 
 	"github.com/m3db/m3cluster/client"
 	"github.com/m3db/m3cluster/kv"
-	"github.com/m3db/m3metrics/aggregation"
 	"github.com/m3db/m3metrics/filters"
 	"github.com/m3db/m3metrics/matcher/cache"
 	"github.com/m3db/m3metrics/metric/id"
@@ -39,16 +38,15 @@ import (
 
 // Configuration is config used to create a Matcher.
 type Configuration struct {
-	InitWatchTimeout      time.Duration                  `yaml:"initWatchTimeout"`
-	RulesKVConfig         kv.OverrideConfiguration       `yaml:"rulesKVConfig"`
-	NamespacesKey         string                         `yaml:"namespacesKey" validate:"nonzero"`
-	RuleSetKeyFmt         string                         `yaml:"ruleSetKeyFmt" validate:"nonzero"`
-	NamespaceTag          string                         `yaml:"namespaceTag" validate:"nonzero"`
-	DefaultNamespace      string                         `yaml:"defaultNamespace" validate:"nonzero"`
-	NameTagKey            string                         `yaml:"nameTagKey" validate:"nonzero"`
-	MatchRangePast        *time.Duration                 `yaml:"matchRangePast"`
-	SortedTagIteratorPool pool.ObjectPoolConfiguration   `yaml:"sortedTagIteratorPool"`
-	AggregationTypes      aggregation.TypesConfiguration `yaml:"aggregationTypes"`
+	InitWatchTimeout      time.Duration                `yaml:"initWatchTimeout"`
+	RulesKVConfig         kv.OverrideConfiguration     `yaml:"rulesKVConfig"`
+	NamespacesKey         string                       `yaml:"namespacesKey" validate:"nonzero"`
+	RuleSetKeyFmt         string                       `yaml:"ruleSetKeyFmt" validate:"nonzero"`
+	NamespaceTag          string                       `yaml:"namespaceTag" validate:"nonzero"`
+	DefaultNamespace      string                       `yaml:"defaultNamespace" validate:"nonzero"`
+	NameTagKey            string                       `yaml:"nameTagKey" validate:"nonzero"`
+	MatchRangePast        *time.Duration               `yaml:"matchRangePast"`
+	SortedTagIteratorPool pool.ObjectPoolConfiguration `yaml:"sortedTagIteratorPool"`
 }
 
 // NewNamespaces creates a matcher.Namespaces.
@@ -120,15 +118,10 @@ func (cfg *Configuration) NewOptions(
 		return m3.IsRollupID(name, tags, sortedTagIteratorPool)
 	}
 
-	aggTypeOpts, err := cfg.AggregationTypes.NewOptions(instrumentOpts)
-	if err != nil {
-		return nil, err
-	}
 	ruleSetOpts := rules.NewOptions().
 		SetTagsFilterOptions(tagsFilterOptions).
 		SetNewRollupIDFn(m3.NewRollupID).
-		SetIsRollupIDFn(isRollupIDFn).
-		SetAggregationTypesOptions(aggTypeOpts)
+		SetIsRollupIDFn(isRollupIDFn)
 
 	// Configure ruleset key function.
 	ruleSetKeyFn := func(namespace []byte) string {

matcher/namespaces.go

@@ -58,7 +58,14 @@ type Namespaces interface {
 
 	// ReverseMatch reverse matches the matching policies for a given id in a given namespace
 	// between [fromNanos, toNanos), taking into account the metric type and aggregation type for the given id.
-	ReverseMatch(namespace, id []byte, fromNanos, toNanos int64, mt metric.Type, at aggregation.Type) rules.MatchResult
+	ReverseMatch(
+		namespace, id []byte,
+		fromNanos, toNanos int64,
+		mt metric.Type,
+		at aggregation.Type,
+		isMultiAggregationTypesAllowed bool,
+		aggTypesOpts aggregation.TypesOptions,
+	) rules.MatchResult
 
 	// Close closes the namespaces.
 	Close()
@@ -183,12 +190,14 @@ func (n *namespaces) ReverseMatch(
 	fromNanos, toNanos int64,
 	mt metric.Type,
 	at aggregation.Type,
+	isMultiAggregationTypesAllowed bool,
+	aggTypesOpts aggregation.TypesOptions,
 ) rules.MatchResult {
 	ruleSet, exists := n.ruleSet(namespace)
 	if !exists {
 		return rules.EmptyMatchResult
 	}
-	return ruleSet.ReverseMatch(id, fromNanos, toNanos, mt, at)
+	return ruleSet.ReverseMatch(id, fromNanos, toNanos, mt, at, isMultiAggregationTypesAllowed, aggTypesOpts)
 }
 
 func (n *namespaces) ruleSet(namespace []byte) (RuleSet, bool) {

matcher/ruleset.go

@@ -167,6 +167,8 @@ func (r *ruleSet) ReverseMatch(
 	fromNanos, toNanos int64,
 	mt metric.Type,
 	at aggregation.Type,
+	isMultiAggregationTypesAllowed bool,
+	aggTypesOpts aggregation.TypesOptions,
 ) rules.MatchResult {
 	callStart := r.nowFn()
 	r.RLock()
@@ -175,7 +177,7 @@ func (r *ruleSet) ReverseMatch(
 		r.metrics.nilMatcher.Inc(1)
 		return rules.EmptyMatchResult
 	}
-	res := r.matcher.ReverseMatch(id, fromNanos, toNanos, mt, at)
+	res := r.matcher.ReverseMatch(id, fromNanos, toNanos, mt, at, isMultiAggregationTypesAllowed, aggTypesOpts)
 	r.RUnlock()
 	r.metrics.match.ReportSuccess(r.nowFn().Sub(callStart))
 	return res

matcher/ruleset_test.go

@@ -87,17 +87,21 @@ func TestRuleSetReverseMatchWithMatcher(t *testing.T) {
 	rs.matcher = mockMatcher
 
 	var (
-		now       = rs.nowFn()
-		fromNanos = now.Add(-time.Second).UnixNano()
-		toNanos   = now.Add(time.Second).UnixNano()
+		now                            = rs.nowFn()
+		fromNanos                      = now.Add(-time.Second).UnixNano()
+		toNanos                        = now.Add(time.Second).UnixNano()
+		isMultiAggregationTypesAllowed = true
+		aggTypesOpts                   = aggregation.NewTypesOptions()
 	)
 
-	require.Equal(t, mockMatcher.res, rs.ReverseMatch([]byte("foo"), fromNanos, toNanos, metric.CounterType, aggregation.Sum))
+	require.Equal(t, mockMatcher.res, rs.ReverseMatch([]byte("foo"), fromNanos, toNanos, metric.CounterType, aggregation.Sum, isMultiAggregationTypesAllowed, aggTypesOpts))
 	require.Equal(t, []byte("foo"), mockMatcher.id)
 	require.Equal(t, fromNanos, mockMatcher.fromNanos)
 	require.Equal(t, toNanos, mockMatcher.toNanos)
 	require.Equal(t, metric.CounterType, mockMatcher.metricType)
 	require.Equal(t, aggregation.Sum, mockMatcher.aggregationType)
+	require.Equal(t, isMultiAggregationTypesAllowed, mockMatcher.isMultiAggregationTypesAllowed)
+	require.Equal(t, aggTypesOpts, mockMatcher.aggTypesOpts)
 }
 
 func TestToRuleSetNilValue(t *testing.T) {
@@ -217,12 +221,14 @@ func TestRuleSetProcessSuccess(t *testing.T) {
 }
 
 type mockMatcher struct {
-	id              []byte
-	fromNanos       int64
-	toNanos         int64
-	res             rules.MatchResult
-	metricType      metric.Type
-	aggregationType aggregation.Type
+	id                             []byte
+	fromNanos                      int64
+	toNanos                        int64
+	res                            rules.MatchResult
+	metricType                     metric.Type
+	aggregationType                aggregation.Type
+	isMultiAggregationTypesAllowed bool
+	aggTypesOpts                   aggregation.TypesOptions
 }
 
 func (mm *mockMatcher) ForwardMatch(
@@ -240,12 +246,16 @@ func (mm *mockMatcher) ReverseMatch(
 	fromNanos, toNanos int64,
 	mt metric.Type,
 	at aggregation.Type,
+	isMultiAggregationTypesAllowed bool,
+	aggTypesOpts aggregation.TypesOptions,
 ) rules.MatchResult {
 	mm.id = id
 	mm.fromNanos = fromNanos
 	mm.toNanos = toNanos
 	mm.metricType = mt
 	mm.aggregationType = at
+	mm.isMultiAggregationTypesAllowed = isMultiAggregationTypesAllowed
+	mm.aggTypesOpts = aggTypesOpts
 	return mm.res
 }
 
@@ -257,15 +267,15 @@ type mockRuleSet struct {
 	matcher      *mockMatcher
 }
 
-func (r *mockRuleSet) Namespace() []byte                          { return []byte(r.namespace) }
-func (r *mockRuleSet) Version() int                               { return r.version }
-func (r *mockRuleSet) CutoverNanos() int64                        { return r.cutoverNanos }
-func (r *mockRuleSet) LastUpdatedAtNanos() int64                  { return 0 }
-func (r *mockRuleSet) CreatedAtNanos() int64                      { return 0 }
-func (r *mockRuleSet) Tombstoned() bool                           { return r.tombstoned }
-func (r *mockRuleSet) Proto() (*rulepb.RuleSet, error)            { return nil, nil }
-func (r *mockRuleSet) ActiveSet(timeNanos int64) rules.Matcher    { return r.matcher }
-func (r *mockRuleSet) ToMutableRuleSet() rules.MutableRuleSet     { return nil }
+func (r *mockRuleSet) Namespace() []byte                        { return []byte(r.namespace) }
+func (r *mockRuleSet) Version() int                             { return r.version }
+func (r *mockRuleSet) CutoverNanos() int64                      { return r.cutoverNanos }
+func (r *mockRuleSet) LastUpdatedAtNanos() int64                { return 0 }
+func (r *mockRuleSet) CreatedAtNanos() int64                    { return 0 }
+func (r *mockRuleSet) Tombstoned() bool                         { return r.tombstoned }
+func (r *mockRuleSet) Proto() (*rulepb.RuleSet, error)          { return nil, nil }
+func (r *mockRuleSet) ActiveSet(timeNanos int64) rules.Matcher  { return r.matcher }
+func (r *mockRuleSet) ToMutableRuleSet() rules.MutableRuleSet   { return nil }
 func (r *mockRuleSet) MappingRules() (view.MappingRules, error) { return nil, nil }
 func (r *mockRuleSet) RollupRules() (view.RollupRules, error)   { return nil, nil }
 func (r *mockRuleSet) Latest() (view.RuleSet, error)            { return view.RuleSet{}, nil }