feat(alerting): generelize aggregator functions

pkg/services/alerting/executor.go

@@ -11,15 +11,24 @@ type Executor interface {
 
 type ExecutorImpl struct{}
 
-type fn func(float64, float64) bool
+type compareFn func(float64, float64) bool
+type aggregationFn func(*m.TimeSeries) float64
 
-var operators map[string]fn = map[string]fn{
+var operators map[string]compareFn = map[string]compareFn{
 	">":  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 },
 }
 
+var aggregator map[string]aggregationFn = map[string]aggregationFn{
+	"avg":  func(series *m.TimeSeries) float64 { return series.Avg },
+	"sum":  func(series *m.TimeSeries) float64 { return series.Sum },
+	"min":  func(series *m.TimeSeries) float64 { return series.Min },
+	"max":  func(series *m.TimeSeries) float64 { return series.Max },
+	"mean": func(series *m.TimeSeries) float64 { return series.Mean },
+}
+
 func (this *ExecutorImpl) Execute(rule m.AlertRule, responseQueue chan *AlertResult) {
 	response, err := graphite.GraphiteClient{}.GetSeries(rule)
 
@@ -32,28 +41,14 @@ func (this *ExecutorImpl) Execute(rule m.AlertRule, responseQueue chan *AlertRes
 
 func (this *ExecutorImpl) ValidateRule(rule m.AlertRule, series m.TimeSeriesSlice) *AlertResult {
 	for _, v := range series {
-		var avg float64
-		var sum float64
-		for _, dp := range v.Points {
-			sum += dp[0]
-		}
-
-		avg = sum / float64(len(v.Points))
-
-		if rule.CritOperator != "" && operators[rule.CritOperator](float64(rule.CritLevel), avg) {
-			return &AlertResult{State: m.AlertStateCritical, Id: rule.Id, ActualValue: avg}
-		}
-
-		if rule.WarnOperator != "" && operators[rule.WarnOperator](float64(rule.WarnLevel), avg) {
-			return &AlertResult{State: m.AlertStateWarn, Id: rule.Id, ActualValue: avg}
-		}
+		var aggValue = aggregator[rule.Aggregator](v)
 
-		if rule.CritOperator != "" && operators[rule.CritOperator](float64(rule.CritLevel), sum) {
-			return &AlertResult{State: m.AlertStateCritical, Id: rule.Id, ActualValue: sum}
+		if rule.CritOperator != "" && operators[rule.CritOperator](float64(rule.CritLevel), aggValue) {
+			return &AlertResult{State: m.AlertStateCritical, Id: rule.Id, ActualValue: aggValue}
 		}
 
-		if rule.WarnOperator != "" && operators[rule.WarnOperator](float64(rule.WarnLevel), sum) {
-			return &AlertResult{State: m.AlertStateWarn, Id: rule.Id, ActualValue: sum}
+		if rule.WarnOperator != "" && operators[rule.WarnOperator](float64(rule.WarnLevel), aggValue) {
+			return &AlertResult{State: m.AlertStateWarn, Id: rule.Id, ActualValue: aggValue}
 		}
 	}
 

pkg/services/alerting/executor_test.go

@@ -13,46 +13,67 @@ func TestAlertingExecutor(t *testing.T) {
 		Convey("Show return ok since avg is above 2", func() {
 			rule := m.AlertRule{CritLevel: 10, CritOperator: "<", Aggregator: "sum"}
 
-			timeseries := []*m.TimeSeries{
+			timeSeries := []*m.TimeSeries{
 				m.NewTimeSeries("test1", [][2]float64{{2, 0}}),
 			}
 
-			result := executor.ValidateRule(rule, timeseries)
+			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{
+			timeSeries := []*m.TimeSeries{
 				m.NewTimeSeries("test1", [][2]float64{{2, 0}}),
 			}
 
-			result := executor.ValidateRule(rule, timeseries)
+			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{
+			timeSeries := []*m.TimeSeries{
 				m.NewTimeSeries("test1", [][2]float64{{9, 0}, {9, 0}}),
 			}
 
-			result := executor.ValidateRule(rule, timeseries)
+			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)
+		})
+
+		Convey("Show return ok since min is below 10", func() {
+			rule := m.AlertRule{CritLevel: 10, CritOperator: "<", Aggregator: "min"}
+
+			timeSeries := []*m.TimeSeries{
+				m.NewTimeSeries("test1", [][2]float64{{11, 0}, {9, 0}}),
+			}
+
+			result := executor.ValidateRule(rule, timeSeries)
+			So(result.State, ShouldEqual, m.AlertStateOk)
+		})
+
+		Convey("Show return ok since max is above 10", func() {
+			rule := m.AlertRule{CritLevel: 10, CritOperator: "<", Aggregator: "max"}
+
+			timeSeries := []*m.TimeSeries{
+				m.NewTimeSeries("test1", [][2]float64{{1, 0}, {11, 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)
-			})
-		*/
 	})
 }