[query] Implemented the Graphite highest and lowest functions

src/query/graphite/native/builtin_functions.go

@@ -562,36 +562,52 @@ func takeByFunction(input singlePathSpec, n int, sr ts.SeriesReducer, sort ts.Di
 	return common.Head(r, n)
 }
 
+func getReducer(f string) (ts.SeriesReducer, error) {
+	sa := ts.SeriesReducerApproach(f)
+	r, ok := sa.SafeReducer()
+	if !ok {
+		return r, errors.NewInvalidParamsError(fmt.Errorf("invalid function %s", f))
+	}
+	return r, nil
+}
+
+// highest takes one metric or a wildcard seriesList followed by an integer N and an aggregation function.
+// Out of all metrics passed, draws only the N metrics with the highest
+// aggregated value over the time period specified.
+func highest(_ *common.Context, input singlePathSpec, n int, f string) (ts.SeriesList, error) {
+	reducer, err := getReducer(f)
+	if err != nil {
+		return ts.NewSeriesList(), err
+	}
+	return takeByFunction(input, n, reducer, ts.Descending)
+}
+
 // highestSum takes one metric or a wildcard seriesList followed by an integer
 // n.  Out of all metrics passed, draws only the N metrics with the highest
 // total value in the time period specified.
-func highestSum(_ *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
-	sr := ts.SeriesReducerSum.Reducer()
-	return takeByFunction(input, n, sr, ts.Descending)
+func highestSum(ctx *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
+	return highest(ctx, input, n, "sum")
 }
 
 // highestMax takes one metric or a wildcard seriesList followed by an integer
 // n.  Out of all metrics passed, draws only the N metrics with the highest
 // maximum value in the time period specified.
-func highestMax(_ *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
-	sr := ts.SeriesReducerMax.Reducer()
-	return takeByFunction(input, n, sr, ts.Descending)
+func highestMax(ctx *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
+	return highest(ctx, input, n, "max")
 }
 
 // highestCurrent takes one metric or a wildcard seriesList followed by an
 // integer n.  Out of all metrics passed, draws only the N metrics with the
 // highest value at the end of the time period specified.
-func highestCurrent(_ *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
-	sr := ts.SeriesReducerLast.Reducer()
-	return takeByFunction(input, n, sr, ts.Descending)
+func highestCurrent(ctx *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
+	return highest(ctx, input, n, "current")
 }
 
 // highestAverage takes one metric or a wildcard seriesList followed by an
 // integer n.  Out of all metrics passed, draws only the top N metrics with the
 // highest average value for the time period specified.
-func highestAverage(_ *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
-	sr := ts.SeriesReducerAvg.Reducer()
-	return takeByFunction(input, n, sr, ts.Descending)
+func highestAverage(ctx *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
+	return highest(ctx, input, n, "average")
 }
 
 // fallbackSeries takes one metric or a wildcard seriesList, and a second fallback metric.
@@ -608,25 +624,33 @@ func fallbackSeries(_ *common.Context, input singlePathSpec, fallback singlePath
 // N. Draws the N most deviant metrics.  To find the deviants, the standard
 // deviation (sigma) of each series is taken and ranked.  The top N standard
 // deviations are returned.
-func mostDeviant(_ *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
-	sr := ts.SeriesReducerStdDev.Reducer()
-	return takeByFunction(input, n, sr, ts.Descending)
+func mostDeviant(ctx *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
+	return highest(ctx, input, n, "stddev")
+}
+
+// lowest takes one metric or a wildcard seriesList followed by an integer N and an aggregation function.
+// Out of all metrics passed, draws only the N metrics with the lowest
+// aggregated value over the time period specified.
+func lowest(_ *common.Context, input singlePathSpec, n int, f string) (ts.SeriesList, error) {
+	reducer, err := getReducer(f)
+	if err != nil {
+		return ts.NewSeriesList(), err
+	}
+	return takeByFunction(input, n, reducer, ts.Ascending)
 }
 
 // lowestAverage takes one metric or a wildcard seriesList followed by an
 // integer n.  Out of all metrics passed, draws only the top N metrics with the
 // lowest average value for the time period specified.
-func lowestAverage(_ *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
-	sr := ts.SeriesReducerAvg.Reducer()
-	return takeByFunction(input, n, sr, ts.Ascending)
+func lowestAverage(ctx *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
+	return lowest(ctx, input, n, "average")
 }
 
 // lowestCurrent takes one metric or a wildcard seriesList followed by an
 // integer n.  Out of all metrics passed, draws only the N metrics with the
 // lowest value at the end of the time period specified.
-func lowestCurrent(_ *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
-	sr := ts.SeriesReducerLast.Reducer()
-	return takeByFunction(input, n, sr, ts.Ascending)
+func lowestCurrent(ctx *common.Context, input singlePathSpec, n int) (ts.SeriesList, error) {
+	return lowest(ctx, input, n, "current")
 }
 
 // windowSizeFunc calculates window size for moving average calculation
@@ -2173,6 +2197,10 @@ func init() {
 	MustRegisterFunction(group)
 	MustRegisterFunction(groupByNode)
 	MustRegisterFunction(groupByNodes)
+	MustRegisterFunction(highest).WithDefaultParams(map[uint8]interface{}{
+		2: 1,         // n,
+		3: "average", // f
+	})
 	MustRegisterFunction(highestAverage)
 	MustRegisterFunction(highestCurrent)
 	MustRegisterFunction(highestMax)
@@ -2192,6 +2220,10 @@ func init() {
 	MustRegisterFunction(logarithm).WithDefaultParams(map[uint8]interface{}{
 		2: 10, // base
 	})
+	MustRegisterFunction(lowest).WithDefaultParams(map[uint8]interface{}{
+		2: 1,         // n,
+		3: "average", // f
+	})
 	MustRegisterFunction(lowestAverage)
 	MustRegisterFunction(lowestCurrent)
 	MustRegisterFunction(maxSeries)

src/query/graphite/native/builtin_functions_test.go

@@ -1090,10 +1090,20 @@ type nIntParamGoldenData struct {
 	outputs []common.TestSeries
 }
 
+// nIntParamGoldenDataWithAgg holds test data for functions that take an additional "n" int parameter
+// It also holds an aggregation function
+type nIntParamGoldenDataWithAgg struct {
+	nIntParamGoldenData
+	aggFunc string
+}
+
 // rankingFunc selects the n lowest or highest series based on certain metric of the
 // series (e.g., maximum, minimum, average).
 type rankingFunc func(ctx *common.Context, input singlePathSpec, n int) (ts.SeriesList, error)
 
+// testRanking can be used to test the ranking alias functions
+// (e.g. lowestAverage, highestMax, highestAverage, lowestCurrent)
+// these functions are all aliases of the "meta-ranking" functions (i.e. highest and lowest)
 func testRanking(t *testing.T, ctx *common.Context, tests []nIntParamGoldenData, f rankingFunc) {
 	start := time.Now()
 	step := 100
@@ -1113,6 +1123,73 @@ func testRanking(t *testing.T, ctx *common.Context, tests []nIntParamGoldenData,
 	}
 }
 
+// testOrderedAggregationFunc is a helper function for testing lowest and highest
+func testOrderedAggregationFunc(t *testing.T, ctx *common.Context, tests []nIntParamGoldenDataWithAgg, isLowest bool) {
+	f := highest
+	if isLowest {
+		f = lowest
+	}
+
+	start := time.Now()
+	step := 100
+	for _, test := range tests {
+		input := singlePathSpec{Values: generateSeriesList(ctx, start, test.inputs, step)}
+		outputs, err := f(ctx, input, test.n, test.aggFunc)
+
+		if test.n < 0 {
+			require.NotNil(t, err)
+			require.Equal(t, "n must be positive", err.Error())
+			assert.Nil(t, outputs.Values, "Nil timeseries should be returned")
+			continue
+		}
+
+		require.NoError(t, err)
+		common.CompareOutputsAndExpected(t, step, start,
+			test.outputs, outputs.Values)
+	}
+}
+
+func TestHighest(t *testing.T) {
+	ctx := common.NewTestContext()
+	defer ctx.Close()
+
+	tests := []nIntParamGoldenDataWithAgg{
+		{
+			nIntParamGoldenData{
+				testInput,
+				0,
+				nil,
+			},
+			"sum",
+		},
+		{
+			nIntParamGoldenData{
+				testInput,
+				1,
+				[]common.TestSeries{testInput[0]},
+			},
+			"current",
+		},
+		{
+			nIntParamGoldenData{
+				testInput,
+				2,
+				[]common.TestSeries{testInput[4], testInput[2]},
+			},
+			"average",
+		},
+		{
+			nIntParamGoldenData{
+				testInput,
+				len(testInput) + 10, // force sort
+				[]common.TestSeries{testInput[0], testInput[3], testInput[4], testInput[2], testInput[1]},
+			},
+			"last",
+		},
+	}
+	testOrderedAggregationFunc(t, ctx, tests, false)
+}
+
 func TestHighestCurrent(t *testing.T) {
 	ctx := common.NewTestContext()
 	defer ctx.Close()
@@ -1283,6 +1360,47 @@ func TestMostDeviant(t *testing.T) {
 	testRanking(t, ctx, tests, mostDeviant)
 }
 
+func TestLowest(t *testing.T) {
+	ctx := common.NewTestContext()
+	defer ctx.Close()
+
+	tests := []nIntParamGoldenDataWithAgg{
+		{
+			nIntParamGoldenData{
+				testInput,
+				0,
+				nil,
+			},
+			"max",
+		},
+		{
+			nIntParamGoldenData{
+				testInput,
+				2,
+				[]common.TestSeries{testInput[1], testInput[3]},
+			},
+			"sum",
+		},
+		{
+			nIntParamGoldenData{
+				testInput,
+				2,
+				[]common.TestSeries{testInput[1], testInput[2]},
+			},
+			"current",
+		},
+		{
+			nIntParamGoldenData{
+				testInput,
+				3,
+				[]common.TestSeries{testInput[1], testInput[3], testInput[0]},
+			},
+			"average",
+		},
+	}
+	testOrderedAggregationFunc(t, ctx, tests, true)
+}
+
 func TestLowestAverage(t *testing.T) {
 	ctx := common.NewTestContext()
 	defer ctx.Close()
@@ -3159,6 +3277,7 @@ func TestFunctionsRegistered(t *testing.T) {
 		"group",
 		"groupByNode",
 		"groupByNodes",
+		"highest",
 		"highestAverage",
 		"highestCurrent",
 		"highestMax",
@@ -3175,6 +3294,7 @@ func TestFunctionsRegistered(t *testing.T) {
 		"limit",
 		"log",
 		"logarithm",
+		"lowest",
 		"lowestAverage",
 		"lowestCurrent",
 		"max",

src/query/graphite/ts/series_reducer.go

@@ -28,11 +28,15 @@ type SeriesReducerApproach string
 // The standard set of reducers
 const (
 	SeriesReducerAvg    SeriesReducerApproach = "avg"
-	SeriesReducerSum    SeriesReducerApproach = "total"
+	SeriesReducerSum    SeriesReducerApproach = "sum"
 	SeriesReducerMin    SeriesReducerApproach = "min"
 	SeriesReducerMax    SeriesReducerApproach = "max"
 	SeriesReducerStdDev SeriesReducerApproach = "stddev"
 	SeriesReducerLast   SeriesReducerApproach = "last"
+
+	SeriesReducerAverage SeriesReducerApproach = "average" // alias for "avg"
+	SeriesReducerTotal   SeriesReducerApproach = "total"   // alias for "sum"
+	SeriesReducerCurrent SeriesReducerApproach = "current" // alias for "last"
 )
 
 // SeriesReducer reduces a series to a single value.
@@ -55,10 +59,13 @@ func (sa SeriesReducerApproach) Reducer() SeriesReducer {
 }
 
 var seriesReducers = map[SeriesReducerApproach]SeriesReducer{
-	SeriesReducerAvg:    func(b *Series) float64 { return b.SafeAvg() },
-	SeriesReducerSum:    func(b *Series) float64 { return b.SafeSum() },
-	SeriesReducerMin:    func(b *Series) float64 { return b.SafeMin() },
-	SeriesReducerMax:    func(b *Series) float64 { return b.SafeMax() },
-	SeriesReducerStdDev: func(b *Series) float64 { return b.SafeStdDev() },
-	SeriesReducerLast:   func(b *Series) float64 { return b.SafeLastValue() },
+	SeriesReducerAvg:     func(b *Series) float64 { return b.SafeAvg() },
+	SeriesReducerAverage: func(b *Series) float64 { return b.SafeAvg() },
+	SeriesReducerTotal:   func(b *Series) float64 { return b.SafeSum() },
+	SeriesReducerSum:     func(b *Series) float64 { return b.SafeSum() },
+	SeriesReducerMin:     func(b *Series) float64 { return b.SafeMin() },
+	SeriesReducerMax:     func(b *Series) float64 { return b.SafeMax() },
+	SeriesReducerStdDev:  func(b *Series) float64 { return b.SafeStdDev() },
+	SeriesReducerLast:    func(b *Series) float64 { return b.SafeLastValue() },
+	SeriesReducerCurrent: func(b *Series) float64 { return b.SafeLastValue() },
 }