From 742448efe9aa6fc9c555366b1af893f1a6736dd4 Mon Sep 17 00:00:00 2001 From: Geoff Wilson Date: Wed, 26 Aug 2026 10:55:32 -0400 Subject: [PATCH 1/2] nflog: add muted_alerts to the notification log entry Entry records the alerts that were firing and resolved at the time of the last notification, but not the ones that were muted. The mute stages drop muted alerts before the dedup stage runs, so they are absent from both lists and indistinguishable from alerts that stopped firing. Add a muted_alerts field alongside them, an IsMutedSubset helper mirroring IsFiringSubset and IsResolvedSubset, and a mutedAlerts parameter on Log(). The field number 9 is new; existing field numbers are untouched, so entries gossiped by peers that predate this field decode with MutedAlerts unset, and entries this version writes are readable by those peers. Nothing populates the field yet: SetNotifiesStage passes nil, and the dedup stage does not read it. This is the schema and write path only, so that the behavior change can be reviewed on its own. Generated code regenerated with scripts/genproto.sh. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FqRsyADYBfpUxp1JpzQsW8 --- nflog/nflog.go | 3 +- nflog/nflog_test.go | 82 +++++++++++++++++++++++++++++++++++- nflog/nflogpb/nflog.pb.go | 18 ++++++-- nflog/nflogpb/nflog.proto | 4 ++ nflog/nflogpb/set.go | 11 +++++ nflog/nflogpb/set_test.go | 44 +++++++++++++++++++ notify/notify.go | 2 +- notify/notify_test.go | 16 ++++--- notify/set_notifies_stage.go | 2 +- 9 files changed, 168 insertions(+), 14 deletions(-) diff --git a/nflog/nflog.go b/nflog/nflog.go index fd51e9c972..20cfd2f665 100644 --- a/nflog/nflog.go +++ b/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), diff --git a/nflog/nflog_test.go b/nflog/nflog_test.go index 538182b01c..7904788805 100644 --- a/nflog/nflog_test.go +++ b/nflog/nflog_test.go @@ -367,8 +367,9 @@ 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)) @@ -376,6 +377,85 @@ func TestQuery(t *testing.T) { 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) { diff --git a/nflog/nflogpb/nflog.pb.go b/nflog/nflogpb/nflog.pb.go index f7ea274b1d..20039dc020 100644 --- a/nflog/nflogpb/nflog.pb.go +++ b/nflog/nflogpb/nflog.pb.go @@ -107,7 +107,11 @@ type Entry struct { // ResolvedAlerts list of hashes of resolved alerts at the last notification time. ResolvedAlerts []uint64 `protobuf:"varint,7,rep,packed,name=resolved_alerts,json=resolvedAlerts,proto3" json:"resolved_alerts,omitempty"` // Data specific to the receiver which sent the notification - ReceiverData map[string]*ReceiverDataValue `protobuf:"bytes,8,rep,name=receiver_data,json=receiverData,proto3" json:"receiver_data,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + ReceiverData map[string]*ReceiverDataValue `protobuf:"bytes,8,rep,name=receiver_data,json=receiverData,proto3" json:"receiver_data,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + // MutedAlerts list of hashes of alerts that were muted at the last + // notification time, and therefore excluded from FiringAlerts and + // ResolvedAlerts. + MutedAlerts []uint64 `protobuf:"varint,9,rep,packed,name=muted_alerts,json=mutedAlerts,proto3" json:"muted_alerts,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -198,6 +202,13 @@ func (x *Entry) GetReceiverData() map[string]*ReceiverDataValue { return nil } +func (x *Entry) GetMutedAlerts() []uint64 { + if x != nil { + return x.MutedAlerts + } + return nil +} + // MeshEntry is a wrapper message to communicate a notify log // entry through a mesh network. type MeshEntry struct { @@ -362,7 +373,7 @@ const file_nflog_proto_rawDesc = "" + "\n" + "group_name\x18\x01 \x01(\tR\tgroupName\x12 \n" + "\vintegration\x18\x02 \x01(\tR\vintegration\x12\x10\n" + - "\x03idx\x18\x03 \x01(\rR\x03idx\"\xba\x03\n" + + "\x03idx\x18\x03 \x01(\rR\x03idx\"\xdd\x03\n" + "\x05Entry\x12\x1b\n" + "\tgroup_key\x18\x01 \x01(\fR\bgroupKey\x12-\n" + "\breceiver\x18\x02 \x01(\v2\x11.nflogpb.ReceiverR\breceiver\x12\x1d\n" + @@ -372,7 +383,8 @@ const file_nflog_proto_rawDesc = "" + "\ttimestamp\x18\x05 \x01(\v2\x1a.google.protobuf.TimestampR\ttimestamp\x12#\n" + "\rfiring_alerts\x18\x06 \x03(\x04R\ffiringAlerts\x12'\n" + "\x0fresolved_alerts\x18\a \x03(\x04R\x0eresolvedAlerts\x12E\n" + - "\rreceiver_data\x18\b \x03(\v2 .nflogpb.Entry.ReceiverDataEntryR\freceiverData\x1a[\n" + + "\rreceiver_data\x18\b \x03(\v2 .nflogpb.Entry.ReceiverDataEntryR\freceiverData\x12!\n" + + "\fmuted_alerts\x18\t \x03(\x04R\vmutedAlerts\x1a[\n" + "\x11ReceiverDataEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x120\n" + "\x05value\x18\x02 \x01(\v2\x1a.nflogpb.ReceiverDataValueR\x05value:\x028\x01\"l\n" + diff --git a/nflog/nflogpb/nflog.proto b/nflog/nflogpb/nflog.proto index 65fcb32846..026526f01b 100644 --- a/nflog/nflogpb/nflog.proto +++ b/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 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 diff --git a/nflog/nflogpb/set.go b/nflog/nflogpb/set.go index 698f7f2dac..b0f7fb6c1f 100644 --- a/nflog/nflogpb/set.go +++ b/nflog/nflogpb/set.go @@ -35,6 +35,17 @@ func (m *Entry) IsResolvedSubset(subset map[uint64]struct{}) bool { return isSubset(set, subset) } +// IsMutedSubset returns whether the given subset is a subset of the alerts +// that were muted at the time of the last notification. +func (m *Entry) IsMutedSubset(subset map[uint64]struct{}) bool { + set := map[uint64]struct{}{} + for i := range m.MutedAlerts { + set[m.MutedAlerts[i]] = struct{}{} + } + + return isSubset(set, subset) +} + func isSubset(set, subset map[uint64]struct{}) bool { for k := range subset { _, exists := set[k] diff --git a/nflog/nflogpb/set_test.go b/nflog/nflogpb/set_test.go index e1ce289dcc..b973e520c3 100644 --- a/nflog/nflogpb/set_test.go +++ b/nflog/nflogpb/set_test.go @@ -73,6 +73,50 @@ func TestIsResolvedSubset(t *testing.T) { } } +func TestIsMutedSubset(t *testing.T) { + e := &Entry{ + MutedAlerts: []uint64{1, 2, 3}, + } + + tests := []struct { + subset map[uint64]struct{} + expected bool + }{ + {newSubset(), true}, // empty subset + {newSubset(1), true}, + {newSubset(2), true}, + {newSubset(3), true}, + {newSubset(1, 2), true}, + {newSubset(1, 2), true}, + {newSubset(1, 2, 3), true}, + {newSubset(4), false}, + {newSubset(1, 5), false}, + {newSubset(1, 2, 3, 6), false}, + } + + for _, test := range tests { + if result := e.IsMutedSubset(test.subset); result != test.expected { + t.Errorf("Expected %t, got %t for subset %v", test.expected, result, elements(test.subset)) + } + } +} + +// TestIsMutedSubsetWithoutMutedAlerts covers entries written by a peer that +// does not know about the muted_alerts field. Only the empty subset is a +// subset of no muted alerts. +func TestIsMutedSubsetWithoutMutedAlerts(t *testing.T) { + e := &Entry{ + FiringAlerts: []uint64{1, 2, 3}, + } + + if result := e.IsMutedSubset(newSubset()); !result { + t.Errorf("Expected true, got false for the empty subset") + } + if result := e.IsMutedSubset(newSubset(1)); result { + t.Errorf("Expected false, got true for subset [1]") + } +} + func newSubset(elements ...uint64) map[uint64]struct{} { subset := make(map[uint64]struct{}) for _, el := range elements { diff --git a/notify/notify.go b/notify/notify.go index 0bb2dc6e62..ce638936fe 100644 --- a/notify/notify.go +++ b/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) } diff --git a/notify/notify_test.go b/notify/notify_test.go index a62a15df73..45ccb023fa 100644 --- a/notify/notify_test.go +++ b/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 }, diff --git a/notify/set_notifies_stage.go b/notify/set_notifies_stage.go index 57d501dee7..a308818619 100644 --- a/notify/set_notifies_stage.go +++ b/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) } From 23fbcb539acd61dbfd18e691693c578a50502c6f Mon Sep 17 00:00:00 2001 From: Geoff Wilson Date: Thu, 27 Aug 2026 15:11:43 -0400 Subject: [PATCH 2/2] Remove IsMutedSubset func and tests Signed-off-by: Geoff Wilson --- nflog/nflogpb/set.go | 11 ---------- nflog/nflogpb/set_test.go | 44 --------------------------------------- 2 files changed, 55 deletions(-) diff --git a/nflog/nflogpb/set.go b/nflog/nflogpb/set.go index b0f7fb6c1f..698f7f2dac 100644 --- a/nflog/nflogpb/set.go +++ b/nflog/nflogpb/set.go @@ -35,17 +35,6 @@ func (m *Entry) IsResolvedSubset(subset map[uint64]struct{}) bool { return isSubset(set, subset) } -// IsMutedSubset returns whether the given subset is a subset of the alerts -// that were muted at the time of the last notification. -func (m *Entry) IsMutedSubset(subset map[uint64]struct{}) bool { - set := map[uint64]struct{}{} - for i := range m.MutedAlerts { - set[m.MutedAlerts[i]] = struct{}{} - } - - return isSubset(set, subset) -} - func isSubset(set, subset map[uint64]struct{}) bool { for k := range subset { _, exists := set[k] diff --git a/nflog/nflogpb/set_test.go b/nflog/nflogpb/set_test.go index b973e520c3..e1ce289dcc 100644 --- a/nflog/nflogpb/set_test.go +++ b/nflog/nflogpb/set_test.go @@ -73,50 +73,6 @@ func TestIsResolvedSubset(t *testing.T) { } } -func TestIsMutedSubset(t *testing.T) { - e := &Entry{ - MutedAlerts: []uint64{1, 2, 3}, - } - - tests := []struct { - subset map[uint64]struct{} - expected bool - }{ - {newSubset(), true}, // empty subset - {newSubset(1), true}, - {newSubset(2), true}, - {newSubset(3), true}, - {newSubset(1, 2), true}, - {newSubset(1, 2), true}, - {newSubset(1, 2, 3), true}, - {newSubset(4), false}, - {newSubset(1, 5), false}, - {newSubset(1, 2, 3, 6), false}, - } - - for _, test := range tests { - if result := e.IsMutedSubset(test.subset); result != test.expected { - t.Errorf("Expected %t, got %t for subset %v", test.expected, result, elements(test.subset)) - } - } -} - -// TestIsMutedSubsetWithoutMutedAlerts covers entries written by a peer that -// does not know about the muted_alerts field. Only the empty subset is a -// subset of no muted alerts. -func TestIsMutedSubsetWithoutMutedAlerts(t *testing.T) { - e := &Entry{ - FiringAlerts: []uint64{1, 2, 3}, - } - - if result := e.IsMutedSubset(newSubset()); !result { - t.Errorf("Expected true, got false for the empty subset") - } - if result := e.IsMutedSubset(newSubset(1)); result { - t.Errorf("Expected false, got true for subset [1]") - } -} - func newSubset(elements ...uint64) map[uint64]struct{} { subset := make(map[uint64]struct{}) for _, el := range elements {