feat(notify): make mute stage honor send_resolved

Make mute stage honor `send_resolved` in receivers.
This fixes cases where an alert was already sent to a receiver
before being silenced. Sending resolved notifications for a now silenced
alert helps update the receiver's status.
For example an opened PagerDuty incident will be closed.

Signed-off-by: Siavash Safi <[email protected]>

Author: siavashs

notify/notify.go

@@ -541,17 +541,37 @@ const (
 
 // MuteStage filters alerts through a Muter.
 type MuteStage struct {
-	muter   types.Muter
-	metrics *Metrics
+	muter           types.Muter
+	metrics         *Metrics
+	notificationLog NotificationLog
+	receivers       map[string][]Integration
 }
 
 // NewMuteStage return a new MuteStage.
 func NewMuteStage(m types.Muter, metrics *Metrics) *MuteStage {
 	return &MuteStage{muter: m, metrics: metrics}
 }
 
+// NewMuteStageWithSendResolved returns a new MuteStage that honors send_resolved
+// for silenced alerts. This should be used for silence.Silencer to allow resolved
+// notifications for previously notified alerts.
+func NewMuteStageWithSendResolved(m types.Muter, notificationLog NotificationLog, receivers map[string][]Integration, metrics *Metrics) *MuteStage {
+	return &MuteStage{
+		muter:           m,
+		metrics:         metrics,
+		notificationLog: notificationLog,
+		receivers:       receivers,
+	}
+}
+
 // Exec implements the Stage interface.
 func (n *MuteStage) Exec(ctx context.Context, logger *slog.Logger, alerts ...*types.Alert) (context.Context, []*types.Alert, error) {
+	// If this is a silencer with send_resolved support, use the enhanced logic
+	if _, isSilencer := n.muter.(*silence.Silencer); isSilencer && n.notificationLog != nil && n.receivers != nil {
+		return n.execWithSendResolved(ctx, logger, alerts...)
+	}
+
+	// Standard muting logic for inhibitors or silencers without send_resolved support
 	var (
 		filtered []*types.Alert
 		muted    []*types.Alert
@@ -583,6 +603,114 @@ func (n *MuteStage) Exec(ctx context.Context, logger *slog.Logger, alerts ...*ty
 	return ctx, filtered, nil
 }
 
+// execWithSendResolved implements enhanced silence filtering that honors send_resolved config.
+// It allows resolved notifications for silenced alerts if the alert was previously notified
+// and the receiver has send_resolved enabled.
+func (n *MuteStage) execWithSendResolved(ctx context.Context, logger *slog.Logger, alerts ...*types.Alert) (context.Context, []*types.Alert, error) {
+	var (
+		filtered []*types.Alert
+		muted    []*types.Alert
+	)
+
+	// Get the receiver name from context
+	receiverName, ok := ReceiverName(ctx)
+	if !ok {
+		return ctx, nil, errors.New("receiver name missing")
+	}
+
+	// Get the group key from context
+	groupKey, ok := GroupKey(ctx)
+	if !ok {
+		return ctx, nil, errors.New("group key missing")
+	}
+
+	// Check if any integration for this receiver has send_resolved enabled
+	receiverIntegrations, receiverExists := n.receivers[receiverName]
+	if !receiverExists {
+		// Receiver not found, fall back to standard behavior
+		for _, a := range alerts {
+			if n.muter.Mutes(a.Labels) {
+				muted = append(muted, a)
+			} else {
+				filtered = append(filtered, a)
+			}
+		}
+		if len(muted) > 0 {
+			n.metrics.numNotificationSuppressedTotal.WithLabelValues(SuppressedReasonSilence).Add(float64(len(muted)))
+			logger.Debug("Notifications will not be sent for silenced alerts", "alerts", fmt.Sprintf("%v", muted), "reason", SuppressedReasonSilence)
+		}
+		return ctx, filtered, nil
+	}
+
+	// Check if any integration has send_resolved enabled
+	hasSendResolved := false
+	for _, integration := range receiverIntegrations {
+		if integration.SendResolved() {
+			hasSendResolved = true
+			break
+		}
+	}
+
+	// Process each alert
+	for _, a := range alerts {
+		if !n.muter.Mutes(a.Labels) {
+			// Alert is not silenced, let it through
+			filtered = append(filtered, a)
+			continue
+		}
+
+		// Alert is silenced
+		if !a.Resolved() || !hasSendResolved {
+			// Alert is firing or receiver doesn't have send_resolved, filter it out
+			muted = append(muted, a)
+			continue
+		}
+
+		// Alert is resolved and receiver has send_resolved enabled
+		// Check if this alert was previously notified
+		wasNotified := false
+		for _, integration := range receiverIntegrations {
+			if !integration.SendResolved() {
+				continue
+			}
+
+			recv := &nflogpb.Receiver{
+				GroupName:   receiverName,
+				Integration: integration.Name(),
+				Idx:         uint32(integration.Index()),
+			}
+
+			entries, err := n.notificationLog.Query(nflog.QGroupKey(groupKey), nflog.QReceiver(recv))
+			if err != nil && !errors.Is(err, nflog.ErrNotFound) {
+				logger.Warn("Failed to query notification log for silenced resolved alert", "alert", a.Name(), "err", err)
+				continue
+			}
+
+			if len(entries) > 0 && len(entries[0].FiringAlerts) > 0 {
+				// This alert was previously notified as firing for this integration
+				wasNotified = true
+				break
+			}
+		}
+
+		if wasNotified {
+			// Alert was previously notified and is now resolved, let it through for resolved notification
+			filtered = append(filtered, a)
+			logger.Debug("Allowing resolved notification for silenced alert", "alert", a.Name())
+		} else {
+			// Alert was not previously notified, filter it out
+			muted = append(muted, a)
+		}
+	}
+
+	if len(muted) > 0 {
+		n.metrics.numNotificationSuppressedTotal.WithLabelValues(SuppressedReasonSilence).Add(float64(len(muted)))
+		logger.Debug("Notifications will not be sent for silenced alerts", "alerts", fmt.Sprintf("%v", muted), "reason", SuppressedReasonSilence)
+	}
+
+	return ctx, filtered, nil
+}
+
 // WaitStage waits for a certain amount of time before continuing or until the
 // context is done.
 type WaitStage struct {

notify/notify_test.go

@@ -823,6 +823,204 @@ func TestMuteStageWithSilences(t *testing.T) {
 	}
 }
 
+func TestMuteStageWithSendResolved(t *testing.T) {
+	// Create silences
+	silences, err := silence.New(silence.Options{Retention: time.Hour})
+	require.NoError(t, err)
+
+	sil := &silencepb.Silence{
+		EndsAt:   utcNow().Add(time.Hour),
+		Matchers: []*silencepb.Matcher{{Name: "mute", Pattern: "me"}},
+	}
+	require.NoError(t, silences.Set(sil))
+
+	reg := prometheus.NewRegistry()
+	marker := types.NewMarker(reg)
+	silencer := silence.NewSilencer(silences, marker, promslog.NewNopLogger())
+
+	// Create notification log
+	nflog, err := nflog.New(nflog.Options{
+		Retention: time.Hour,
+		Logger:    promslog.NewNopLogger(),
+		Metrics:   reg,
+	})
+	require.NoError(t, err)
+
+	// Create receivers with different send_resolved settings
+	receiversWithSendResolved := map[string][]Integration{
+		"with_resolved": {
+			NewIntegration(&testNotifier{}, sendResolved(true), "webhook", 0, "with_resolved"),
+		},
+		"without_resolved": {
+			NewIntegration(&testNotifier{}, sendResolved(false), "webhook", 0, "without_resolved"),
+		},
+		"mixed": {
+			NewIntegration(&testNotifier{}, sendResolved(true), "webhook", 0, "mixed"),
+			NewIntegration(&testNotifier{}, sendResolved(false), "email", 1, "mixed"),
+		},
+	}
+
+	metrics := NewMetrics(reg, featurecontrol.NoopFlags{})
+	stage := NewMuteStageWithSendResolved(silencer, nflog, receiversWithSendResolved, metrics)
+
+	// Test 1: Silenced firing alerts are filtered out
+	t.Run("silenced firing alerts are filtered", func(t *testing.T) {
+		firingAlert := &types.Alert{
+			Alert: model.Alert{
+				Labels:   model.LabelSet{"mute": "me", "alertname": "test"},
+				EndsAt:   time.Time{}, // Firing alert has zero EndsAt
+				StartsAt: time.Now().Add(-time.Hour),
+			},
+		}
+
+		ctx := context.Background()
+		ctx = WithReceiverName(ctx, "with_resolved")
+		ctx = WithGroupKey(ctx, "test-group")
+
+		_, alerts, err := stage.Exec(ctx, promslog.NewNopLogger(), firingAlert)
+		require.NoError(t, err)
+		require.Empty(t, alerts, "firing silenced alerts should be filtered")
+	})
+
+	// Test 2: Silenced resolved alerts without prior notification are filtered
+	t.Run("silenced resolved alerts without prior notification are filtered", func(t *testing.T) {
+		resolvedAlert := &types.Alert{
+			Alert: model.Alert{
+				Labels:   model.LabelSet{"mute": "me", "alertname": "test2"},
+				StartsAt: time.Now().Add(-2 * time.Hour),
+				EndsAt:   time.Now().Add(-time.Hour),
+			},
+		}
+
+		ctx := context.Background()
+		ctx = WithReceiverName(ctx, "with_resolved")
+		ctx = WithGroupKey(ctx, "test-group-2")
+
+		_, alerts, err := stage.Exec(ctx, promslog.NewNopLogger(), resolvedAlert)
+		require.NoError(t, err)
+		require.Empty(t, alerts, "resolved silenced alerts without prior notification should be filtered")
+	})
+
+	// Test 3: Silenced resolved alerts with prior notification and send_resolved=true are allowed
+	t.Run("silenced resolved alerts with prior notification and send_resolved are allowed", func(t *testing.T) {
+		resolvedAlert := &types.Alert{
+			Alert: model.Alert{
+				Labels:   model.LabelSet{"mute": "me", "alertname": "test3"},
+				StartsAt: time.Now().Add(-2 * time.Hour),
+				EndsAt:   time.Now().Add(-time.Hour),
+			},
+		}
+
+		groupKey := "test-group-3"
+		receiverName := "with_resolved"
+
+		// Log a previous notification for this alert
+		recv := &nflogpb.Receiver{
+			GroupName:   receiverName,
+			Integration: "webhook",
+			Idx:         0,
+		}
+		err := nflog.Log(recv, groupKey, []uint64{1234}, []uint64{}, time.Hour)
+		require.NoError(t, err)
+
+		ctx := context.Background()
+		ctx = WithReceiverName(ctx, receiverName)
+		ctx = WithGroupKey(ctx, groupKey)
+
+		_, alerts, err := stage.Exec(ctx, promslog.NewNopLogger(), resolvedAlert)
+		require.NoError(t, err)
+		require.Len(t, alerts, 1, "resolved silenced alerts with prior notification should be allowed when send_resolved=true")
+	})
+
+	// Test 4: Silenced resolved alerts with send_resolved=false are filtered even with prior notification
+	t.Run("silenced resolved alerts without send_resolved are filtered", func(t *testing.T) {
+		resolvedAlert := &types.Alert{
+			Alert: model.Alert{
+				Labels:   model.LabelSet{"mute": "me", "alertname": "test4"},
+				StartsAt: time.Now().Add(-2 * time.Hour),
+				EndsAt:   time.Now().Add(-time.Hour),
+			},
+		}
+
+		groupKey := "test-group-4"
+		receiverName := "without_resolved"
+
+		// Log a previous notification for this alert
+		recv := &nflogpb.Receiver{
+			GroupName:   receiverName,
+			Integration: "webhook",
+			Idx:         0,
+		}
+		err := nflog.Log(recv, groupKey, []uint64{5678}, []uint64{}, time.Hour)
+		require.NoError(t, err)
+
+		ctx := context.Background()
+		ctx = WithReceiverName(ctx, receiverName)
+		ctx = WithGroupKey(ctx, groupKey)
+
+		_, alerts, err := stage.Exec(ctx, promslog.NewNopLogger(), resolvedAlert)
+		require.NoError(t, err)
+		require.Empty(t, alerts, "resolved silenced alerts should be filtered when send_resolved=false")
+	})
+
+	// Test 5: Non-silenced alerts pass through
+	t.Run("non-silenced alerts pass through", func(t *testing.T) {
+		firingAlert := &types.Alert{
+			Alert: model.Alert{
+				Labels:   model.LabelSet{"not": "muted", "alertname": "test5"},
+				EndsAt:   time.Time{},
+				StartsAt: time.Now().Add(-time.Hour),
+			},
+		}
+		resolvedAlert := &types.Alert{
+			Alert: model.Alert{
+				Labels:   model.LabelSet{"not": "muted", "alertname": "test6"},
+				StartsAt: time.Now().Add(-2 * time.Hour),
+				EndsAt:   time.Now().Add(-time.Hour),
+			},
+		}
+
+		ctx := context.Background()
+		ctx = WithReceiverName(ctx, "with_resolved")
+		ctx = WithGroupKey(ctx, "test-group-5")
+
+		_, alerts, err := stage.Exec(ctx, promslog.NewNopLogger(), firingAlert, resolvedAlert)
+		require.NoError(t, err)
+		require.Len(t, alerts, 2, "non-silenced alerts should pass through")
+	})
+
+	// Test 6: Mixed receiver with at least one send_resolved=true integration
+	t.Run("mixed receiver with send_resolved allows resolved silenced alerts", func(t *testing.T) {
+		resolvedAlert := &types.Alert{
+			Alert: model.Alert{
+				Labels:   model.LabelSet{"mute": "me", "alertname": "test7"},
+				StartsAt: time.Now().Add(-2 * time.Hour),
+				EndsAt:   time.Now().Add(-time.Hour),
+			},
+		}
+
+		groupKey := "test-group-7"
+		receiverName := "mixed"
+
+		// Log a previous notification for the webhook integration (send_resolved=true)
+		recv := &nflogpb.Receiver{
+			GroupName:   receiverName,
+			Integration: "webhook",
+			Idx:         0,
+		}
+		err := nflog.Log(recv, groupKey, []uint64{9999}, []uint64{}, time.Hour)
+		require.NoError(t, err)
+
+		ctx := context.Background()
+		ctx = WithReceiverName(ctx, receiverName)
+		ctx = WithGroupKey(ctx, groupKey)
+
+		_, alerts, err := stage.Exec(ctx, promslog.NewNopLogger(), resolvedAlert)
+		require.NoError(t, err)
+		require.Len(t, alerts, 1, "resolved silenced alerts should be allowed when at least one integration has send_resolved=true")
+	})
+}
+
 func TestTimeMuteStage(t *testing.T) {
 	sydney, err := time.LoadLocation("Australia/Sydney")
 	if err != nil {
@@ -1080,3 +1278,9 @@ func BenchmarkHashAlert(b *testing.B) {
 		hashAlert(alert)
 	}
 }
+
+type testNotifier struct{}
+
+func (n *testNotifier) Notify(ctx context.Context, alerts ...*types.Alert) (bool, error) {
+	return true, nil
+}