nflog: add muted_alerts to the notification log entry

nflog/nflog.go

@@ -461,7 +461,7 @@ func stateKey(k string, r *pb.Receiver) string {
 	return fmt.Sprintf("%s:%s", k, receiverKey(r))
 }
 
-func (l *Log) Log(r *pb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, store *Store, expiry time.Duration) error {
+func (l *Log) Log(r *pb.Receiver, gkey string, firingAlerts, resolvedAlerts, mutedAlerts []uint64, store *Store, expiry time.Duration) error {
 	// Write all st with the same timestamp.
 	now := l.now()
 	key := stateKey(gkey, r)
@@ -494,6 +494,7 @@ func (l *Log) Log(r *pb.Receiver, gkey string, firingAlerts, resolvedAlerts []ui
 			Timestamp:      timestamppb.New(now),
 			FiringAlerts:   firingAlerts,
 			ResolvedAlerts: resolvedAlerts,
+			MutedAlerts:    mutedAlerts,
 			ReceiverData:   receiverData,
 		},
 		ExpiresAt: timestamppb.New(expiresAt),

nflog/nflog_test.go

@@ -367,15 +367,95 @@ func TestQuery(t *testing.T) {
 	// existing entry
 	firingAlerts := []uint64{1, 2, 3}
 	resolvedAlerts := []uint64{4, 5}
+	mutedAlerts := []uint64{6, 7}
 
-	err = nl.Log(recv, "key", firingAlerts, resolvedAlerts, nil, 0)
+	err = nl.Log(recv, "key", firingAlerts, resolvedAlerts, mutedAlerts, nil, 0)
 	require.NoError(t, err, "logging notification failed")
 
 	entries, err := nl.Query(QGroupKey("key"), QReceiver(recv))
 	require.NoError(t, err, "querying nflog failed")
 	entry := entries[0]
 	require.Equal(t, firingAlerts, entry.FiringAlerts)
 	require.Equal(t, resolvedAlerts, entry.ResolvedAlerts)
+	require.Equal(t, mutedAlerts, entry.MutedAlerts)
+}
+
+// TestLogMutedAlerts checks that muted alerts survive a round trip through the
+// wire format, and that an entry logged without them decodes with the field
+// unset, as it does for an entry written by a peer that does not know the
+// field.
+func TestLogMutedAlerts(t *testing.T) {
+	now := time.Now().UTC()
+
+	cases := []struct {
+		name  string
+		muted []uint64
+	}{
+		{name: "with muted alerts", muted: []uint64{6, 7}},
+		{name: "without muted alerts", muted: nil},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			in := state{
+				"key:abc/test/1": &pb.MeshEntry{
+					Entry: &pb.Entry{
+						GroupKey:       []byte("key"),
+						Receiver:       &pb.Receiver{GroupName: "abc", Integration: "test", Idx: 1},
+						Timestamp:      timestamppb.New(now),
+						FiringAlerts:   []uint64{1, 2, 3},
+						ResolvedAlerts: []uint64{4, 5},
+						MutedAlerts:    c.muted,
+					},
+					ExpiresAt: timestamppb.New(now.Add(time.Minute)),
+				},
+			}
+
+			msg, err := in.MarshalBinary()
+			require.NoError(t, err)
+
+			out, err := decodeState(bytes.NewReader(msg))
+			require.NoError(t, err, "decoding message failed")
+
+			for id, expected := range in {
+				actual, ok := out[id]
+				require.True(t, ok, "entry %s missing from decoded state", id)
+				require.True(t, proto.Equal(expected, actual), "entry %s mismatch after decoding", id)
+				require.Equal(t, c.muted, actual.Entry.MutedAlerts)
+			}
+		})
+	}
+}
+
+// TestStateMergeMutedAlerts checks that merging stays timestamp-based: the
+// newer entry replaces the older one wholesale, muted alerts included, with no
+// per-field merging of the alert lists.
+func TestStateMergeMutedAlerts(t *testing.T) {
+	now := time.Now()
+
+	newEntry := func(ts time.Time, muted []uint64) *pb.MeshEntry {
+		return &pb.MeshEntry{
+			Entry: &pb.Entry{
+				Timestamp:   timestamppb.New(ts),
+				GroupKey:    []byte("key"),
+				Receiver:    &pb.Receiver{GroupName: "a1", Idx: 1, Integration: "integr"},
+				MutedAlerts: muted,
+			},
+			ExpiresAt: timestamppb.New(now.Add(time.Minute)),
+		}
+	}
+
+	const key = "key:a1/integr/1"
+
+	// A newer entry without muted alerts replaces an older one that has them.
+	res := state{key: newEntry(now, []uint64{1, 2})}
+	res.merge(newEntry(now.Add(time.Minute), nil), now)
+	require.Empty(t, res[key].Entry.MutedAlerts)
+
+	// An older entry is dropped, so its muted alerts do not resurface.
+	res = state{key: newEntry(now, nil)}
+	res.merge(newEntry(now.Add(-time.Minute), []uint64{1, 2}), now)
+	require.Empty(t, res[key].Entry.MutedAlerts)
 }
 
 func TestStateDecodingError(t *testing.T) {

nflog/nflogpb/nflog.proto

@@ -37,6 +37,10 @@ message Entry {
   repeated uint64 resolved_alerts = 7;
   // Data specific to the receiver which sent the notification
   map<string, ReceiverDataValue> receiver_data = 8;
+  // MutedAlerts list of hashes of alerts that were muted at the last
+  // notification time, and therefore excluded from FiringAlerts and
+  // ResolvedAlerts.
+  repeated uint64 muted_alerts = 9;
 }
 
 // MeshEntry is a wrapper message to communicate a notify log

notify/notify.go

@@ -141,7 +141,7 @@ func (f StageFunc) Exec(ctx context.Context, l *slog.Logger, alerts ...*alert.Al
 }
 
 type NotificationLog interface {
-	Log(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, store *nflog.Store, expiry time.Duration) error
+	Log(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts, mutedAlerts []uint64, store *nflog.Store, expiry time.Duration) error
 	Query(params ...nflog.QueryParam) ([]*nflogpb.Entry, error)
 }
 

notify/notify_test.go

@@ -59,15 +59,15 @@ type testNflog struct {
 	qres []*nflogpb.Entry
 	qerr error
 
-	logFunc func(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, receiverData *nflog.Store, expiry time.Duration) error
+	logFunc func(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts, mutedAlerts []uint64, receiverData *nflog.Store, expiry time.Duration) error
 }
 
 func (l *testNflog) Query(p ...nflog.QueryParam) ([]*nflogpb.Entry, error) {
 	return l.qres, l.qerr
 }
 
-func (l *testNflog) Log(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, receiverData *nflog.Store, expiry time.Duration) error {
-	return l.logFunc(r, gkey, firingAlerts, resolvedAlerts, receiverData, expiry)
+func (l *testNflog) Log(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts, mutedAlerts []uint64, receiverData *nflog.Store, expiry time.Duration) error {
+	return l.logFunc(r, gkey, firingAlerts, resolvedAlerts, mutedAlerts, receiverData, expiry)
 }
 
 func (l *testNflog) GC() (int, error) {
@@ -667,11 +667,12 @@ func TestSetNotifiesStage(t *testing.T) {
 	ctx = WithResolvedAlerts(ctx, []uint64{})
 	ctx = WithRepeatInterval(ctx, time.Hour)
 
-	tnflog.logFunc = func(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, receiverData *nflog.Store, expiry time.Duration) error {
+	tnflog.logFunc = func(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts, mutedAlerts []uint64, receiverData *nflog.Store, expiry time.Duration) error {
 		require.Equal(t, s.recv, r)
 		require.Equal(t, "1", gkey)
 		require.Equal(t, []uint64{0, 1, 2}, firingAlerts)
 		require.Equal(t, []uint64{}, resolvedAlerts)
+		require.Nil(t, mutedAlerts)
 		require.Equal(t, 2*time.Hour, expiry)
 		return nil
 	}
@@ -683,11 +684,12 @@ func TestSetNotifiesStage(t *testing.T) {
 	ctx = WithFiringAlerts(ctx, []uint64{})
 	ctx = WithResolvedAlerts(ctx, []uint64{0, 1, 2})
 
-	tnflog.logFunc = func(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, receiverData *nflog.Store, expiry time.Duration) error {
+	tnflog.logFunc = func(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts, mutedAlerts []uint64, receiverData *nflog.Store, expiry time.Duration) error {
 		require.Equal(t, s.recv, r)
 		require.Equal(t, "1", gkey)
 		require.Equal(t, []uint64{}, firingAlerts)
 		require.Equal(t, []uint64{0, 1, 2}, resolvedAlerts)
+		require.Nil(t, mutedAlerts)
 		require.Equal(t, 2*time.Hour, expiry)
 		return nil
 	}
@@ -702,7 +704,7 @@ func TestReceiverData_PreservationWhenNotifierDoesNotUpdate(t *testing.T) {
 	callCount := 0
 
 	tnflog := &testNflog{
-		logFunc: func(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, receiverData *nflog.Store, expiry time.Duration) error {
+		logFunc: func(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts, mutedAlerts []uint64, receiverData *nflog.Store, expiry time.Duration) error {
 			storedData = receiverData
 			return nil
 		},
@@ -919,7 +921,7 @@ func TestNflogStore_NoLeakBetweenNotificationSequences(t *testing.T) {
 	var capturedStoreValues []map[string]string
 
 	tnflog := &testNflog{
-		logFunc: func(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, receiverData *nflog.Store, expiry time.Duration) error {
+		logFunc: func(r *nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts, mutedAlerts []uint64, receiverData *nflog.Store, expiry time.Duration) error {
 			storedData = receiverData
 			return nil
 		},

notify/set_notifies_stage.go

@@ -76,5 +76,5 @@ func (n SetNotifiesStage) Exec(ctx context.Context, l *slog.Logger, alerts ...*a
 
 	// Extract receiver data from context if present (it's ok for it to be nil).
 	store, _ := NflogStore(ctx)
-	return ctx, alerts, n.nflog.Log(n.recv, gkey, firing, resolved, store, expiry)
+	return ctx, alerts, n.nflog.Log(n.recv, gkey, firing, resolved, nil, store, expiry)
 }