handle pending and completed state for alert notifications

pkg/models/alert_notifications.go

@@ -100,9 +100,7 @@ type SetAlertNotificationStateToPendingCommand struct {
 }
 
 type SetAlertNotificationStateToCompleteCommand struct {
-	Id      int64
-	Version int64
-	SentAt  int64
+	State *AlertNotificationState
 }
 
 type GetNotificationStateQuery struct {

pkg/services/alerting/notifier.go

@@ -8,6 +8,7 @@ import (
 	"github.com/grafana/grafana/pkg/bus"
 	"github.com/grafana/grafana/pkg/components/imguploader"
 	"github.com/grafana/grafana/pkg/log"
+	"github.com/grafana/grafana/pkg/metrics"
 	"github.com/grafana/grafana/pkg/services/rendering"
 	"github.com/grafana/grafana/pkg/setting"
 
@@ -58,20 +59,32 @@ func (n *notificationService) SendIfNeeded(context *EvalContext) error {
 }
 
 func (n *notificationService) sendAndMarkAsComplete(evalContext *EvalContext, notifierState *NotifierState) error {
-	err := notifierState.notifier.Notify(evalContext)
+	not := notifierState.notifier
+	n.log.Debug("Sending notification", "type", not.GetType(), "id", not.GetNotifierId(), "isDefault", not.GetIsDefault())
+	metrics.M_Alerting_Notification_Sent.WithLabelValues(not.GetType()).Inc()
 
-	cmd := &m.SetAlertNotificationStateToCompleteCommand{
-		Id:      notifierState.state.Id,
-		Version: notifierState.state.Version,
-		SentAt:  time.Now().Unix(),
+	err := not.Notify(evalContext)
+
+	if err != nil {
+		n.log.Error("failed to send notification", "id", not.GetNotifierId())
+	} else {
+		notifierState.state.SentAt = time.Now().Unix()
 	}
 
-	err = bus.DispatchCtx(evalContext.Ctx, cmd)
-	if err == m.ErrAlertNotificationStateVersionConflict {
+	if evalContext.IsTestRun {
 		return nil
 	}
 
-	if err != nil {
+	cmd := &m.SetAlertNotificationStateToCompleteCommand{
+		State: notifierState.state,
+	}
+
+	if err = bus.DispatchCtx(evalContext.Ctx, cmd); err != nil {
+		if err == m.ErrAlertNotificationStateVersionConflict {
+			n.log.Error("notification state out of sync", "id", not.GetNotifierId())
+			return nil
+		}
+
 		return err
 	}
 
@@ -79,19 +92,19 @@ func (n *notificationService) sendAndMarkAsComplete(evalContext *EvalContext, no
 }
 
 func (n *notificationService) sendNotification(evalContext *EvalContext, notifierState *NotifierState) error {
-	n.log.Debug("trying to send notification", "id", notifierState.notifier.GetNotifierId())
-
-	setPendingCmd := &m.SetAlertNotificationStateToPendingCommand{
-		State: notifierState.state,
-	}
+	if !evalContext.IsTestRun {
+		setPendingCmd := &m.SetAlertNotificationStateToPendingCommand{
+			State: notifierState.state,
+		}
 
-	err := bus.DispatchCtx(evalContext.Ctx, setPendingCmd)
-	if err == m.ErrAlertNotificationStateVersionConflict {
-		return nil
-	}
+		err := bus.DispatchCtx(evalContext.Ctx, setPendingCmd)
+		if err == m.ErrAlertNotificationStateVersionConflict {
+			return nil
+		}
 
-	if err != nil {
-		return err
+		if err != nil {
+			return err
+		}
 	}
 
 	return n.sendAndMarkAsComplete(evalContext, notifierState)

pkg/services/alerting/notifiers/base_test.go

@@ -121,32 +121,6 @@ func TestShouldSendAlertNotification(t *testing.T) {
 	}
 }
 
-func TestShouldNotifyWhenNoJournalingIsFound(t *testing.T) {
-	Convey("base notifier", t, func() {
-		//bus.ClearBusHandlers()
-		//
-		//notifier := NewNotifierBase(&m.AlertNotification{
-		//	Id:       1,
-		//	Name:     "name",
-		//	Type:     "email",
-		//	Settings: simplejson.New(),
-		//})
-		//evalContext := alerting.NewEvalContext(context.TODO(), &alerting.Rule{})
-		//
-		//Convey("should not notify query returns error", func() {
-		//	bus.AddHandlerCtx("", func(ctx context.Context, q *m.GetNotificationStateQuery) error {
-		//		return errors.New("some kind of error unknown error")
-		//	})
-		//
-		//	if notifier.ShouldNotify(context.Background(), evalContext) {
-		//		t.Errorf("should not send notifications when query returns error")
-		//	}
-		//})
-
-		t.Error("might not need this anymore, at least not like this, control flow has changedd")
-	})
-}
-
 func TestBaseNotifier(t *testing.T) {
 	Convey("default constructor for notifiers", t, func() {
 		bJson := simplejson.New()

pkg/services/sqlstore/alert_notification.go

@@ -255,20 +255,26 @@ func InsertAlertNotificationState(ctx context.Context, cmd *m.InsertAlertNotific
 
 func SetAlertNotificationStateToCompleteCommand(ctx context.Context, cmd *m.SetAlertNotificationStateToCompleteCommand) error {
 	return withDbSession(ctx, func(sess *DBSession) error {
+		version := cmd.State.Version
+		var current m.AlertNotificationState
+		sess.ID(cmd.State.Id).Get(&current)
+
+		cmd.State.State = m.AlertNotificationStateCompleted
+		cmd.State.Version++
+
 		sql := `UPDATE alert_notification_state SET
 			state = ?,
 			version = ?
 		WHERE
 			id = ?`
 
-		res, err := sess.Exec(sql, m.AlertNotificationStateCompleted, cmd.Id, cmd.Version+1)
+		_, err := sess.Exec(sql, cmd.State.State, cmd.State.Version, cmd.State.Id)
+
 		if err != nil {
 			return err
 		}
 
-		affected, _ := res.RowsAffected()
-
-		if affected == 0 {
+		if current.Version != version {
 			return m.ErrAlertNotificationStateVersionConflict
 		}
 
@@ -278,6 +284,10 @@ func SetAlertNotificationStateToCompleteCommand(ctx context.Context, cmd *m.SetA
 
 func SetAlertNotificationStateToPendingCommand(ctx context.Context, cmd *m.SetAlertNotificationStateToPendingCommand) error {
 	return withDbSession(ctx, func(sess *DBSession) error {
+		currentVersion := cmd.State.Version
+		cmd.State.State = m.AlertNotificationStatePending
+		cmd.State.Version++
+
 		sql := `UPDATE alert_notification_state SET
 			state = ?,
 			version = ?
@@ -285,7 +295,8 @@ func SetAlertNotificationStateToPendingCommand(ctx context.Context, cmd *m.SetAl
 			id = ? AND
 			version = ?`
 
-		res, err := sess.Exec(sql, m.AlertNotificationStatePending, cmd.State.Version+1, cmd.State.Id, cmd.State.Version)
+		res, err := sess.Exec(sql, cmd.State.State, cmd.State.Version, cmd.State.Id, currentVersion)
+
 		if err != nil {
 			return err
 		}

pkg/services/sqlstore/alert_notification_test.go

@@ -25,6 +25,7 @@ func TestAlertNotificationSQLAccess(t *testing.T) {
 				So(err, ShouldBeNil)
 				So(query.Result, ShouldNotBeNil)
 				So(query.Result.State, ShouldEqual, "unknown")
+				So(query.Result.Version, ShouldEqual, 0)
 
 				Convey("Get existing state should not create a new state", func() {
 					query2 := &models.GetNotificationStateQuery{AlertId: alertID, OrgId: orgID, NotifierId: notifierID}
@@ -33,50 +34,67 @@ func TestAlertNotificationSQLAccess(t *testing.T) {
 					So(query2.Result, ShouldNotBeNil)
 					So(query2.Result.Id, ShouldEqual, query.Result.Id)
 				})
+
+				Convey("Update existing state to pending with correct version should update database", func() {
+					s := *query.Result
+					cmd := models.SetAlertNotificationStateToPendingCommand{
+						State: &s,
+					}
+					err := SetAlertNotificationStateToPendingCommand(context.Background(), &cmd)
+					So(err, ShouldBeNil)
+					So(cmd.State.Version, ShouldEqual, 1)
+					So(cmd.State.State, ShouldEqual, models.AlertNotificationStatePending)
+
+					query2 := &models.GetNotificationStateQuery{AlertId: alertID, OrgId: orgID, NotifierId: notifierID}
+					err = GetAlertNotificationState(context.Background(), query2)
+					So(err, ShouldBeNil)
+					So(query2.Result.Version, ShouldEqual, 1)
+					So(query2.Result.State, ShouldEqual, models.AlertNotificationStatePending)
+
+					Convey("Update existing state to completed should update database", func() {
+						s := *cmd.State
+						cmd := models.SetAlertNotificationStateToCompleteCommand{
+							State: &s,
+						}
+						err := SetAlertNotificationStateToCompleteCommand(context.Background(), &cmd)
+						So(err, ShouldBeNil)
+
+						query3 := &models.GetNotificationStateQuery{AlertId: alertID, OrgId: orgID, NotifierId: notifierID}
+						err = GetAlertNotificationState(context.Background(), query3)
+						So(err, ShouldBeNil)
+						So(query3.Result.Version, ShouldEqual, 2)
+						So(query3.Result.State, ShouldEqual, models.AlertNotificationStateCompleted)
+					})
+
+					Convey("Update existing state to completed should update database, but return version mismatch", func() {
+						cmd.State.Version = 1000
+						s := *cmd.State
+						cmd := models.SetAlertNotificationStateToCompleteCommand{
+							State: &s,
+						}
+						err := SetAlertNotificationStateToCompleteCommand(context.Background(), &cmd)
+						So(err, ShouldEqual, models.ErrAlertNotificationStateVersionConflict)
+
+						query3 := &models.GetNotificationStateQuery{AlertId: alertID, OrgId: orgID, NotifierId: notifierID}
+						err = GetAlertNotificationState(context.Background(), query3)
+						So(err, ShouldBeNil)
+						So(query3.Result.Version, ShouldEqual, 1001)
+						So(query3.Result.State, ShouldEqual, models.AlertNotificationStateCompleted)
+					})
+				})
+
+				Convey("Update existing state to pending with incorrect version should return version mismatch error", func() {
+					s := *query.Result
+					s.Version = 1000
+					cmd := models.SetAlertNotificationStateToPendingCommand{
+						State: &s,
+					}
+					err := SetAlertNotificationStateToPendingCommand(context.Background(), &cmd)
+					So(err, ShouldEqual, models.ErrAlertNotificationStateVersionConflict)
+				})
 			})
 		})
 
-		//Convey("Can insert new state for alert notifier", func() {
-		//	createCmd := &models.InsertAlertNotificationCommand{
-		//		AlertId:    alertId,
-		//		NotifierId: notifierId,
-		//		OrgId:      orgId,
-		//		SentAt:     1,
-		//		State:      models.AlertNotificationStateCompleted,
-		//	}
-		//
-		//	err := InsertAlertNotificationState(context.Background(), createCmd)
-		//	So(err, ShouldBeNil)
-		//
-		//	err = InsertAlertNotificationState(context.Background(), createCmd)
-		//	So(err, ShouldEqual, models.ErrAlertNotificationStateAlreadyExist)
-		//
-		//	Convey("should be able to update alert notifier state", func() {
-		//		updateCmd := &models.SetAlertNotificationStateToPendingCommand{
-		//			State: models.AlertNotificationState{
-		//				Id:         1,
-		//				SentAt:     1,
-		//				Version:    0,
-		//			}
-		//		}
-		//
-		//		err := SetAlertNotificationStateToPendingCommand(context.Background(), updateCmd)
-		//		So(err, ShouldBeNil)
-		//
-		//		Convey("should not be able to set pending on old version", func() {
-		//			err = SetAlertNotificationStateToPendingCommand(context.Background(), updateCmd)
-		//			So(err, ShouldEqual, models.ErrAlertNotificationStateVersionConflict)
-		//		})
-		//
-		//		Convey("should be able to set state to completed", func() {
-		//			cmd := &models.SetAlertNotificationStateToCompleteCommand{Id: 1}
-		//			err = SetAlertNotificationStateToCompleteCommand(context.Background(), cmd)
-		//			So(err, ShouldBeNil)
-		//		})
-		//	})
-		//	})
-		//})
-
 		Convey("Alert notifications should be empty", func() {
 			cmd := &models.GetAlertNotificationsQuery{
 				OrgId: 2,