Implement removeAboveValue/removeBelowValue

pkg/datasource/functions.go

@@ -57,12 +57,14 @@ var frontendFuncMap map[string]bool
 
 func init() {
 	seriesFuncMap = map[string]DataProcessingFunc{
-		"groupBy":       applyGroupBy,
-		"scale":         applyScale,
-		"offset":        applyOffset,
-		"transformNull": applyTransformNull,
-		"percentile":    applyPercentile,
-		"timeShift":     applyTimeShiftPost,
+		"groupBy":          applyGroupBy,
+		"scale":            applyScale,
+		"offset":           applyOffset,
+		"removeAboveValue": applyRemoveAboveValue,
+		"removeBelowValue": applyRemoveBelowValue,
+		"transformNull":    applyTransformNull,
+		"percentile":       applyPercentile,
+		"timeShift":        applyTimeShiftPost,
 	}
 
 	aggFuncMap = map[string]AggDataProcessingFunc{
@@ -203,6 +205,26 @@ func applyOffset(series timeseries.TimeSeries, params ...interface{}) (timeserie
 	return series.Transform(transformFunc), nil
 }
 
+func applyRemoveAboveValue(series timeseries.TimeSeries, params ...interface{}) (timeseries.TimeSeries, error) {
+	threshold, err := MustFloat64(params[0])
+	if err != nil {
+		return nil, errParsingFunctionParam(err)
+	}
+
+	transformFunc := timeseries.TransformRemoveAboveValue(threshold)
+	return series.Transform(transformFunc), nil
+}
+
+func applyRemoveBelowValue(series timeseries.TimeSeries, params ...interface{}) (timeseries.TimeSeries, error) {
+	threshold, err := MustFloat64(params[0])
+	if err != nil {
+		return nil, errParsingFunctionParam(err)
+	}
+
+	transformFunc := timeseries.TransformRemoveBelowValue(threshold)
+	return series.Transform(transformFunc), nil
+}
+
 func applyTransformNull(series timeseries.TimeSeries, params ...interface{}) (timeseries.TimeSeries, error) {
 	nullValue, err := MustFloat64(params[0])
 	if err != nil {