Add filterSeries function

docs/graphite.md

@@ -27,30 +27,33 @@ See also:
 
 Here are the currently included functions:
 
-| Function name and signature                           | Alias       | Metrictank |
-| ----------------------------------------------------- | ----------- | ---------- |
-| alias(seriesList, alias) seriesList                   |             | Stable     |
-| aliasByNode(seriesList, nodeList) seriesList          | aliasByTags | Stable     |
-| aliasSub(seriesList, pattern, replacement) seriesList |             | Stable     |
-| averageSeries(seriesLists) series                     | avg         | Stable     |
-| consolidateBy(seriesList, func) seriesList            |             | Stable     |
-| countSeries(seriesLists) series                       |             | Stable     |
-| diffSeries(seriesLists) series                        |             | Stable     |
-| divideSeries(dividend, divisor) seriesList            |             | Stable     |
-| divideSeriesLists(dividends, divisors) seriesList     |             | Stable     |
-| exclude(seriesList, pattern) seriesList               |             | Stable     |
-| grep(seriesList, pattern) seriesList                  |             | Stable     |
-| groupByTags(seriesList, func, tagList) seriesList     |             | Stable     |
-| isNonNull(seriesList) seriesList                      |             | Stable     |
-| maxSeries(seriesList) series                          | max         | Stable     |
-| minSeries(seriesList) series                          | min         | Stable     |
-| multiplySeries(seriesList) series                     |             | Stable     |
-| movingAverage(seriesLists, windowSize) seriesList     |             | Unstable   |
-| perSecond(seriesLists) seriesList                     |             | Stable     |
-| rangeOfSeries(seriesList) series                      |             | Stable     |
-| scale(seriesList, num) series                         |             | Stable     |
-| scaleToSeconds(seriesList, seconds) series            |             | Stable     |
-| stddevSeries(seriesList) series                       |             | Stable     |
-| sumSeries(seriesLists) series                         | sum         | Stable     |
-| summarize(seriesList) seriesList                      |             | Stable     |
-| transformNull(seriesList, default=0) seriesList       |             | Stable     |
+
+| Function name and signature                                    | Alias       | Metrictank |
+| -------------------------------------------------------------- | ----------- | ---------- |
+| alias(seriesList, alias) seriesList                            |             | Stable     |
+| aliasByNode(seriesList, nodeList) seriesList                   | aliasByTags | Stable     |
+| aliasSub(seriesList, pattern, replacement) seriesList          |             | Stable     |
+| averageSeries(seriesLists) series                              | avg         | Stable     |
+| consolidateBy(seriesList, func) seriesList                     |             | Stable     |
+| countSeries(seriesLists) series                                |             | Stable     |
+| diffSeries(seriesLists) series                                 |             | Stable     |
+| divideSeries(dividend, divisor) seriesList                     |             | Stable     |
+| divideSeriesLists(dividends, divisors) seriesList              |             | Stable     |
+| exclude(seriesList, pattern) seriesList                        |             | Stable     |
+| filterSeries(seriesList, func, operator, threshold) seriesList |             | Stable     |
+| grep(seriesList, pattern) seriesList                           |             | Stable     |
+| groupByTags(seriesList, func, tagList) seriesList              |             | Stable     |
+| isNonNull(seriesList) seriesList                               |             | Stable     |
+| maxSeries(seriesList) series                                   | max         | Stable     |
+| minSeries(seriesList) series                                   | min         | Stable     |
+| multiplySeries(seriesList) series                              |             | Stable     |
+| movingAverage(seriesLists, windowSize) seriesList              |             | Unstable   |
+| perSecond(seriesLists) seriesList                              |             | Stable     |
+| rangeOfSeries(seriesList) series                               |             | Stable     |
+| scale(seriesList, num) series                                  |             | Stable     |
+| scaleToSeconds(seriesList, seconds) series                     |             | Stable     |
+| stddevSeries(seriesList) series                                |             | Stable     |
+| sumSeries(seriesLists) series                                  | sum         | Stable     |
+| summarize(seriesList) seriesList                               |             | Stable     |
+| transformNull(seriesList, default=0) seriesList                |             | Stable     |
+

expr/func_filterseries.go

@@ -0,0 +1,88 @@
+package expr
+
+import (
+	"errors"
+	"math"
+
+	"github.com/grafana/metrictank/api/models"
+	"github.com/grafana/metrictank/consolidation"
+)
+
+type FuncFilterSeries struct {
+	in        GraphiteFunc
+	fn        string
+	operator  string
+	threshold float64
+}
+
+func NewFilterSeries() GraphiteFunc {
+	return &FuncFilterSeries{}
+}
+
+func (s *FuncFilterSeries) Signature() ([]Arg, []Arg) {
+	return []Arg{
+		ArgSeriesList{val: &s.in},
+		ArgString{key: "func", val: &s.fn, validator: []Validator{IsConsolFunc}},
+		ArgString{key: "operator", val: &s.operator, validator: []Validator{IsOperator}},
+		ArgFloat{key: "threshold", val: &s.threshold},
+	}, []Arg{ArgSeriesList{}}
+}
+
+func (s *FuncFilterSeries) Context(context Context) Context {
+	return context
+}
+
+func getOperatorFunc(operator string) (func(float64, float64) bool, error) {
+	switch operator {
+	case "=":
+		return func(val, threshold float64) bool {
+			return val == threshold
+		}, nil
+
+	case "!=":
+		return func(val, threshold float64) bool {
+			return val != threshold
+		}, nil
+
+	case ">":
+		return func(val, threshold float64) bool {
+			return val > threshold
+		}, nil
+	case ">=":
+		return func(val, threshold float64) bool {
+			return val >= threshold
+		}, nil
+
+	case "<":
+		return func(val, threshold float64) bool {
+			return math.IsNaN(val) || val < threshold
+		}, nil
+	case "<=":
+		return func(val, threshold float64) bool {
+			return math.IsNaN(val) || val <= threshold
+		}, nil
+	}
+	return func(v1, v2 float64) bool { return false }, errors.New("Unsupported operator: " + operator)
+}
+
+func (s *FuncFilterSeries) Exec(cache map[Req][]models.Series) ([]models.Series, error) {
+	series, err := s.in.Exec(cache)
+	if err != nil {
+		return nil, err
+	}
+
+	consolidationFunc := consolidation.GetAggFunc(consolidation.FromConsolidateBy(s.fn))
+	operatorFunc, err := getOperatorFunc(s.operator)
+	if err != nil {
+		return nil, err
+	}
+
+	out := make([]models.Series, 0, len(series))
+	for _, serie := range series {
+		if operatorFunc(consolidationFunc(serie.Datapoints), s.threshold) {
+			out = append(out, serie)
+		}
+	}
+
+	return out, nil
+}