Added missing batch/agg functions

batch/aggregator.go

@@ -6,14 +6,12 @@ package batch
 import (
 	"gopkg.in/raintank/schema.v1"
 	"math"
+	"sort"
 )
 
 type AggFunc func(in []schema.Point) float64
 
 func Avg(in []schema.Point) float64 {
-	if len(in) == 0 {
-		panic("avg() called in aggregator with 0 terms")
-	}
 	valid := float64(0)
 	sum := float64(0)
 	for _, term := range in {
@@ -42,9 +40,6 @@ func Cnt(in []schema.Point) float64 {
 }
 
 func Lst(in []schema.Point) float64 {
-	if len(in) == 0 {
-		panic("last() called in aggregator with 0 terms")
-	}
 	lst := math.NaN()
 	for _, v := range in {
 		if !math.IsNaN(v.Val) {
@@ -55,9 +50,6 @@ func Lst(in []schema.Point) float64 {
 }
 
 func Min(in []schema.Point) float64 {
-	if len(in) == 0 {
-		panic("min() called in aggregator with 0 terms")
-	}
 	valid := false
 	min := math.Inf(1)
 	for _, v := range in {
@@ -75,9 +67,6 @@ func Min(in []schema.Point) float64 {
 }
 
 func Max(in []schema.Point) float64 {
-	if len(in) == 0 {
-		panic("max() called in aggregator with 0 terms")
-	}
 	valid := false
 	max := math.Inf(-1)
 	for _, v := range in {
@@ -94,6 +83,97 @@ func Max(in []schema.Point) float64 {
 	return max
 }
 
+func Mult(in []schema.Point) float64 {
+	if len(in) == 0 {
+		return math.NaN()
+	}
+	mult := float64(1)
+	for _, fact := range in {
+		if math.IsNaN(fact.Val) {
+			// NaN * anything equals NaN()
+			return math.NaN()
+		}
+		mult *= fact.Val
+	}
+	return mult
+}
+
+func Med(in []schema.Point) float64 {
+	med := math.NaN()
+	vals := make([]float64, 0, len(in))
+	for i := 0; i < len(in); i++ {
+		p := in[i].Val
+		if !math.IsNaN(p) {
+			vals = append(vals, p)
+		}
+	}
+	if len(vals) != 0 {
+		sort.Float64s(vals)
+		mid := len(vals) / 2
+		if len(vals)%2 == 0 {
+			med = (vals[mid-1] + vals[mid]) / 2
+		} else {
+			med = vals[mid]
+		}
+	}
+	return med
+}
+
+func Diff(in []schema.Point) float64 {
+	diff := math.NaN()
+	for i := 0; i < len(in); i++ {
+		p := in[i].Val
+		if !math.IsNaN(p) {
+			if math.IsNaN(diff) {
+				diff = p
+			} else {
+				diff -= p
+			}
+		}
+	}
+	return diff
+}
+
+func StdDev(in []schema.Point) float64 {
+	avg := Avg(in)
+	if math.IsNaN(avg) {
+		return avg
+	}
+	num := float64(0)
+	sumDeviationsSquared := float64(0)
+	for i := 0; i < len(in); i++ {
+		p := in[i].Val
+		if !math.IsNaN(p) {
+			num++
+			deviation := p - avg
+			sumDeviationsSquared += deviation * deviation
+		}
+	}
+	std := math.Sqrt(sumDeviationsSquared / num)
+	return std
+}
+
+func Range(in []schema.Point) float64 {
+	valid := false
+	min := math.Inf(1)
+	max := math.Inf(-1)
+	for _, v := range in {
+		if !math.IsNaN(v.Val) {
+			valid = true
+			if v.Val < min {
+				min = v.Val
+			}
+			if v.Val > max {
+				max = v.Val
+			}
+		}
+	}
+	if !valid {
+		return math.NaN()
+	}
+	return max - min
+}
+
 func Sum(in []schema.Point) float64 {
 	valid := false
 	sum := float64(0)

consolidation/consolidation.go

@@ -23,6 +23,11 @@ const (
 	Max
 	Min
 	Cnt // not available through http api
+	Mult
+	Med
+	Diff
+	StdDev
+	Range
 )
 
 // String provides human friendly names
@@ -40,6 +45,16 @@ func (c Consolidator) String() string {
 		return "MinimumConsolidator"
 	case Max:
 		return "MaximumConsolidator"
+	case Mult:
+		return "MultiplyConsolidator"
+	case Med:
+		return "MedianConsolidator"
+	case Diff:
+		return "DifferenceConsolidator"
+	case StdDev:
+		return "StdDevConsolidator"
+	case Range:
+		return "RangeConsolidator"
 	case Sum:
 		return "SumConsolidator"
 	}
@@ -96,6 +111,16 @@ func FromConsolidateBy(c string) Consolidator {
 		return Min
 	case "max":
 		return Max
+	case "mult", "multiply":
+		return Mult
+	case "med", "median":
+		return Med
+	case "diff":
+		return Diff
+	case "stddev":
+		return StdDev
+	case "range":
+		return Range
 	case "sum":
 		return Sum
 	}
@@ -116,14 +141,34 @@ func GetAggFunc(consolidator Consolidator) batch.AggFunc {
 		consFunc = batch.Min
 	case Max:
 		consFunc = batch.Max
+	case Mult:
+		consFunc = batch.Mult
+	case Med:
+		consFunc = batch.Med
+	case Diff:
+		consFunc = batch.Diff
+	case StdDev:
+		consFunc = batch.StdDev
+	case Range:
+		consFunc = batch.Range
 	case Sum:
 		consFunc = batch.Sum
 	}
 	return consFunc
 }
 
 func Validate(fn string) error {
-	if fn == "avg" || fn == "average" || fn == "last" || fn == "min" || fn == "max" || fn == "sum" {
+	if fn == "avg" ||
+		fn == "average" ||
+		fn == "count" || fn == "last" || // bonus
+		fn == "min" ||
+		fn == "max" ||
+		fn == "mult" || fn == "multiply" ||
+		fn == "med" || fn == "median" ||
+		fn == "diff" ||
+		fn == "stddev" ||
+		fn == "range" ||
+		fn == "sum" {
 		return nil
 	}
 	return errUnknownConsolidationFunction