added graphite function groupByNodes

src/query/graphite/native/aggregation_functions.go

@@ -281,6 +281,68 @@ func groupByNode(ctx *common.Context, series singlePathSpec, node int, fname str
 	return r, nil
 }
 
+func groupByNodes(ctx *common.Context, series singlePathSpec, fname string, nodes ...int) (ts.SeriesList, error) {
+
+	if len(nodes) == 1 {
+		return groupByNode(ctx, series, nodes[0], fname)
+	}
+
+	metaSeries := make(map[string][]*ts.Series)
+	for _, s := range series.Values {
+		parts := strings.Split(s.Name(), ".")
+
+		var keys []string
+
+		for i := 0; i < len(nodes); i++ {
+			n := nodes[i]
+			if n < 0 {
+				n = len(parts) + n
+			}
+
+			if n >= len(parts) || n < 0 {
+				err := errors.NewInvalidParamsError(fmt.Errorf("could not group %s by node %d; not enough parts", s.Name(), nodes[0]))
+				return ts.NewSeriesList(), err
+			}
+			keys = append(keys, parts[n])
+		}
+		key := strings.Join(keys, ".")
+		metaSeries[key] = append(metaSeries[key], s)
+	}
+
+	if fname == "" {
+		fname = "sum"
+	}
+
+	f, fexists := summarizeFuncs[fname]
+	if !fexists {
+		return ts.NewSeriesList(), errors.NewInvalidParamsError(fmt.Errorf("invalid func %s", fname))
+	}
+
+	newSeries := make([]*ts.Series, 0, len(metaSeries))
+	for key, metaSeries := range metaSeries {
+		seriesList := ts.SeriesList{
+			Values:   metaSeries,
+			Metadata: series.Metadata,
+		}
+		output, err := combineSeries(ctx, multiplePathSpecs(seriesList), key, f.consolidationFunc)
+		if err != nil {
+			return ts.NewSeriesList(), err
+		}
+		output.Values[0].Specification = f.specificationFunc(seriesList)
+		newSeries = append(newSeries, output.Values...)
+	}
+
+	r := ts.SeriesList(series)
+
+	r.Values = newSeries
+
+	// Ranging over hash map to create results destroys
+	// any sort order on the incoming series list
+	r.SortApplied = false
+
+	return r, nil
+}
+
 // combineSeries combines multiple series into a single series using a
 // consolidation func.  If the series use different time intervals, the
 // coarsest time will apply.

src/query/graphite/native/aggregation_functions_test.go

@@ -401,6 +401,91 @@ func TestGroupByNode(t *testing.T) {
 	}
 }
 
