Merge pull request #13477 from grafana/alert_rule_state_changes_version

use alert state changes counter for alert notification dedupping.

pkg/models/alert.go

@@ -75,7 +75,7 @@ type Alert struct {
 
 	EvalData     *simplejson.Json
 	NewStateDate time.Time
-	StateChanges int
+	StateChanges int64
 
 	Created time.Time
 	Updated time.Time
@@ -156,7 +156,7 @@ type SetAlertStateCommand struct {
 	Error    string
 	EvalData *simplejson.Json
 
-	Timestamp time.Time
+	Result Alert
 }
 
 //Queries

pkg/models/alert_notifications.go

@@ -97,7 +97,8 @@ type AlertNotificationState struct {
 }
 
 type SetAlertNotificationStateToPendingCommand struct {
-	State *AlertNotificationState
+	AlertRuleStateUpdatedVersion int64
+	State                        *AlertNotificationState
 }
 
 type SetAlertNotificationStateToCompleteCommand struct {

pkg/services/alerting/notifier.go

@@ -94,7 +94,8 @@ func (n *notificationService) sendAndMarkAsComplete(evalContext *EvalContext, no
 func (n *notificationService) sendNotification(evalContext *EvalContext, notifierState *NotifierState) error {
 	if !evalContext.IsTestRun {
 		setPendingCmd := &m.SetAlertNotificationStateToPendingCommand{
-			State: notifierState.state,
+			State:                        notifierState.state,
+			AlertRuleStateUpdatedVersion: evalContext.Rule.StateChanges,
 		}
 
 		err := bus.DispatchCtx(evalContext.Ctx, setPendingCmd)
@@ -172,7 +173,7 @@ func (n *notificationService) getNeededNotifiers(orgId int64, notificationIds []
 	for _, notification := range query.Result {
 		not, err := n.createNotifierFor(notification)
 		if err != nil {
-			n.log.Error("Could not create notifier", "notifier", notification.Id)
+			n.log.Error("Could not create notifier", "notifier", notification.Id, "error", err)
 			continue
 		}
 

pkg/services/alerting/result_handler.go

@@ -67,6 +67,11 @@ func (handler *DefaultResultHandler) Handle(evalContext *EvalContext) error {
 			}
 
 			handler.log.Error("Failed to save state", "error", err)
+		} else {
+
+			// StateChanges is used for de dupping alert notifications
+			// when two servers are raising.
+			evalContext.Rule.StateChanges = cmd.Result.StateChanges
 		}
 
 		// save annotation

pkg/services/alerting/rule.go

@@ -23,6 +23,8 @@ type Rule struct {
 	State               m.AlertStateType
 	Conditions          []Condition
 	Notifications       []int64
+
+	StateChanges int64
 }
 
 type ValidationError struct {

pkg/services/sqlstore/alert.go

@@ -279,6 +279,8 @@ func SetAlertState(cmd *m.SetAlertStateCommand) error {
 		}
 
 		sess.ID(alert.Id).Update(&alert)
+
+		cmd.Result = alert
 		return nil
 	})
 }

pkg/services/sqlstore/alert_notification.go

@@ -277,19 +277,26 @@ func SetAlertNotificationStateToPendingCommand(ctx context.Context, cmd *m.SetAl
 		sql := `UPDATE alert_notification_state SET
 			state = ?,
 			version = ?,
-			updated_at = ?
+			updated_at = ?,
+			alert_rule_state_updated_version = ?
 		WHERE
 			id = ? AND
-			version = ?`
+			(version = ? OR alert_rule_state_updated_version < ?)`
 
-		res, err := sess.Exec(sql, cmd.State.State, cmd.State.Version, timeNow().Unix(), cmd.State.Id, currentVersion)
+		res, err := sess.Exec(sql,
+			cmd.State.State,
+			cmd.State.Version,
+			timeNow().Unix(),
+			cmd.AlertRuleStateUpdatedVersion,
+			cmd.State.Id,
+			currentVersion,
+			cmd.AlertRuleStateUpdatedVersion)
 
 		if err != nil {
 			return err
 		}
 
 		affected, _ := res.RowsAffected()
-
 		if affected == 0 {
 			return m.ErrAlertNotificationStateVersionConflict
 		}

pkg/services/sqlstore/alert_notification_test.go

@@ -100,6 +100,29 @@ func TestAlertNotificationSQLAccess(t *testing.T) {
 					err := SetAlertNotificationStateToPendingCommand(context.Background(), &cmd)
 					So(err, ShouldEqual, models.ErrAlertNotificationStateVersionConflict)
 				})
+
+				Convey("Updating existing state to pending with incorrect version since alert rule state update version is higher", func() {
+					s := *query.Result
+					cmd := models.SetAlertNotificationStateToPendingCommand{
+						State:                        &s,
+						AlertRuleStateUpdatedVersion: 1000,
+					}
+					err := SetAlertNotificationStateToPendingCommand(context.Background(), &cmd)
+					So(err, ShouldBeNil)
+
+					So(cmd.State.Version, ShouldEqual, 1)
+					So(cmd.State.State, ShouldEqual, models.AlertNotificationStatePending)
+				})
+
+				Convey("different version and same alert state change version should return error", func() {
+					s := *query.Result
+					s.Version = 1000
+					cmd := models.SetAlertNotificationStateToPendingCommand{
+						State: &s,
+					}
+					err := SetAlertNotificationStateToPendingCommand(context.Background(), &cmd)
+					So(err, ShouldNotBeNil)
+				})
 			})
 
 			Reset(func() {

pkg/services/sqlstore/migrations/alert_mig.go

@@ -130,4 +130,8 @@ func addAlertMigrations(mg *Migrator) {
 	mg.AddMigration("create alert_notification_state table v1", NewAddTableMigration(alert_notification_state))
 	mg.AddMigration("add index alert_notification_state org_id & alert_id & notifier_id",
 		NewAddIndexMigration(alert_notification_state, alert_notification_state.Indices[0]))
+
+	mg.AddMigration("Add alert_rule_state_updated_version to alert_notification_state", NewAddColumnMigration(alert_notification_state, &Column{
+		Name: "alert_rule_state_updated_version", Type: DB_BigInt, Nullable: true,
+	}))
 }