Fix race condition in dispatch.go

[grobinson-grafana]

This commit fixes a race condition in dispatch.go that would cause a firing alert to be deleted from the aggregation group when instead it should have been flushed.

The root cause is a race condition that can occur when dispatch.go deletes resolved alerts from the aggregation group following a successful notification. If a firing alert with the same fingerprint is added back to the aggregation group at the same time then the firing alert can be deleted.

The issue was first reported in #3006.

How to observe the race condition

1. git checkout v0.27.0
2. Add the following sleep to dispatch.go. This makes it easier to observe the race condition because the scheduler can resume the test between ag.alerts.Get(fp) and ag.alerts.Delete(fp), during which it should replace the resolved alert with a firing alert.

diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go
index 640b22ab..1eafa054 100644
--- a/dispatch/dispatch.go
+++ b/dispatch/dispatch.go
@@ -521,6 +521,7 @@ func (ag *aggrGroup) flush(notify func(...*types.Alert) bool) {
                        // again since we notified about it.
                        fp := a.Fingerprint()
                        got, err := ag.alerts.Get(fp)
+                       <-time.After(time.Millisecond)
                        if err != nil {
                                // This should never happen.
                                level.Error(ag.logger).Log("msg", "failed to get alert", "err", err, "alert", a.String())

3. Add and run the following test to dispatch_test.go. It should fail:

type bufferedStage struct {
	c chan struct{}
}

func (s *bufferedStage) Exec(ctx context.Context, _ log.Logger, _ ...*types.Alert) (context.Context, []*types.Alert, error) {
	s.c <- struct{}{}
	return ctx, nil, nil
}

func TestDispatcherRaceFiringAlertDeleted(t *testing.T) {
	r := Route{RouteOpts: RouteOpts{GroupWait: 0, GroupInterval: time.Second}}
	s := bufferedStage{c: make(chan struct{}, 1)}
	logger := log.NewLogfmtLogger(os.Stdout)
	marker := types.NewMarker(prometheus.NewRegistry())
	alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, nil, logger, nil)
	require.NoError(t, err)
	defer alerts.Close()
	timeout := func(d time.Duration) time.Duration { return time.Duration(0) }
	dispatcher := NewDispatcher(alerts, &r, &s, marker, timeout, nil, logger, NewDispatcherMetrics(false, prometheus.NewRegistry()))
	go dispatcher.Run()
	defer dispatcher.Stop()

	// Add a new firing alert and wait for it to be flushed.
	a1 := newAlert(model.LabelSet{"foo": "bar"})
	a1.StartsAt = time.Now().Add(-time.Second)
	a1.UpdatedAt = a1.StartsAt
	require.NoError(t, alerts.Put(a1))
	<-s.c

	// Change the firing alert to resolved and wait for it to be flushed too.
	a2 := newAlert(model.LabelSet{"foo": "bar"})
	a2.StartsAt = a1.StartsAt
	a2.UpdatedAt = a1.UpdatedAt
	a2.EndsAt = time.Now()
	require.NoError(t, alerts.Put(a2))
	<-s.c

	// Change the resolved alert back to firing. If the race condition is still
	// present, it will be deleted instead of flushed.
	a3 := newAlert(model.LabelSet{"foo": "bar"})
	a3.StartsAt = time.Now()
	a3.UpdatedAt = time.Now()
	require.NoError(t, alerts.Put(a3))
	select {
	case <-s.c:
	case <-time.After(2 * time.Second):
		t.Fatal()
	}
}

How to observe the fix

1. Check out this branch, add the sleep and re-run the same test. It should pass.

diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go
index d78fbf68..efb0c246 100644
--- a/dispatch/dispatch.go
+++ b/dispatch/dispatch.go
@@ -517,6 +517,7 @@ func (ag *aggrGroup) flush(notify func(...*types.Alert) bool) {

        if notify(alertsSlice...) {
                for _, a := range alertsSlice {
+                       <-time.After(time.Millisecond)
                        // Only delete if the fingerprint has not been inserted again since we notified about it.
                        if err := ag.alerts.DeleteIf(a.Fingerprint(), func(b *types.Alert) bool {
                                return a.Resolved() && a.UpdatedAt == b.UpdatedAt

[zecke]

Should we add another alert and verify it isn't deleted (by accident)?

[zecke]

LGTM. Definitely closes a time of check/time of use kind of problem. Now Set/DeleteIf/gc are protected by the same mutex.

[gotjosh]

LGTM, but please see my comments.

Great job @grobinson-grafana !

[gotjosh]

[nit] How do you feel about making the whole operation atomic? You'd only take the lock once guaranteed that all the alerts are cleared in one go. I feel this makes it easier to reason about it as there's no use case where an alert could get updated once we start checking.

diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go
index d78fbf68..6e44fbc2 100644
--- a/dispatch/dispatch.go
+++ b/dispatch/dispatch.go
@@ -516,14 +516,7 @@ func (ag *aggrGroup) flush(notify func(...*types.Alert) bool) {
 	level.Debug(ag.logger).Log("msg", "flushing", "alerts", fmt.Sprintf("%v", alertsSlice))
 
 	if notify(alertsSlice...) {
-		for _, a := range alertsSlice {
-			// Only delete if the fingerprint has not been inserted again since we notified about it.
-			if err := ag.alerts.DeleteIf(a.Fingerprint(), func(b *types.Alert) bool {
-				return a.Resolved() && a.UpdatedAt == b.UpdatedAt
-			}); err != nil {
-				level.Error(ag.logger).Log("msg", "error on delete alert", "err", err, "alert", a.String())
-			}
-		}
+		ag.alerts.DeleteResolved(alertsSlice)
 	}
 }
 
diff --git a/store/store.go b/store/store.go
index bffa4eb6..4110a971 100644
--- a/store/store.go
+++ b/store/store.go
@@ -154,3 +154,19 @@ func (a *Alerts) Empty() bool {
 
 	return len(a.c) == 0
 }
+
+func (a *Alerts) DeleteResolved(slice types.AlertSlice) {
+	a.Lock()
+	defer a.Unlock()
+
+	for _, as := range slice {
+		fp := as.Fingerprint()
+		alert, ok := a.c[as.Fingerprint()]
+		if !ok {
+		}
+
+		if as.Resolved() && as.UpdatedAt == alert.UpdatedAt {
+			delete(a.c, fp)
+		}
+	}
+}

[grobinson-grafana]

I quite like the condition being a closure, so what if we had something like:

func (a *Alerts) DeleteFingerprints(fp []model.Fingerprint, fn func(int, *types.Alert) bool) error {
	...
}

I did think about adding something like DeleteResolved but with fingerprints:

func (a *Alerts) DeleteResolved(fp []model.Fingerprint) error {
	...
}

but I decided I didn't like the semantics of it because it brought business logic from the dispatcher [1] into the store, and I felt the store should be agnostic of that kind of business logic.

[1] For example:

1. The name DeleteResolved felt confusing as you would expect a store with a method called DeleteResolved to delete all resolved alerts it knows about, not the resolved alerts the caller knows about (it doesn't delete all resolved alerts, just those in the argument).
2. If the caller thinks the alert is firing, but the store thinks the alert is resolved, then DeleteResolved will not delete the alert. This feels weird given the method is called DeleteResolved.
3. A resolved alert is deleted iff UpdatedAt hasn't changed since the last flush (but why does a generic store care about flushes?).

What do you think?

[gotjosh]

That's a fair point.

I'm OK with:

func (a *Alerts) DeleteFingerprints(fp []model.Fingerprint, fn func(int, *types.Alert) bool) error {

As what I care about is the atomicity of the operation.

I think your point about logic is valid but they both share the same types anyways 🤷 so I kind feel that where the logic goes doesn't matter to me as much.

[grobinson-grafana]

In the end I couldn't make DeleteFingerprints as clean as I wanted it to be so instead I've taken a different, although still generic approach, where I've replaced DeleteIf with DeleteIfNotModified. I then use resolvedSlice from dispatch.go as the argument.

[gotjosh]

If we chose to go with my approach or your approach - we'd need to remove the Delete method as it is no longer used.

[gotjosh]

I think the same holds true for Get, which is also used in store but I don't see it used anywhere else (but tests).

[grobinson-grafana]

I don't have a strong opinion here, and I'm happy to remove Get and Delete, but I did just want to raise a counter-argument for this to see if we still think it makes sense.

I understand that Get and Delete will be un-used once we merge this PR, but does that mean we should delete them? These operations seem kind of important for a store interface, even if we don't make use of them today. It's more of a question of semantics rather than pragmatics, but I feel it's important to at least discuss it.

[gotjosh]

I'm a simple man - I see code that's not used I delete it. These operations are simple enough that they can be introduced at any time, and in my experience, "just-in-case methods" tend to cause more confusion than good.

That being said - I'm also not fussed, you can take your pick here.

[grobinson-grafana]

OK! 🙂 Get and Delete have been deleted! 👍

[grobinson-grafana]

Oops! It turns out I cannot remove Get because it's used in provider/mem/mem.go.

[gotjosh]

Yeah, but if you look at the places where that Get is used you'll realise it's only in tests.

[grobinson-grafana]

It's also used on Line 207 of provider/mem/mem.go which isn't a test?

[gotjosh]

Sorry, I could have been clearer. I meant the Get on this file.

diff --git a/provider/mem/mem.go b/provider/mem/mem.go
index cfc3bfc3..b84133e4 100644
--- a/provider/mem/mem.go
+++ b/provider/mem/mem.go
@@ -20,12 +20,10 @@ import (
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/common/model"
-
 	"github.com/prometheus/alertmanager/provider"
 	"github.com/prometheus/alertmanager/store"
 	"github.com/prometheus/alertmanager/types"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 const alertChannelLength = 200
@@ -190,11 +188,6 @@ func (a *Alerts) GetPending() provider.AlertIterator {
 	return provider.NewAlertIterator(ch, done, nil)
 }
 
-// Get returns the alert for a given fingerprint.
-func (a *Alerts) Get(fp model.Fingerprint) (*types.Alert, error) {
-	return a.alerts.Get(fp)
-}
-
 // Put adds the given alert to the set.
 func (a *Alerts) Put(alerts ...*types.Alert) error {
 	for _, alert := range alerts {

[gotjosh]

Discussed this offline, we can't remove it because due to the interface this struct implements.

[grobinson-grafana]

I wanted to add this additional test for completeness.

[gotjosh]

LGTM