[query] Fix graphite summarize/summarize to add missing agg functions (…

…#3583)

src/query/graphite/common/aggregation.go

@@ -79,6 +79,7 @@ var SafeAggregationFns = map[string]SafeAggregationFn{
 	"range":    SafeRange,
 	"multiply": SafeMul,
 	"last":     SafeLast,
+	"count":    SafeCount,
 }
 
 // SafeSort sorts the input slice and returns the number of NaNs in the input.
@@ -246,6 +247,16 @@ func SafeLast(input []float64) (float64, int, bool) {
 	return safeValues[len(safeValues)-1], nans, true
 }
 
+// SafeCount returns the number of valid values in the input slice and the number of NaNs in the input.
+func SafeCount(input []float64) (float64, int, bool) {
+	safeValues, nans, ok := safeValues(input)
+	if !ok {
+		return 0, 0, false
+	}
+
+	return float64(len(safeValues)), nans, true
+}
+
 func safeValues(input []float64) ([]float64, int, bool) {
 	nans := 0
 	safeValues := make([]float64, 0, len(input))

src/query/graphite/common/aggregation_test.go

@@ -167,6 +167,16 @@ func TestAggregationFunctions(t *testing.T) {
 				1,
 			},
 		},
+		{
+			input: input{
+				"count",
+				[]float64{1, 2, 3, math.NaN()},
+			},
+			output: output{
+				3,
+				1,
+			},
+		},
 	}
 
 	for _, test := range tests {

src/query/graphite/native/summarize.go

@@ -42,20 +42,19 @@ func summarize(
 		fname = "sum"
 	}
 
+	safeAggFn, ok := common.SafeAggregationFns[fname]
+	if !ok {
+		return ts.NewSeriesList(), errors.NewInvalidParamsError(fmt.Errorf(
+			"aggregate function not supported: %s", fname))
+	}
+
 	interval, err := common.ParseInterval(intervalS)
 	if err != nil || interval <= 0 {
 		err := errors.NewInvalidParamsError(fmt.Errorf(
 			"invalid interval %s: %v", interval, err))
 		return ts.NewSeriesList(), err
 	}
 
-	f, fexists := summarizeFuncs[fname]
-	if !fexists {
-		err := errors.NewInvalidParamsError(fmt.Errorf(
-			"invalid func %s", fname))
-		return ts.NewSeriesList(), err
-	}
-
 	alignString := ""
 	if alignToFrom {
 		alignString = ", true"
@@ -64,7 +63,7 @@ func summarize(
 	results := make([]*ts.Series, len(series.Values))
 	for i, series := range series.Values {
 		name := fmt.Sprintf("summarize(%s, \"%s\", \"%s\"%s)", series.Name(), intervalS, fname, alignString)
-		results[i] = summarizeTimeSeries(ctx, name, series, interval, f, alignToFrom)
+		results[i] = summarizeTimeSeries(ctx, name, series, interval, safeAggFn, alignToFrom)
 	}
 
 	r := ts.SeriesList(series)
@@ -73,26 +72,22 @@ func summarize(
 }
 
 type summarizeBucket struct {
-	count int
-	accum float64
-	vals  []float64
+	vals []float64
 }
 
 func summarizeTimeSeries(
 	ctx *common.Context,
 	newName string,
 	series *ts.Series,
 	interval time.Duration,
-	funcInfo funcInfo,
+	safeAggFn common.SafeAggregationFn,
 	alignToFrom bool,
 ) *ts.Series {
 	var (
 		startTimeInSecs = int(series.StartTime().Unix())
 		intervalInSecs  = int(interval / time.Second)
 		intervalInMsecs = intervalInSecs * 1000
 		buckets         = make(map[int]*summarizeBucket)
-		f               = funcInfo.consolidationFunc
-		fname           = funcInfo.fname
 	)
 
 	for i := 0; i < series.Len(); i++ {
@@ -107,15 +102,9 @@ func summarizeTimeSeries(
 		}
 
 		if bucket, exists := buckets[bucketInterval]; exists {
-			if fname == "median" {
-				bucket.vals = append(bucket.vals, n)
-			} else {
-				bucket.accum = f(bucket.accum, n, bucket.count)
-			}
-			bucket.count++
+			bucket.vals = append(bucket.vals, n)
 		} else {
-			vals := []float64{n}
-			buckets[bucketInterval] = &summarizeBucket{1, n, vals}
+			buckets[bucketInterval] = &summarizeBucket{[]float64{n}}
 		}
 	}
 
@@ -146,10 +135,9 @@ func summarizeTimeSeries(
 
 		bucket, bucketExists := buckets[bucketInterval]
 		if bucketExists {
-			if fname == "median" {
-				newValues.SetValueAt(i, ts.Median(bucket.vals, bucket.count))
-			} else {
-				newValues.SetValueAt(i, bucket.accum)
+			safeValue, _, safe := safeAggFn(bucket.vals)
+			if safe {
+				newValues.SetValueAt(i, safeValue)
 			}
 		}
 	}
@@ -193,63 +181,3 @@ func sumSpecificationFunc(series ts.SeriesList) string {
 func averageSpecificationFunc(series ts.SeriesList) string {
 	return wrapPathExpr("averageSeries", series)
 }
-
-func maxSpecificationFunc(series ts.SeriesList) string {
-	return wrapPathExpr("maxSeries", series)
-}
-
-func minSpecificationFunc(series ts.SeriesList) string {
-	return wrapPathExpr("minSeries", series)
-}
-
-func lastSpecificationFunc(series ts.SeriesList) string {
-	return wrapPathExpr("lastSeries", series)
-}
-
-type funcInfo struct {
-	fname             string
-	consolidationFunc ts.ConsolidationFunc
-	specificationFunc specificationFunc
-}
-
-var (
-	sumFuncInfo = funcInfo{
-		fname:             "sum",
-		consolidationFunc: ts.Sum,
-		specificationFunc: sumSpecificationFunc,
-	}
-	maxFuncInfo = funcInfo{
-		fname:             "max",
-		consolidationFunc: ts.Max,
-		specificationFunc: maxSpecificationFunc,
-	}
-	minFuncInfo = funcInfo{
-		fname:             "min",
-		consolidationFunc: ts.Min,
-		specificationFunc: minSpecificationFunc,
-	}
-	lastFuncInfo = funcInfo{
-		fname:             "last",
-		consolidationFunc: ts.Last,
-		specificationFunc: lastSpecificationFunc,
-	}
-	avgFuncInfo = funcInfo{
-		fname:             "avg",
-		consolidationFunc: ts.Avg,
-		specificationFunc: averageSpecificationFunc,
-	}
-
-	summarizeFuncs = map[string]funcInfo{
-		"sum":           sumFuncInfo,
-		"max":           maxFuncInfo,
-		"min":           minFuncInfo,
-		"last":          lastFuncInfo,
-		"avg":           avgFuncInfo,
-		"average":       avgFuncInfo,
-		"sumSeries":     sumFuncInfo,
-		"maxSeries":     maxFuncInfo,
-		"minSeries":     minFuncInfo,
-		"averageSeries": avgFuncInfo,
-		"":              sumFuncInfo,
-	}
-)

src/query/graphite/native/summarize_test.go

@@ -150,6 +150,20 @@ func TestSmartSummarize(t *testing.T) {
 			end,
 			[]float64{15, 51},
 		},
+		{"smartSummarize(foo, \"40s\", \"median\")",
+			"40s",
+			"median",
+			start,
+			end,
+			[]float64{1.5, 5.5, 9.5},
+		},
+		{"smartSummarize(foo, \"30s\", \"median\")",
+			"30s",
+			"median",
+			start,
+			end,
+			[]float64{1, 4, 7, 10},
+		},
 	}
 
 	for _, test := range tests {