+func TestGroupByNodes(t *testing.T) {
+	var (
+		start, _ = time.Parse(time.RFC1123, "Mon, 27 Jul 2015 19:41:19 GMT")
+		end, _   = time.Parse(time.RFC1123, "Mon, 27 Jul 2015 19:43:19 GMT")
+		ctx      = common.NewContext(common.ContextOptions{Start: start, End: end})
+		inputs   = []*ts.Series{
+			ts.NewSeries(ctx, "servers.foo-1.pod1.status.500", start,
+				ts.NewConstantValues(ctx, 2, 12, 10000)),
+			ts.NewSeries(ctx, "servers.foo-2.pod1.status.500", start,
+				ts.NewConstantValues(ctx, 4, 12, 10000)),
+			ts.NewSeries(ctx, "servers.foo-3.pod1.status.500", start,
+				ts.NewConstantValues(ctx, 6, 12, 10000)),
+			ts.NewSeries(ctx, "servers.foo-1.pod2.status.500", start,
+				ts.NewConstantValues(ctx, 8, 12, 10000)),
+			ts.NewSeries(ctx, "servers.foo-2.pod2.status.500", start,
+				ts.NewConstantValues(ctx, 10, 12, 10000)),
+
+			ts.NewSeries(ctx, "servers.foo-1.pod1.status.400", start,
+				ts.NewConstantValues(ctx, 20, 12, 10000)),
+			ts.NewSeries(ctx, "servers.foo-2.pod1.status.400", start,
+				ts.NewConstantValues(ctx, 30, 12, 10000)),
+			ts.NewSeries(ctx, "servers.foo-3.pod2.status.400", start,
+				ts.NewConstantValues(ctx, 40, 12, 10000)),
+		}
+	)
+	defer ctx.Close()
+
+	type result struct {
+		name      string
+		sumOfVals float64
+	}
+
+	tests := []struct {
+		fname           string
+		nodes           []int
+		expectedResults []result
+	}{
+		{"sum", []int{4}, []result{
+			{"400", (20 + 30 + 40) * 12},
+			{"500", (2 + 4 + 6 + 8 + 10) * 12},
+		}},
+		{"sum", []int{1, 2}, []result{
+			{"foo-1.pod1", (2 + 20) * 12},
+			{"foo-1.pod2", 8 * 12},
+			{"foo-2.pod1", (4 + 30) * 12},
+			{"foo-2.pod2", 10 * 12},
+			{"foo-3.pod1", 6 * 12},
+			{"foo-3.pod2", 40 * 12},
+		}},
+		{"sum", []int{1, 2, 3}, []result{
+			{"foo-1.pod1.status", (2 + 20) * 12},
+			{"foo-1.pod2.status", 8 * 12},
+			{"foo-2.pod1.status", (4 + 30) * 12},
+			{"foo-2.pod2.status", 10 * 12},
+			{"foo-3.pod1.status", 6 * 12},
+			{"foo-3.pod2.status", 40 * 12},
+		}},
+		{"avg", []int{1, 2}, []result{
+			{"foo-1.pod1", ((2 + 20) / 2) * 12},
+			{"foo-1.pod2", 8 * 12},
+			{"foo-2.pod1", ((4 + 30) / 2) * 12},
+			{"foo-2.pod2", 10 * 12},
+			{"foo-3.pod1", 6 * 12},
+			{"foo-3.pod2", 40 * 12},
+		}},
+	}
+
+	for _, test := range tests {
+		outSeries, err := groupByNodes(ctx, singlePathSpec{
+			Values: inputs,
+		}, test.fname, test.nodes...)
+		require.NoError(t, err)
+		require.Equal(t, len(test.expectedResults), len(outSeries.Values))
+
+		outSeries, _ = sortByName(ctx, singlePathSpec(outSeries))
+		for i, expected := range test.expectedResults {
+			series := outSeries.Values[i]
+			assert.Equal(t, expected.name, series.Name(),
+				"wrong name for %d %s (%d)", test.nodes, test.fname, i)
+			assert.Equal(t, expected.sumOfVals, series.SafeSum(),
+				"wrong result for %d %s (%d)", test.nodes, test.fname, i)
+		}
+	}
+}
+
 func TestWeightedAverage(t *testing.T) {
 	ctx, _ := newConsolidationTestSeries()
 	defer ctx.Close()

src/query/graphite/native/builtin_functions.go

@@ -557,8 +557,8 @@ func lowestCurrent(_ *common.Context, input singlePathSpec, n int) (ts.SeriesLis
 type windowSizeFunc func(stepSize int) int
 
 type windowSizeParsed struct {
-	deltaValue time.Duration
-	stringValue string
+	deltaValue     time.Duration
+	stringValue    string
 	windowSizeFunc windowSizeFunc
 }
 
@@ -1892,6 +1892,7 @@ func init() {
 	MustRegisterFunction(fallbackSeries)
 	MustRegisterFunction(group)
 	MustRegisterFunction(groupByNode)
+	MustRegisterFunction(groupByNodes)
 	MustRegisterFunction(highestAverage)
 	MustRegisterFunction(highestCurrent)
 	MustRegisterFunction(highestMax)

src/query/graphite/native/builtin_functions_test.go

@@ -2934,6 +2934,7 @@ func TestFunctionsRegistered(t *testing.T) {
 		"fallbackSeries",
 		"group",
 		"groupByNode",
+		"groupByNodes",
 		"highestAverage",
 		"highestCurrent",
 		"highestMax",