Implement top/bottom

pkg/datasource/functions.go

@@ -45,6 +45,8 @@ var seriesFuncMap map[string]DataProcessingFunc
 
 var aggFuncMap map[string]AggDataProcessingFunc
 
+var filterFuncMap map[string]AggDataProcessingFunc
+
 var frontendFuncMap map[string]bool
 
 func init() {
@@ -61,6 +63,11 @@ func init() {
 		"percentileAgg": applyPercentileAgg,
 	}
 
+	filterFuncMap = map[string]AggDataProcessingFunc{
+		"top":    applyTop,
+		"bottom": applyBottom,
+	}
+
 	// Functions processing on the frontend
 	frontendFuncMap = map[string]bool{
 		"setAlias":        true,
@@ -85,6 +92,12 @@ func applyFunctions(series []*timeseries.TimeSeriesData, functions []QueryFuncti
 				return nil, err
 			}
 			series = result
+		} else if applyFilterFunc, ok := filterFuncMap[f.Def.Name]; ok {
+			result, err := applyFilterFunc(series, f.Params...)
+			if err != nil {
+				return nil, err
+			}
+			series = result
 		} else if _, ok := frontendFuncMap[f.Def.Name]; ok {
 			continue
 		} else {
@@ -207,6 +220,30 @@ func applyPercentileAgg(series []*timeseries.TimeSeriesData, params ...interface
 	return []*timeseries.TimeSeriesData{aggregatedSeries}, nil
 }
 
+func applyTop(series []*timeseries.TimeSeriesData, params ...interface{}) ([]*timeseries.TimeSeriesData, error) {
+	n, err := MustFloat64(params[0])
+	pAgg, err := MustString(params[1])
+	if err != nil {
+		return nil, errParsingFunctionParam(err)
+	}
+
+	aggFunc := getAggFunc(pAgg)
+	filteredSeries := timeseries.Filter(series, int(n), "top", aggFunc)
+	return filteredSeries, nil
+}
+
+func applyBottom(series []*timeseries.TimeSeriesData, params ...interface{}) ([]*timeseries.TimeSeriesData, error) {
+	n, err := MustFloat64(params[0])
+	pAgg, err := MustString(params[1])
+	if err != nil {
+		return nil, errParsingFunctionParam(err)
+	}
+
+	aggFunc := getAggFunc(pAgg)
+	filteredSeries := timeseries.Filter(series, int(n), "bottom", aggFunc)
+	return filteredSeries, nil
+}
+
 func getAggFunc(agg string) timeseries.AggFunc {
 	switch agg {
 	case "avg":