feat(alerting): add timeserie aggregation functions

pkg/models/timeseries.go

@@ -1,8 +1,47 @@
 package models
 
+import "math"
+
 type TimeSeries struct {
 	Name   string       `json:"name"`
 	Points [][2]float64 `json:"points"`
+
+	Avg  float64
+	Sum  float64
+	Min  float64
+	Max  float64
+	Mean float64
 }
 
 type TimeSeriesSlice []*TimeSeries
+
+func NewTimeSeries(name string, points [][2]float64) *TimeSeries {
+	ts := &TimeSeries{
+		Name:   name,
+		Points: points,
+	}
+
+	ts.Min = points[0][0]
+	ts.Max = points[0][0]
+
+	for _, v := range points {
+		value := v[0]
+
+		if value > ts.Max {
+			ts.Max = value
+		}
+
+		if value < ts.Min {
+			ts.Min = value
+		}
+
+		ts.Sum += value
+	}
+
+	ts.Avg = ts.Sum / float64(len(points))
+	midPosition := int64(math.Floor(float64(len(points)) / float64(2)))
+
+	ts.Mean = points[midPosition][0]
+
+	return ts
+}

pkg/models/timeseries_test.go

@@ -0,0 +1,36 @@
+package models
+
+import (
+	. "github.com/smartystreets/goconvey/convey"
+	"testing"
+)
+
+func TestTimeSeries(t *testing.T) {
+	Convey("timeseries aggregation tests", t, func() {
+		ts := NewTimeSeries("test", [][2]float64{
+			{1, 0},
+			{2, 0},
+			{3, 0},
+		})
+
+		Convey("sum", func() {
+			So(ts.Sum, ShouldEqual, 6)
+		})
+
+		Convey("avg", func() {
+			So(ts.Avg, ShouldEqual, 2)
+		})
+
+		Convey("min", func() {
+			So(ts.Min, ShouldEqual, 1)
+		})
+
+		Convey("max", func() {
+			So(ts.Max, ShouldEqual, 3)
+		})
+
+		Convey("mean", func() {
+			So(ts.Mean, ShouldEqual, 2)
+		})
+	})
+}

pkg/services/alerting/executor.go

@@ -11,6 +11,15 @@ type Executor interface {
 
 type ExecutorImpl struct{}
 
+type fn func(float64, float64) bool
+
+var operators map[string]fn = map[string]fn{
+	">":  func(num1, num2 float64) bool { return num1 > num2 },
+	">=": func(num1, num2 float64) bool { return num1 >= num2 },
+	"<":  func(num1, num2 float64) bool { return num1 < num2 },
+	"<=": func(num1, num2 float64) bool { return num1 <= num2 },
+}
+
 func (this *ExecutorImpl) Execute(rule m.AlertRule, responseQueue chan *AlertResult) {
 	response, err := graphite.GraphiteClient{}.GetSeries(rule)
 
@@ -18,10 +27,10 @@ func (this *ExecutorImpl) Execute(rule m.AlertRule, responseQueue chan *AlertRes
 		responseQueue <- &AlertResult{State: "CRITICAL", Id: rule.Id}
 	}
 
-	responseQueue <- this.executeRules(response, rule)
+	responseQueue <- this.ValidateRule(rule, response)
 }
 
-func (this *ExecutorImpl) executeRules(series m.TimeSeriesSlice, rule m.AlertRule) *AlertResult {
+func (this *ExecutorImpl) ValidateRule(rule m.AlertRule, series m.TimeSeriesSlice) *AlertResult {
 	for _, v := range series {
 		var avg float64
 		var sum float64
@@ -31,19 +40,19 @@ func (this *ExecutorImpl) executeRules(series m.TimeSeriesSlice, rule m.AlertRul
 
 		avg = sum / float64(len(v.Points))
 
-		if float64(rule.CritLevel) < avg {
+		if rule.CritOperator != "" && operators[rule.CritOperator](float64(rule.CritLevel), avg) {
 			return &AlertResult{State: m.AlertStateCritical, Id: rule.Id, ActualValue: avg}
 		}
 
-		if float64(rule.WarnLevel) < avg {
+		if rule.WarnOperator != "" && operators[rule.WarnOperator](float64(rule.WarnLevel), avg) {
 			return &AlertResult{State: m.AlertStateWarn, Id: rule.Id, ActualValue: avg}
 		}
 
-		if float64(rule.CritLevel) < sum {
+		if rule.CritOperator != "" && operators[rule.CritOperator](float64(rule.CritLevel), sum) {
 			return &AlertResult{State: m.AlertStateCritical, Id: rule.Id, ActualValue: sum}
 		}
 
-		if float64(rule.WarnLevel) < sum {
+		if rule.WarnOperator != "" && operators[rule.WarnOperator](float64(rule.WarnLevel), sum) {
 			return &AlertResult{State: m.AlertStateWarn, Id: rule.Id, ActualValue: sum}
 		}
 	}

pkg/services/alerting/executor_test.go

@@ -0,0 +1,58 @@
+package alerting
+
+import (
+	m "github.com/grafana/grafana/pkg/models"
+	. "github.com/smartystreets/goconvey/convey"
+	"testing"
+)
+
+func TestAlertingExecutor(t *testing.T) {
+	Convey("Test alert execution", t, func() {
+		executor := &ExecutorImpl{}
+
+		Convey("Show return ok since avg is above 2", func() {
+			rule := m.AlertRule{CritLevel: 10, CritOperator: "<", Aggregator: "sum"}
+
+			timeseries := []*m.TimeSeries{
+				m.NewTimeSeries("test1", [][2]float64{{2, 0}}),
+			}
+
+			result := executor.ValidateRule(rule, timeseries)
+			So(result.State, ShouldEqual, m.AlertStateOk)
+		})
+
+		Convey("Show return critical since below 2", func() {
+			rule := m.AlertRule{CritLevel: 10, CritOperator: ">", Aggregator: "sum"}
+
+			timeseries := []*m.TimeSeries{
+				m.NewTimeSeries("test1", [][2]float64{{2, 0}}),
+			}
+
+			result := executor.ValidateRule(rule, timeseries)
+			So(result.State, ShouldEqual, m.AlertStateCritical)
+		})
+
+		Convey("Show return critical since sum is above 10", func() {
+			rule := m.AlertRule{CritLevel: 10, CritOperator: "<", Aggregator: "sum"}
+
+			timeseries := []*m.TimeSeries{
+				m.NewTimeSeries("test1", [][2]float64{{9, 0}, {9, 0}}),
+			}
+
+			result := executor.ValidateRule(rule, timeseries)
+			So(result.State, ShouldEqual, m.AlertStateCritical)
+		})
+		/*
+			Convey("Show return ok since avg is below 10", func() {
+				rule := m.AlertRule{CritLevel: 10, CritOperator: "<", Aggregator: "avg"}
+
+				timeseries := []*m.TimeSeries{
+					m.NewTimeSeries("test1", [][2]float64{{9, 0}, {9, 0}}),
+				}
+
+				result := executor.ValidateRule(rule, timeseries)
+				So(result.State, ShouldEqual, m.AlertStateOk)
+			})
+		*/
+	})
+}