Merge pull request #1324 from DeleurApps/group

Added group function

Author: Dieterbe

docs/graphite.md

@@ -77,7 +77,7 @@ See also:
 | fallbackSeries                                                 |              | Stable     |
 | filterSeries(seriesList, func, operator, threshold) seriesList |              | Stable     |
 | grep(seriesList, pattern) seriesList                           |              | Stable     |
-| group                                                          |              | No         |
+| group                                                          |              | Stable     |
 | groupByNode                                                    |              | No         |
 | groupByNodes                                                   |              | No         |
 | groupByTags(seriesList, func, tagList) seriesList              |              | Stable     |

expr/func_group.go

@@ -0,0 +1,31 @@
+package expr
+
+import (
+	"github.com/grafana/metrictank/api/models"
+)
+
+type FuncGroup struct {
+	in []GraphiteFunc
+}
+
+func NewGroup() GraphiteFunc {
+	return &FuncGroup{}
+}
+
+func (s *FuncGroup) Signature() ([]Arg, []Arg) {
+	return []Arg{
+		ArgSeriesLists{val: &s.in}}, []Arg{ArgSeriesList{}}
+}
+
+func (s *FuncGroup) Context(context Context) Context {
+	return context
+}
+
+func (s *FuncGroup) Exec(cache map[Req][]models.Series) ([]models.Series, error) {
+	series, _, err := consumeFuncs(cache, s.in)
+	if err != nil {
+		return nil, err
+	}
+
+	return series, nil
+}

expr/func_group_test.go

@@ -0,0 +1,158 @@
+package expr
+
+import (
+	"strconv"
+	"testing"
+
+	"github.com/grafana/metrictank/api/models"
+	"github.com/grafana/metrictank/test"
+	"github.com/raintank/schema"
+)
+
+func getNewGroup(in [][]models.Series) *FuncGroup {
+	f := NewGroup()
+	s := f.(*FuncGroup)
+	for i := range in {
+		s.in = append(s.in, NewMock(in[i]))
+	}
+	return s
+}
+
+func TestGroup(t *testing.T) {
+	f := getNewGroup([][]models.Series{
+		{
+			{
+				Interval:   10,
+				QueryPatt:  "a",
+				Target:     "a",
+				Datapoints: getCopy(a),
+			},
+		},
+		{
+			{
+				Interval:   10,
+				QueryPatt:  "b",
+				Target:     "b",
+				Datapoints: getCopy(b),
+			},
+		},
+		{
+			{
+				Interval:   10,
+				QueryPatt:  "abc",
+				Target:     "abc",
+				Datapoints: getCopy(a),
+			},
+			{
+				Interval:   10,
+				QueryPatt:  "abc",
+				Target:     "abc",
+				Datapoints: getCopy(b),
+			},
+			{
+				Interval:   10,
+				QueryPatt:  "abc",
+				Target:     "abc",
+				Datapoints: getCopy(c),
+			},
+		},
+	},
+	)
+	out := []models.Series{
+		{
+			Interval:   10,
+			QueryPatt:  "a",
+			Target:     "a",
+			Datapoints: getCopy(a),
+		},
+		{
+			Interval:   10,
+			QueryPatt:  "b",
+			Target:     "b",
+			Datapoints: getCopy(b),
+		},
+		{
+			Interval:   10,
+			QueryPatt:  "abc",
+			Target:     "abc",
+			Datapoints: getCopy(a),
+		},
+		{
+			Interval:   10,
+			QueryPatt:  "abc",
+			Target:     "abc",
+			Datapoints: getCopy(b),
+		},
+		{
+			Interval:   10,
+			QueryPatt:  "abc",
+			Target:     "abc",
+			Datapoints: getCopy(c),
+		},
+	}
+
+	got, err := f.Exec(make(map[Req][]models.Series))
+	if err := equalOutput(out, got, nil, err); err != nil {
+		t.Fatal(err)
+	}
+}
+func BenchmarkGroup10k_1NoNulls(b *testing.B) {
+	benchmarkGroup(b, 1, test.RandFloats10k, test.RandFloats10k)
+}
+func BenchmarkGroup10k_10NoNulls(b *testing.B) {
+	benchmarkGroup(b, 10, test.RandFloats10k, test.RandFloats10k)
+}
+func BenchmarkGroup10k_100NoNulls(b *testing.B) {
+	benchmarkGroup(b, 100, test.RandFloats10k, test.RandFloats10k)
+}
+func BenchmarkGroup10k_1000NoNulls(b *testing.B) {
+	benchmarkGroup(b, 1000, test.RandFloats10k, test.RandFloats10k)
+}
+func BenchmarkGroup10k_1SomeSeriesHalfNulls(b *testing.B) {
+	benchmarkGroup(b, 1, test.RandFloats10k, test.RandFloatsWithNulls10k)
+}
+func BenchmarkGroup10k_10SomeSeriesHalfNulls(b *testing.B) {
+	benchmarkGroup(b, 10, test.RandFloats10k, test.RandFloatsWithNulls10k)
+}
+func BenchmarkGroup10k_100SomeSeriesHalfNulls(b *testing.B) {
+	benchmarkGroup(b, 100, test.RandFloats10k, test.RandFloatsWithNulls10k)
+}
+func BenchmarkGroup10k_1000SomeSeriesHalfNulls(b *testing.B) {
+	benchmarkGroup(b, 1000, test.RandFloats10k, test.RandFloatsWithNulls10k)
+}
+func BenchmarkGroup10k_1AllSeriesHalfNulls(b *testing.B) {
+	benchmarkGroup(b, 1, test.RandFloatsWithNulls10k, test.RandFloatsWithNulls10k)
+}
+func BenchmarkGroup10k_10AllSeriesHalfNulls(b *testing.B) {
+	benchmarkGroup(b, 10, test.RandFloatsWithNulls10k, test.RandFloatsWithNulls10k)
+}
+func BenchmarkGroup10k_100AllSeriesHalfNulls(b *testing.B) {
+	benchmarkGroup(b, 100, test.RandFloatsWithNulls10k, test.RandFloatsWithNulls10k)
+}
+func BenchmarkGroup10k_1000AllSeriesHalfNulls(b *testing.B) {
+	benchmarkGroup(b, 1000, test.RandFloatsWithNulls10k, test.RandFloatsWithNulls10k)
+}
+func benchmarkGroup(b *testing.B, numSeries int, fn0, fn1 func() []schema.Point) {
+	var input []models.Series
+	for i := 0; i < numSeries; i++ {
+		series := models.Series{
+			QueryPatt: strconv.Itoa(i),
+		}
+		if i%2 == 0 {
+			series.Datapoints = fn0()
+		} else {
+			series.Datapoints = fn1()
+		}
+		input = append(input, series)
+	}
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		f := NewGroup()
+		f.(*FuncGroup).in = append(f.(*FuncGroup).in, NewMock(input))
+		got, err := f.Exec(make(map[Req][]models.Series))
+		if err != nil {
+			b.Fatalf("%s", err)
+		}
+		results = got
+	}
+}

expr/funcs.go

@@ -71,6 +71,7 @@ func init() {
 		"fallbackSeries":        {NewFallbackSeries, true},
 		"filterSeries":          {NewFilterSeries, true},
 		"grep":                  {NewGrep, true},
+		"group":                 {NewGroup, true},
 		"groupByTags":           {NewGroupByTags, true},
 		"highest":               {NewHighestLowestConstructor("", true), true},
 		"highestAverage":        {NewHighestLowestConstructor("average", true), true},