Merge pull request #1863 from grafana/add-sum-series-with-wildcards

Adds aggregateWithWildcards (and sum / multiply / avg)

Author: Dieterbe

api/models/series.go

@@ -201,6 +201,15 @@ func (s *Series) buildTargetFromTags() {
 	s.Target = buf.String()
 }
 
+// SeriesCopy returns a deep copy of the series slice
+func SeriesCopy(series []Series) []Series {
+	out := make([]Series, len(series))
+	for i, s := range series {
+		out[i] = s.Copy(nil)
+	}
+	return out
+}
+
 // Copy returns a deep copy.
 // The returned value does not link to the same memory space for any of the properties
 func (s Series) Copy(emptyDatapoints []schema.Point) Series {

docs/graphite.md

@@ -41,7 +41,7 @@ See also:
 | absolute                                                       |              | Stable     |
 | aggregate                                                      |              | Stable     |
 | aggregateLine                                                  |              | No         |
-| aggregateWithWildcards                                         |              | No         |
+| aggregateWithWildcards                                         |              | Stable     |
 | alias(seriesList, alias) seriesList                            |              | Stable     |
 | aliasByMetric                                                  |              | Stable     |
 | aliasByNode(seriesList, nodeList) seriesList                   | aliasByTags  | Stable     |
@@ -56,7 +56,7 @@ See also:
 | averageBelow                                                   |              | Stable     |
 | averageOutsidePercentile                                       |              | No         |
 | averageSeries(seriesLists) series                              | avg          | Stable     |
-| averageSeriesWithWildcards                                     |              | No         |
+| averageSeriesWithWildcards                                     |              | Stable     |
 | cactiStyle                                                     |              | No         |
 | changed                                                        |              | No         |
 | color                                                          |              | No         |
@@ -123,7 +123,7 @@ See also:
 | movingSum                                                      |              | No         |
 | movingWindow                                                   |              | No         |
 | multiplySeries(seriesList) series                              |              | Stable     |
-| multiplySeriesWithWildcards                                    |              | No         |
+| multiplySeriesWithWildcards                                    |              | Stable     |
 | nonNegatievDerivative(seriesList, maxValue) seriesList         |              | Stable     |
 | nPercentile                                                    |              | No         |
 | offset                                                         |              | No         |
@@ -164,7 +164,7 @@ See also:
 | substr                                                         |              | No         |
 | summarize(seriesList) seriesList                               |              | Stable     |
 | sumSeries(seriesLists) series                                  | sum          | Stable     |
-| sumSeriesWithWildcards                                         |              | No         |
+| sumSeriesWithWildcards                                         |              | Stable     |
 | threshold                                                      |              | No         |
 | timeFunction                                                   | time         | No         |
 | timeShift                                                      |              | No         |

expr/func_aggregatewithwildcards.go

@@ -0,0 +1,77 @@
+package expr
+
+import (
+	"github.com/grafana/metrictank/api/models"
+)
+
+type FuncAggregateWithWildcards struct {
+	in    GraphiteFunc
+	fn    string
+	nodes []expr
+}
+
+func NewAggregateWithWildcardsConstructor(fn string) func() GraphiteFunc {
+	return func() GraphiteFunc {
+		return &FuncAggregateWithWildcards{fn: fn}
+	}
+}
+
+func NewAggregateWithWildcards() GraphiteFunc {
+	return &FuncAggregateWithWildcards{}
+}
+
+func (s *FuncAggregateWithWildcards) Signature() ([]Arg, []Arg) {
+	if s.fn == "" {
+		return []Arg{
+			ArgSeriesList{val: &s.in},
+			ArgString{key: "func", val: &s.fn, validator: []Validator{IsAggFunc}},
+			ArgStringsOrInts{key: "positions", val: &s.nodes, validator: []Validator{IntZeroOrPositive}},
+		}, []Arg{ArgSeriesList{}}
+	}
+	return []Arg{
+		ArgSeriesList{val: &s.in},
+		ArgStringsOrInts{key: "positions", val: &s.nodes, validator: []Validator{IntZeroOrPositive}},
+	}, []Arg{ArgSeriesList{}}
+}
+
+func (s *FuncAggregateWithWildcards) Context(context Context) Context {
+	return context
+}
+
+func (s *FuncAggregateWithWildcards) Exec(dataMap DataMap) ([]models.Series, error) {
+	series, err := s.in.Exec(dataMap)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(series) == 0 {
+		return series, nil
+	}
+
+	agg := seriesAggregator{function: getCrossSeriesAggFunc(s.fn), name: s.fn}
+
+	groups := make(map[string][]models.Series)
+	var keys []string
+
+	for _, serie := range series {
+		key := filterNodesByPositions(serie.Target, s.nodes)
+		_, ok := groups[key]
+		if !ok {
+			keys = append(keys, key)
+		}
+		groups[key] = append(groups[key], serie)
+	}
+	out := make([]models.Series, 0, len(groups))
+
+	for _, key := range keys {
+		res, err := aggregate(dataMap, groups[key], []string{key}, agg, 0)
+		if err != nil {
+			return nil, err
+		}
+		res[0].Target = key
+		res[0].QueryPatt = key
+		out = append(out, res...)
+	}
+
+	return out, nil
+}

expr/func_aggregatewithwildcards_test.go

@@ -0,0 +1,128 @@
+package expr
+
+import (
+	"testing"
+
+	"github.com/grafana/metrictank/api/models"
+)
+
+func TestAggregateWithWildcardsZero(t *testing.T) {
+	f := makeAggregateWithWildcards("sum", []models.Series{}, 0, []expr{{etype: etInt, int: 1}})
+	got, err := f.Exec(make(map[Req][]models.Series))
+	if err := equalOutput([]models.Series{}, got, nil, err); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestAggregateWithWildcardsIdentity(t *testing.T) {
+	testAggregateWithWildcards(
+		"identity",
+		"average",
+		[]models.Series{
+			getSeriesNamed("single", a),
+		},
+		getSeriesNamed("single", a),
+		t,
+		0,
+		[]expr{},
+	)
+	testAggregateWithWildcards(
+		"identity",
+		"sum",
+		[]models.Series{
+			getSeriesNamed("single", a),
+		},
+		getSeriesNamed("single", a),
+		t,
+		0,
+		[]expr{},
+	)
+}
+func TestAggregateWithWildcardsAvg(t *testing.T) {
+	testAggregateWithWildcards(
+		"avgSeries",
+		"average",
+		[]models.Series{
+			getSeriesNamed("foo.fighters.bar", a),
+			getSeriesNamed("foo.rock.bar", b),
+		},
+		getSeriesNamed("foo.bar", avgab),
+		t,
+		0,
+		[]expr{{etype: etInt, int: 1}},
+	)
+}
+
+func TestAggregateWithWildcardsSum(t *testing.T) {
+	testAggregateWithWildcards(
+		"sumSeries",
+		"sum",
+		[]models.Series{
+			getSeriesNamed("foo.hello.bar", a),
+			getSeriesNamed("foo.world.bar", b),
+		},
+		getSeriesNamed("foo.bar", sumab),
+		t,
+		0,
+		[]expr{{etype: etInt, int: 1}},
+	)
+}
+
+func TestAggregateWithWildcardsMultiply(t *testing.T) {
+	testAggregateWithWildcards(
+		"multiplySeries",
+		"multiply",
+		[]models.Series{
+			getSeriesNamed("foo.hello.world.bar", a),
+			getSeriesNamed("foo.something.else.bar", b),
+		},
+		getSeriesNamed("foo.bar", multab),
+		t,
+		0,
+		[]expr{
+			{etype: etInt, int: 1},
+			{etype: etInt, int: 2},
+		},
+	)
+}
+
+func makeAggregateWithWildcards(agg string, in []models.Series, xFilesFactor float64, nodes []expr) GraphiteFunc {
+	f := NewAggregateWithWildcardsConstructor(agg)()
+	aggww := f.(*FuncAggregateWithWildcards)
+	aggww.in = NewMock(in)
+	aggww.nodes = nodes
+	return f
+}
+
+func testAggregateWithWildcards(name, agg string, in []models.Series, out models.Series, t *testing.T, xFilesFactor float64, nodes []expr) {
+	inputCopy := models.SeriesCopy(in) // to later verify that it is unchanged
+
+	f := makeAggregateWithWildcards(agg, in, xFilesFactor, nodes)
+
+	dataMap := initDataMap(in)
+
+	got, err := f.Exec(dataMap)
+	if err := equalOutput([]models.Series{out}, got, nil, err); err != nil {
+		t.Fatalf("Case %s: %s", name, err)
+	}
+
+	t.Run("DidNotModifyInput", func(t *testing.T) {
+		if err := equalOutput(inputCopy, in, nil, nil); err != nil {
+			t.Fatalf("Case %s: Input was modified, err = %s", name, err)
+		}
+	})
+
+	t.Run("DoesNotDoubleReturnPoints", func(t *testing.T) {
+		if err := dataMap.CheckForOverlappingPoints(); err != nil {
+			t.Fatalf("Case %s: Point slices in datamap overlap, err = %s", name, err)
+		}
+	})
+	t.Run("OutputIsCanonical", func(t *testing.T) {
+		for i, s := range got {
+			if !s.IsCanonical() {
+				t.Fatalf("Case %s: output series %d is not canonical: %v", name, i, s)
+			}
+		}
+	})
+
+}