Merge branch 'highestlowest'

consolidation/consolidation.go

@@ -107,8 +107,8 @@ func FromConsolidateBy(c string) Consolidator {
 		return Avg
 	case "cnt":
 		return Cnt // bonus. not supported by graphite
-	case "lst", "last":
-		return Lst // bonus. not supported by graphite
+	case "lst", "last", "current":
+		return Lst
 	case "min":
 		return Min
 	case "max":
@@ -121,9 +121,9 @@ func FromConsolidateBy(c string) Consolidator {
 		return Diff
 	case "stddev":
 		return StdDev
-	case "range":
+	case "range", "rangeOf":
 		return Range
-	case "sum":
+	case "sum", "total":
 		return Sum
 	}
 	return None
@@ -160,17 +160,17 @@ func GetAggFunc(consolidator Consolidator) batch.AggFunc {
 }
 
 func Validate(fn string) error {
-	if fn == "avg" ||
-		fn == "average" ||
-		fn == "count" || fn == "last" || // bonus
+	if fn == "avg" || fn == "average" ||
+		fn == "count" ||
+		fn == "last" || fn == "current" ||
 		fn == "min" ||
 		fn == "max" ||
 		fn == "mult" || fn == "multiply" ||
 		fn == "med" || fn == "median" ||
 		fn == "diff" ||
 		fn == "stddev" ||
-		fn == "range" ||
-		fn == "sum" {
+		fn == "range" || fn == "rangeOf" ||
+		fn == "sum" || fn == "total" {
 		return nil
 	}
 	return errUnknownConsolidationFunction

docs/graphite.md

@@ -43,7 +43,14 @@ Here are the currently included functions:
 | filterSeries(seriesList, func, operator, threshold) seriesList |             | Stable     |
 | grep(seriesList, pattern) seriesList                           |             | Stable     |
 | groupByTags(seriesList, func, tagList) seriesList              |             | Stable     |
+| highest(seriesList, n, func) seriesList                        |             | Stable     |
+| highestAverage(seriesList, n, func) seriesList                 |             | Stable     |
+| highestCurrent(seriesList, n, func) seriesList                 |             | Stable     |
+| highestMax(seriesList, n, func) seriesList                     |             | Stable     |
 | isNonNull(seriesList) seriesList                               |             | Stable     |
+| lowest(seriesList, n, func) seriesList                         |             | Stable     |
+| lowestAverage(seriesList, n, func) seriesList                  |             | Stable     |
+| lowestCurrent(seriesList, n, func) seriesList                  |             | Stable     |
 | maxSeries(seriesList) series                                   | max         | Stable     |
 | minSeries(seriesList) series                                   | min         | Stable     |
 | multiplySeries(seriesList) series                              |             | Stable     |
@@ -58,4 +65,3 @@ Here are the currently included functions:
 | sumSeries(seriesLists) series                                  | sum         | Stable     |
 | summarize(seriesList) seriesList                               |             | Stable     |
 | transformNull(seriesList, default=0) seriesList                |             | Stable     |
-

expr/func_highestlowest.go

@@ -0,0 +1,87 @@
+package expr
+
+import (
+	"math"
+	"sort"
+
+	"github.com/grafana/metrictank/consolidation"
+
+	"github.com/grafana/metrictank/api/models"
+)
+
+type FuncHighestLowest struct {
+	in      GraphiteFunc
+	n       int64
+	fn      string
+	highest bool
+}
+
+func NewHighestLowestConstructor(fn string, highest bool) func() GraphiteFunc {
+	return func() GraphiteFunc {
+		return &FuncHighestLowest{fn: fn, highest: highest}
+	}
+}
+
+func (s *FuncHighestLowest) Signature() ([]Arg, []Arg) {
+	if s.fn != "" {
+		return []Arg{
+			ArgSeriesList{val: &s.in},
+			ArgInt{key: "n", val: &s.n},
+		}, []Arg{ArgSeriesList{}}
+	}
+	return []Arg{
+		ArgSeriesList{val: &s.in},
+		ArgInt{key: "n", val: &s.n},
+		ArgString{key: "func", val: &s.fn, validator: []Validator{IsConsolFunc}},
+	}, []Arg{ArgSeriesList{}}
+}
+
+func (s *FuncHighestLowest) Context(context Context) Context {
+	return context
+}
+
+type ScoredSeries struct {
+	score float64
+	serie models.Series
+}
+
+func (s *FuncHighestLowest) Exec(cache map[Req][]models.Series) ([]models.Series, error) {
+	series, err := s.in.Exec(cache)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(series) == 0 {
+		return series, nil
+	}
+
+	consolidationFunc := consolidation.GetAggFunc(consolidation.FromConsolidateBy(s.fn))
+
+	// score series by their consolidated value
+	scored := make([]ScoredSeries, len(series))
+	for i, serie := range series {
+		scored[i] = ScoredSeries{
+			score: consolidationFunc(serie.Datapoints),
+			serie: serie,
+		}
+	}
+
+	sort.SliceStable(scored, func(i, j int) bool {
+		iVal := scored[i].score
+		jVal := scored[j].score
+		if s.highest {
+			return math.IsNaN(jVal) && !math.IsNaN(iVal) || iVal > jVal
+		}
+		return math.IsNaN(jVal) && !math.IsNaN(iVal) || iVal < jVal
+	})
+
+	if s.n > int64(len(series)) {
+		s.n = int64(len(series))
+	}
+
+	for i := 0; i < int(s.n); i++ {
+		series[i] = scored[i].serie
+	}
+
+	return series[:s.n], nil
+}