From b7eb4e7685b487c2ca43e4a24a331e929bf38dec Mon Sep 17 00:00:00 2001 From: Siavash Safi Date: Mon, 31 Aug 2026 12:51:31 +0200 Subject: [PATCH] fix(events): include alert details in all notify events Notification events did not include alert details like fingerprint in all cases. This made it defficult to track an alert end to end when querying events produced by event recorder. This change: - adds firing, muted, resolved alerts to notify context - emits the alerts in all notify events' details - uses fingerprint instead of nflog's xxhash This is follow up for #5409 Signed-off-by: Siavash Safi --- eventrecorder/events.go | 16 ++- eventrecorder/events/v2/events.pb.go | 16 +-- eventrecorder/events_test.go | 15 +++ notify/context.go | 15 +++ notify/event.go | 77 ++++++----- notify/mute.go | 3 +- notify/mute_test.go | 23 ++++ notify/notify_test.go | 145 +++++++++++++++------ notify/retry_stage.go | 2 +- proto/eventrecorder/events/v2/events.proto | 4 +- 10 files changed, 225 insertions(+), 91 deletions(-) diff --git a/eventrecorder/events.go b/eventrecorder/events.go index 929cade8eb..22d83db1ff 100644 --- a/eventrecorder/events.go +++ b/eventrecorder/events.go @@ -133,14 +133,18 @@ func NewAlertGroup(groupKey string, groupLabels model.LabelSet, groupID, receive }} } -// NewGroupedAlert snapshots an alert and its notification-pipeline hash. -func NewGroupedAlert(hash uint64, a *alert.Alert) GroupedAlert { - return GroupedAlert{message: &events.GroupedAlert{Hash: hash, Details: alertToEvents(a)}} +// NewGroupedAlert snapshots an alert and its fingerprint. +func NewGroupedAlert(a *alert.Alert) GroupedAlert { + var fingerprint model.Fingerprint + if a != nil { + fingerprint = a.Fingerprint() + } + return GroupedAlert{message: &events.GroupedAlert{Fingerprint: uint64(fingerprint), Details: alertToEvents(a)}} } -// NewGroupedAlertReference snapshots a hash-only grouped-alert reference. -func NewGroupedAlertReference(hash uint64) GroupedAlert { - return GroupedAlert{message: &events.GroupedAlert{Hash: hash}} +// NewGroupedAlertReference snapshots a fingerprint-only grouped-alert reference. +func NewGroupedAlertReference(fingerprint model.Fingerprint) GroupedAlert { + return GroupedAlert{message: &events.GroupedAlert{Fingerprint: uint64(fingerprint)}} } // NewAlertmanagerStartupEvent constructs startup event data. diff --git a/eventrecorder/events/v2/events.pb.go b/eventrecorder/events/v2/events.pb.go index 374d5776e2..0a6c9f5026 100644 --- a/eventrecorder/events/v2/events.pb.go +++ b/eventrecorder/events/v2/events.pb.go @@ -1,6 +1,6 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 +// protoc-gen-go v1.36.12 // protoc (unknown) // source: events/v2/events.proto @@ -600,10 +600,10 @@ func (x *Alert) GetResolved() bool { return false } -// GroupedAlert is a reference to an alert within an aggregation group. +// GroupedAlert identifies an alert within an aggregation group. type GroupedAlert struct { state protoimpl.MessageState `protogen:"open.v1"` - Hash uint64 `protobuf:"varint,1,opt,name=hash,proto3" json:"hash,omitempty"` + Fingerprint uint64 `protobuf:"varint,1,opt,name=fingerprint,proto3" json:"fingerprint,omitempty"` Details *Alert `protobuf:"bytes,2,opt,name=details,proto3,oneof" json:"details,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -639,9 +639,9 @@ func (*GroupedAlert) Descriptor() ([]byte, []int) { return file_events_v2_events_proto_rawDescGZIP(), []int{5} } -func (x *GroupedAlert) GetHash() uint64 { +func (x *GroupedAlert) GetFingerprint() uint64 { if x != nil { - return x.Hash + return x.Fingerprint } return 0 } @@ -1650,9 +1650,9 @@ const file_events_v2_events_proto_rawDesc = "" + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\x1a>\n" + "\x10AnnotationsEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"_\n" + - "\fGroupedAlert\x12\x12\n" + - "\x04hash\x18\x01 \x01(\x04R\x04hash\x12/\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"m\n" + + "\fGroupedAlert\x12 \n" + + "\vfingerprint\x18\x01 \x01(\x04R\vfingerprint\x12/\n" + "\adetails\x18\x02 \x01(\v2\x10.events.v2.AlertH\x00R\adetails\x88\x01\x01B\n" + "\n" + "\b_details\"\xcb\x02\n" + diff --git a/eventrecorder/events_test.go b/eventrecorder/events_test.go index 57f36180ad..e3d6cfc25f 100644 --- a/eventrecorder/events_test.go +++ b/eventrecorder/events_test.go @@ -32,16 +32,31 @@ func TestAlertEventSnapshotsLabels(t *testing.T) { Labels: model.LabelSet{"alertname": "Down", "severity": "warning"}, Annotations: model.LabelSet{"summary": "test"}, StartsAt: time.Now(), EndsAt: time.Now().Add(time.Hour), }} + fingerprint := a.Fingerprint() event := NewAlertCreatedEvent(a) a.Labels["severity"] = "critical" a.Annotations["summary"] = "changed" got := event.message.GetAlertCreated().Alert + require.Equal(t, uint64(fingerprint), got.Fingerprint) require.Equal(t, "warning", got.Labels["severity"]) require.Equal(t, "test", got.Annotations["summary"]) } +func TestGroupedAlertUsesAlertFingerprint(t *testing.T) { + a := &alert.Alert{Alert: model.Alert{Labels: model.LabelSet{"alertname": "Down", "instance": "api-1"}}} + fingerprint := a.Fingerprint() + + grouped := NewGroupedAlert(a) + require.Equal(t, uint64(fingerprint), grouped.message.Fingerprint) + require.Equal(t, uint64(fingerprint), grouped.message.Details.Fingerprint) + + reference := NewGroupedAlertReference(fingerprint) + require.Equal(t, uint64(fingerprint), reference.message.Fingerprint) + require.Nil(t, reference.message.Details) +} + func TestSilenceEventSnapshotsAnnotationsAndMatchers(t *testing.T) { silence := &silencepb.Silence{ Annotations: map[string]string{"owner": "ops"}, diff --git a/notify/context.go b/notify/context.go index 2d67fbee97..e499da6026 100644 --- a/notify/context.go +++ b/notify/context.go @@ -18,6 +18,7 @@ import ( "github.com/prometheus/common/model" + "github.com/prometheus/alertmanager/alert" "github.com/prometheus/alertmanager/nflog" "github.com/prometheus/alertmanager/pkg/labels" ) @@ -44,6 +45,7 @@ const ( keyFlushID keyGroupMatchers keyRouteLabels + keyMutedAlertDetails ) // WithReceiverName populates a context with a receiver name. @@ -200,6 +202,19 @@ func MutedAlerts(ctx context.Context) (map[uint64]struct{}, bool) { return v, ok } +func withMutedAlertDetails(ctx context.Context, alerts []*alert.Alert) context.Context { + existing, _ := mutedAlertDetails(ctx) + results := make([]*alert.Alert, 0, len(existing)+len(alerts)) + results = append(results, existing...) + results = append(results, alerts...) + return context.WithValue(ctx, keyMutedAlertDetails, results) +} + +func mutedAlertDetails(ctx context.Context) ([]*alert.Alert, bool) { + v, ok := ctx.Value(keyMutedAlertDetails).([]*alert.Alert) + return v, ok +} + // WithAggrGroupID populates a context with an aggregation group UUID. func WithAggrGroupID(ctx context.Context, id string) context.Context { return context.WithValue(ctx, keyAggrGroupID, id) diff --git a/notify/event.go b/notify/event.go index e93a47d396..114843cb50 100644 --- a/notify/event.go +++ b/notify/event.go @@ -14,20 +14,15 @@ package notify // This file contains helpers for constructing event recorder events -// from the notification pipeline context. It lives in the notify package -// because it accesses unexported context keys (keyFiringAlerts, etc.). +// from notification pipeline state that is internal to the notify package. import ( "context" + "github.com/prometheus/alertmanager/alert" "github.com/prometheus/alertmanager/eventrecorder" - "github.com/prometheus/alertmanager/types" ) -func groupedAlertEvent(alert *types.Alert) eventrecorder.GroupedAlert { - return eventrecorder.NewGroupedAlert(hashAlert(alert), alert) -} - func extractAlertGroupInfo(ctx context.Context) eventrecorder.AlertGroup { groupKey, _ := ExtractGroupKey(ctx) receiverName, _ := ReceiverName(ctx) @@ -40,26 +35,40 @@ func extractAlertGroupInfo(ctx context.Context) eventrecorder.AlertGroup { ) } -func extractGroupedAlerts(ctx context.Context, key notifyKey) []eventrecorder.GroupedAlert { - var result []eventrecorder.GroupedAlert - if list, ok := ctx.Value(key).([]uint64); ok { - for _, hash := range list { - result = append(result, eventrecorder.NewGroupedAlertReference(hash)) - } +func alertDetailsByHash(alerts []*alert.Alert) map[uint64]*alert.Alert { + result := make(map[uint64]*alert.Alert, len(alerts)) + for _, alert := range alerts { + result[hashAlert(alert)] = alert } return result } -func extractMutedGroupedAlerts(ctx context.Context) []eventrecorder.GroupedAlert { - var result []eventrecorder.GroupedAlert - if muted, ok := MutedAlerts(ctx); ok { - for hash := range muted { - result = append(result, eventrecorder.NewGroupedAlertReference(hash)) +func alertDetailsForHashes(alerts map[uint64]*alert.Alert, hashes []uint64) []*alert.Alert { + result := make([]*alert.Alert, 0, len(hashes)) + for _, hash := range hashes { + if alert, ok := alerts[hash]; ok { + result = append(result, alert) } } return result } +func groupedAlertsWithDetails(alerts []*alert.Alert) []eventrecorder.GroupedAlert { + result := make([]eventrecorder.GroupedAlert, 0, len(alerts)) + for _, alert := range alerts { + result = append(result, eventrecorder.NewGroupedAlert(alert)) + } + return result +} + +func groupedAlertReferences(alerts []*alert.Alert) []eventrecorder.GroupedAlert { + result := make([]eventrecorder.GroupedAlert, 0, len(alerts)) + for _, alert := range alerts { + result = append(result, eventrecorder.NewGroupedAlertReference(alert.Fingerprint())) + } + return result +} + func notifyReasonToEvent(reason NotifyReason) eventrecorder.NotificationReason { switch reason { case ReasonFirstNotification: @@ -77,23 +86,23 @@ func notifyReasonToEvent(reason NotifyReason) eventrecorder.NotificationReason { } } -// NewNotificationEvent constructs notification event data from the pipeline -// context after a successful notification delivery. -func NewNotificationEvent(ctx context.Context, alerts []*types.Alert, integration Integration) eventrecorder.EventData { - groupedAlerts := make([]eventrecorder.GroupedAlert, 0, len(alerts)) - for _, alert := range alerts { - groupedAlerts = append(groupedAlerts, groupedAlertEvent(alert)) - } - +func newNotificationEvent(ctx context.Context, alerts []*alert.Alert, integration Integration) eventrecorder.EventData { notifyReason, _ := NotificationReason(ctx) repeatInterval, _ := RepeatInterval(ctx) flushID, _ := FlushID(ctx) + firingHashes, _ := FiringAlerts(ctx) + resolvedHashes, _ := ResolvedAlerts(ctx) + details := alertDetailsByHash(alerts) + muted, _ := mutedAlertDetails(ctx) + allAlerts := make([]*alert.Alert, 0, len(alerts)+len(muted)) + allAlerts = append(allAlerts, alerts...) + allAlerts = append(allAlerts, muted...) return eventrecorder.NewNotificationEvent(eventrecorder.Notification{ - Alerts: groupedAlerts, - FiringAlerts: extractGroupedAlerts(ctx, keyFiringAlerts), - ResolvedAlerts: extractGroupedAlerts(ctx, keyResolvedAlerts), - MutedAlerts: extractMutedGroupedAlerts(ctx), + Alerts: groupedAlertsWithDetails(allAlerts), + FiringAlerts: groupedAlertReferences(alertDetailsForHashes(details, firingHashes)), + ResolvedAlerts: groupedAlertReferences(alertDetailsForHashes(details, resolvedHashes)), + MutedAlerts: groupedAlertReferences(muted), Group: extractAlertGroupInfo(ctx), RepeatInterval: repeatInterval, Reason: notifyReasonToEvent(notifyReason), @@ -104,11 +113,11 @@ func NewNotificationEvent(ctx context.Context, alerts []*types.Alert, integratio } // NewAlertResolvedEvent constructs alert-resolved event data. -func NewAlertResolvedEvent(groupInfo eventrecorder.AlertGroup, alert *types.Alert) eventrecorder.EventData { - return eventrecorder.NewAlertResolvedEvent(groupInfo, groupedAlertEvent(alert)) +func NewAlertResolvedEvent(groupInfo eventrecorder.AlertGroup, alert *alert.Alert) eventrecorder.EventData { + return eventrecorder.NewAlertResolvedEvent(groupInfo, eventrecorder.NewGroupedAlert(alert)) } // NewAlertGroupedEvent constructs alert-grouped event data. -func NewAlertGroupedEvent(groupInfo eventrecorder.AlertGroup, alert *types.Alert) eventrecorder.EventData { - return eventrecorder.NewAlertGroupedEvent(groupInfo, groupedAlertEvent(alert)) +func NewAlertGroupedEvent(groupInfo eventrecorder.AlertGroup, alert *alert.Alert) eventrecorder.EventData { + return eventrecorder.NewAlertGroupedEvent(groupInfo, eventrecorder.NewGroupedAlert(alert)) } diff --git a/notify/mute.go b/notify/mute.go index 604f17e8a1..3d40be781f 100644 --- a/notify/mute.go +++ b/notify/mute.go @@ -61,7 +61,8 @@ func recordMuted(ctx context.Context, muted []*alert.Alert) context.Context { for _, a := range muted { hashes[hashAlert(a)] = struct{}{} } - return WithMutedAlerts(ctx, hashes) + ctx = WithMutedAlerts(ctx, hashes) + return withMutedAlertDetails(ctx, muted) } // MuteStage filters alerts through a Muter. diff --git a/notify/mute_test.go b/notify/mute_test.go index e92e241554..58e24bd05b 100644 --- a/notify/mute_test.go +++ b/notify/mute_test.go @@ -91,6 +91,29 @@ func TestMuteStage(t *testing.T) { } } +func TestMuteStageAccumulatesMutedAlertDetails(t *testing.T) { + metrics := NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{}) + firstStage := NewMuteStage(MuteFunc(func(_ context.Context, lset model.LabelSet) bool { + return lset["muted_by"] == "first" + }), metrics) + secondStage := NewMuteStage(MuteFunc(func(_ context.Context, lset model.LabelSet) bool { + return lset["muted_by"] == "second" + }), metrics) + first := &alert.Alert{Alert: model.Alert{Labels: model.LabelSet{"alertname": "First", "muted_by": "first"}}} + second := &alert.Alert{Alert: model.Alert{Labels: model.LabelSet{"alertname": "Second", "muted_by": "second"}}} + active := &alert.Alert{Alert: model.Alert{Labels: model.LabelSet{"alertname": "Active"}}} + + ctx, alerts, err := firstStage.Exec(context.Background(), promslog.NewNopLogger(), first, second, active) + require.NoError(t, err) + ctx, alerts, err = secondStage.Exec(ctx, promslog.NewNopLogger(), alerts...) + require.NoError(t, err) + require.Equal(t, []*alert.Alert{active}, alerts) + + muted, ok := mutedAlertDetails(ctx) + require.True(t, ok) + require.Equal(t, []*alert.Alert{first, second}, muted) +} + func TestMuteStageWithSilences(t *testing.T) { silences, err := silence.New(silence.Options{Metrics: prometheus.NewRegistry(), Retention: time.Hour}) if err != nil { diff --git a/notify/notify_test.go b/notify/notify_test.go index c7b90a658d..e536223375 100644 --- a/notify/notify_test.go +++ b/notify/notify_test.go @@ -19,7 +19,10 @@ import ( "fmt" "io" "log/slog" + "os" + "path/filepath" "reflect" + "strings" "testing" "time" @@ -28,9 +31,12 @@ import ( "github.com/prometheus/common/model" "github.com/prometheus/common/promslog" "github.com/stretchr/testify/require" + "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/types/known/timestamppb" + "github.com/prometheus/alertmanager/alert" "github.com/prometheus/alertmanager/eventrecorder" + eventsv2 "github.com/prometheus/alertmanager/eventrecorder/events/v2" "github.com/prometheus/alertmanager/featurecontrol" "github.com/prometheus/alertmanager/nflog" "github.com/prometheus/alertmanager/nflog/nflogpb" @@ -48,15 +54,15 @@ func (s sendResolved) SendResolved() bool { return bool(s) } -type notifierFunc func(ctx context.Context, alerts ...*types.Alert) (bool, error) +type notifierFunc func(ctx context.Context, alerts ...*alert.Alert) (bool, error) -func (f notifierFunc) Notify(ctx context.Context, alerts ...*types.Alert) (bool, error) { +func (f notifierFunc) Notify(ctx context.Context, alerts ...*alert.Alert) (bool, error) { return f(ctx, alerts...) } type failStage struct{} -func (s failStage) Exec(ctx context.Context, l *slog.Logger, as ...*types.Alert) (context.Context, []*types.Alert, error) { +func (s failStage) Exec(ctx context.Context, l *slog.Logger, as ...*alert.Alert) (context.Context, []*alert.Alert, error) { return ctx, nil, fmt.Errorf("some error") } @@ -221,7 +227,7 @@ func TestDedupStageNeedsUpdate(t *testing.T) { func TestDedupStageUsesContextNow(t *testing.T) { base := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC) s := &DedupStage{ - hash: func(*types.Alert) uint64 { return 1 }, + hash: func(*alert.Alert) uint64 { return 1 }, now: func() time.Time { return base.Add(time.Hour) }, @@ -240,7 +246,7 @@ func TestDedupStageUsesContextNow(t *testing.T) { ctx = WithRepeatInterval(ctx, 30*time.Minute) ctx = WithNow(ctx, base.Add(10*time.Minute)) - alerts := []*types.Alert{{Alert: model.Alert{Labels: model.LabelSet{"alertname": "test"}}}} + alerts := []*alert.Alert{{Alert: model.Alert{Labels: model.LabelSet{"alertname": "test"}}}} _, res, err := s.Exec(ctx, promslog.NewNopLogger(), alerts...) require.NoError(t, err) @@ -251,7 +257,7 @@ func TestDedupStage(t *testing.T) { i := 0 now := utcNow() s := &DedupStage{ - hash: func(a *types.Alert) uint64 { + hash: func(a *alert.Alert) uint64 { res := uint64(i) i++ return res @@ -274,7 +280,7 @@ func TestDedupStage(t *testing.T) { ctx = WithRepeatInterval(ctx, time.Hour) - alerts := []*types.Alert{{}, {}, {}} + alerts := []*alert.Alert{{}, {}, {}} // Must catch notification log query errors. s.nflog = &testNflog{ @@ -343,13 +349,13 @@ func TestDedupStage(t *testing.T) { func TestMultiStage(t *testing.T) { var ( - alerts1 = []*types.Alert{{}} - alerts2 = []*types.Alert{{}, {}} - alerts3 = []*types.Alert{{}, {}, {}} + alerts1 = []*alert.Alert{{}} + alerts2 = []*alert.Alert{{}, {}} + alerts3 = []*alert.Alert{{}, {}, {}} ) stage := MultiStage{ - StageFunc(func(ctx context.Context, l *slog.Logger, alerts ...*types.Alert) (context.Context, []*types.Alert, error) { + StageFunc(func(ctx context.Context, l *slog.Logger, alerts ...*alert.Alert) (context.Context, []*alert.Alert, error) { if !reflect.DeepEqual(alerts, alerts1) { t.Fatal("Input not equal to input of MultiStage") } @@ -357,7 +363,7 @@ func TestMultiStage(t *testing.T) { ctx = context.WithValue(ctx, "key", "value") return ctx, alerts2, nil }), - StageFunc(func(ctx context.Context, l *slog.Logger, alerts ...*types.Alert) (context.Context, []*types.Alert, error) { + StageFunc(func(ctx context.Context, l *slog.Logger, alerts ...*alert.Alert) (context.Context, []*alert.Alert, error) { if !reflect.DeepEqual(alerts, alerts2) { t.Fatal("Input not equal to output of previous stage") } @@ -464,12 +470,12 @@ func TestRetryStageSkipsMutedGroup(t *testing.T) { func TestRoutingStage(t *testing.T) { var ( - alerts1 = []*types.Alert{{}} - alerts2 = []*types.Alert{{}, {}} + alerts1 = []*alert.Alert{{}} + alerts2 = []*alert.Alert{{}, {}} ) stage := RoutingStage{ - "name": StageFunc(func(ctx context.Context, l *slog.Logger, alerts ...*types.Alert) (context.Context, []*types.Alert, error) { + "name": StageFunc(func(ctx context.Context, l *slog.Logger, alerts ...*alert.Alert) (context.Context, []*alert.Alert, error) { if !reflect.DeepEqual(alerts, alerts1) { t.Fatal("Input not equal to input of RoutingStage") } @@ -492,9 +498,9 @@ func TestRoutingStage(t *testing.T) { func TestRetryStageWithError(t *testing.T) { fail, retry := true, true - sent := []*types.Alert{} + sent := []*alert.Alert{} i := Integration{ - notifier: notifierFunc(func(ctx context.Context, alerts ...*types.Alert) (bool, error) { + notifier: notifierFunc(func(ctx context.Context, alerts ...*alert.Alert) (bool, error) { if fail { fail = false return retry, errors.New("fail to deliver notification") @@ -506,7 +512,7 @@ func TestRetryStageWithError(t *testing.T) { } r := NewRetryStage(i, "", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{}), eventrecorder.NopRecorder()) - alerts := []*types.Alert{ + alerts := []*alert.Alert{ { Alert: model.Alert{ EndsAt: time.Now().Add(time.Hour), @@ -549,7 +555,7 @@ func TestRetryStageWithErrorCode(t *testing.T) { testData := testData i := Integration{ name: "test", - notifier: notifierFunc(func(ctx context.Context, alerts ...*types.Alert) (bool, error) { + notifier: notifierFunc(func(ctx context.Context, alerts ...*alert.Alert) (bool, error) { if !testData.isNewErrorWithReason { return retry, errors.New("fail to deliver notification") } @@ -559,7 +565,7 @@ func TestRetryStageWithErrorCode(t *testing.T) { } r := NewRetryStage(i, "", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{}), eventrecorder.NopRecorder()) - alerts := []*types.Alert{ + alerts := []*alert.Alert{ { Alert: model.Alert{ EndsAt: time.Now().Add(time.Hour), @@ -586,7 +592,7 @@ func TestRetryStageWithContextCanceled(t *testing.T) { i := Integration{ name: "test", - notifier: notifierFunc(func(ctx context.Context, alerts ...*types.Alert) (bool, error) { + notifier: notifierFunc(func(ctx context.Context, alerts ...*alert.Alert) (bool, error) { cancel() return true, errors.New("request failed: context canceled") }), @@ -594,7 +600,7 @@ func TestRetryStageWithContextCanceled(t *testing.T) { } r := NewRetryStage(i, "", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{}), eventrecorder.NopRecorder()) - alerts := []*types.Alert{ + alerts := []*alert.Alert{ { Alert: model.Alert{ EndsAt: time.Now().Add(time.Hour), @@ -616,9 +622,9 @@ func TestRetryStageWithContextCanceled(t *testing.T) { } func TestRetryStageNoResolved(t *testing.T) { - sent := []*types.Alert{} + sent := []*alert.Alert{} i := Integration{ - notifier: notifierFunc(func(ctx context.Context, alerts ...*types.Alert) (bool, error) { + notifier: notifierFunc(func(ctx context.Context, alerts ...*alert.Alert) (bool, error) { sent = append(sent, alerts...) return false, nil }), @@ -626,7 +632,7 @@ func TestRetryStageNoResolved(t *testing.T) { } r := NewRetryStage(i, "", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{}), eventrecorder.NopRecorder()) - alerts := []*types.Alert{ + alerts := []*alert.Alert{ { Alert: model.Alert{ EndsAt: time.Now().Add(-time.Hour), @@ -651,7 +657,7 @@ func TestRetryStageNoResolved(t *testing.T) { resctx, res, err = r.Exec(ctx, promslog.NewNopLogger(), alerts...) require.NoError(t, err) require.Equal(t, alerts, res) - require.Equal(t, []*types.Alert{alerts[1]}, sent) + require.Equal(t, []*alert.Alert{alerts[1]}, sent) require.NotNil(t, resctx) // All alerts are resolved. @@ -662,14 +668,75 @@ func TestRetryStageNoResolved(t *testing.T) { resctx, res, err = r.Exec(ctx, promslog.NewNopLogger(), alerts...) require.NoError(t, err) require.Equal(t, alerts, res) - require.Equal(t, []*types.Alert{}, sent) + require.Equal(t, []*alert.Alert{}, sent) require.NotNil(t, resctx) } +func TestRetryStageNotificationEventUsesDedupAlertState(t *testing.T) { + path := filepath.Join(t.TempDir(), "events.jsonl") + recorder := eventrecorder.NewRecorderFromConfig(eventrecorder.Config{ + FileOutputs: []eventrecorder.FileOutputConfig{{Name: "test", Path: path}}, + }, "test", promslog.NewNopLogger(), nil) + t.Cleanup(func() { require.NoError(t, recorder.Close()) }) + firing := &alert.Alert{Alert: model.Alert{ + Labels: model.LabelSet{"alertname": "Firing", "instance": "api-1"}, StartsAt: time.Now(), EndsAt: time.Now().Add(time.Hour), + }} + resolved := &alert.Alert{Alert: model.Alert{ + Labels: model.LabelSet{"alertname": "Resolved", "instance": "api-2"}, StartsAt: time.Now().Add(-time.Hour), EndsAt: time.Now().Add(-time.Minute), + }} + muted := &alert.Alert{Alert: model.Alert{ + Labels: model.LabelSet{"alertname": "Muted", "instance": "api-3"}, StartsAt: time.Now(), EndsAt: time.Now().Add(time.Hour), + }} + var sent []*alert.Alert + integration := NewIntegration(notifierFunc(func(_ context.Context, alerts ...*alert.Alert) (bool, error) { + sent = append(sent, alerts...) + firing.EndsAt = time.Now().Add(-time.Minute) + return false, nil + }), sendResolved(false), "webhook", 2, "test") + dedup := NewDedupStage(&integration, &testNflog{}, &nflogpb.Receiver{}) + stage := NewRetryStage(integration, "test", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{}), recorder) + ctx := eventrecorder.WithEventRecording(context.Background()) + ctx = WithGroupKey(ctx, "group") + ctx = WithRepeatInterval(ctx, time.Hour) + ctx = withMutedAlertDetails(ctx, []*alert.Alert{muted}) + ctx, alerts, err := dedup.Exec(ctx, promslog.NewNopLogger(), firing, resolved) + require.NoError(t, err) + + _, alerts, err = stage.Exec(ctx, promslog.NewNopLogger(), alerts...) + require.NoError(t, err) + require.Equal(t, []*alert.Alert{firing, resolved}, alerts) + require.Equal(t, []*alert.Alert{firing}, sent) + require.True(t, firing.Resolved()) + require.NoError(t, recorder.Close()) + + data, err := os.ReadFile(path) + require.NoError(t, err) + var recorded eventsv2.Event + require.NoError(t, protojson.Unmarshal([]byte(strings.TrimSpace(string(data))), &recorded)) + notification := recorded.GetData().GetNotification() + require.Len(t, notification.Alerts, 3) + require.Len(t, notification.FiringAlerts, 1) + require.Len(t, notification.ResolvedAlerts, 1) + require.Len(t, notification.MutedAlerts, 1) + require.Equal(t, uint64(firing.Fingerprint()), notification.Alerts[0].Fingerprint) + require.Equal(t, uint64(resolved.Fingerprint()), notification.Alerts[1].Fingerprint) + require.Equal(t, uint64(muted.Fingerprint()), notification.Alerts[2].Fingerprint) + require.NotNil(t, notification.Alerts[0].Details) + require.NotNil(t, notification.Alerts[1].Details) + require.NotNil(t, notification.Alerts[2].Details) + require.Equal(t, "api-3", notification.Alerts[2].Details.Labels["instance"]) + require.Equal(t, uint64(firing.Fingerprint()), notification.FiringAlerts[0].Fingerprint) + require.Equal(t, uint64(resolved.Fingerprint()), notification.ResolvedAlerts[0].Fingerprint) + require.Equal(t, uint64(muted.Fingerprint()), notification.MutedAlerts[0].Fingerprint) + require.Nil(t, notification.FiringAlerts[0].Details) + require.Nil(t, notification.ResolvedAlerts[0].Details) + require.Nil(t, notification.MutedAlerts[0].Details) +} + func TestRetryStageSendResolved(t *testing.T) { - sent := []*types.Alert{} + sent := []*alert.Alert{} i := Integration{ - notifier: notifierFunc(func(ctx context.Context, alerts ...*types.Alert) (bool, error) { + notifier: notifierFunc(func(ctx context.Context, alerts ...*alert.Alert) (bool, error) { sent = append(sent, alerts...) return false, nil }), @@ -677,7 +744,7 @@ func TestRetryStageSendResolved(t *testing.T) { } r := NewRetryStage(i, "", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{}), eventrecorder.NopRecorder()) - alerts := []*types.Alert{ + alerts := []*alert.Alert{ { Alert: model.Alert{ EndsAt: time.Now().Add(-time.Hour), @@ -718,7 +785,7 @@ func TestSetNotifiesStage(t *testing.T) { nflog: tnflog, ff: featurecontrol.NoopFlags{}, } - alerts := []*types.Alert{{}, {}, {}} + alerts := []*alert.Alert{{}, {}, {}} ctx := context.Background() resctx, res, err := s.Exec(ctx, promslog.NewNopLogger(), alerts...) @@ -922,7 +989,7 @@ func TestReceiverData_PreservationWhenNotifierDoesNotUpdate(t *testing.T) { recv := &nflogpb.Receiver{GroupName: "test"} dedupStage := NewDedupStage(sendResolved(true), tnflog, recv) - notifier := notifierFunc(func(ctx context.Context, alerts ...*types.Alert) (bool, error) { + notifier := notifierFunc(func(ctx context.Context, alerts ...*alert.Alert) (bool, error) { callCount++ if callCount == 1 { @@ -945,7 +1012,7 @@ func TestReceiverData_PreservationWhenNotifierDoesNotUpdate(t *testing.T) { ctx = WithGroupKey(ctx, "testkey") ctx = WithRepeatInterval(ctx, time.Hour) - alerts := []*types.Alert{ + alerts := []*alert.Alert{ { Alert: model.Alert{ Labels: model.LabelSet{"alertname": "test"}, @@ -1037,7 +1104,7 @@ func TestDedupStageExtractsReceiverData_DataPresent(t *testing.T) { ctx = WithGroupKey(ctx, "key") ctx = WithRepeatInterval(ctx, time.Hour) - alerts := []*types.Alert{ + alerts := []*alert.Alert{ { Alert: model.Alert{ Labels: model.LabelSet{"alertname": "test"}, @@ -1079,7 +1146,7 @@ func TestDedupStageExtractsReceiverData_NilReceiverData(t *testing.T) { ctx = WithGroupKey(ctx, "key") ctx = WithRepeatInterval(ctx, time.Hour) - alerts := []*types.Alert{ + alerts := []*alert.Alert{ { Alert: model.Alert{ Labels: model.LabelSet{"alertname": "test"}, @@ -1106,7 +1173,7 @@ func TestDedupStageExtractsReceiverData_NoEntry(t *testing.T) { ctx = WithGroupKey(ctx, "key") ctx = WithRepeatInterval(ctx, time.Hour) - alerts := []*types.Alert{ + alerts := []*alert.Alert{ { Alert: model.Alert{ Labels: model.LabelSet{"alertname": "test"}, @@ -1137,7 +1204,7 @@ func TestNflogStore_NoLeakBetweenNotificationSequences(t *testing.T) { recv := &nflogpb.Receiver{GroupName: "test"} dedupStage := NewDedupStage(sendResolved(true), tnflog, recv) - notifier := notifierFunc(func(ctx context.Context, alerts ...*types.Alert) (bool, error) { + notifier := notifierFunc(func(ctx context.Context, alerts ...*alert.Alert) (bool, error) { callCount++ store, ok := NflogStore(ctx) require.True(t, ok, "Store should be available in context") @@ -1156,7 +1223,7 @@ func TestNflogStore_NoLeakBetweenNotificationSequences(t *testing.T) { retryStage := NewRetryStage(integration, "test", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{}), eventrecorder.NopRecorder()) setNotifiesStage := NewSetNotifiesStage(tnflog, recv, featurecontrol.NoopFlags{}) - alerts := []*types.Alert{ + alerts := []*alert.Alert{ { Alert: model.Alert{ Labels: model.LabelSet{"alertname": "test"}, @@ -1230,7 +1297,7 @@ func TestNflogStore_NoLeakBetweenNotificationSequences(t *testing.T) { } func BenchmarkHashAlert(b *testing.B) { - alert := &types.Alert{ + alert := &alert.Alert{ Alert: model.Alert{ Labels: model.LabelSet{"foo": "the_first_value", "bar": "the_second_value", "another": "value"}, }, diff --git a/notify/retry_stage.go b/notify/retry_stage.go index adb055840c..7c71f9c7d5 100644 --- a/notify/retry_stage.go +++ b/notify/retry_stage.go @@ -185,7 +185,7 @@ func (r RetryStage) exec(ctx context.Context, l *slog.Logger, alerts ...*alert.A } r.recorder.RecordEvent(ctx, func() eventrecorder.EventData { - return NewNotificationEvent(ctx, sent, r.integration) + return newNotificationEvent(ctx, alerts, r.integration) }) return ctx, alerts, nil } diff --git a/proto/eventrecorder/events/v2/events.proto b/proto/eventrecorder/events/v2/events.proto index 5d1ef36c46..0f33e1d758 100644 --- a/proto/eventrecorder/events/v2/events.proto +++ b/proto/eventrecorder/events/v2/events.proto @@ -51,9 +51,9 @@ message Alert { bool resolved = 7; } -// GroupedAlert is a reference to an alert within an aggregation group. +// GroupedAlert identifies an alert within an aggregation group. message GroupedAlert { - uint64 hash = 1; + uint64 fingerprint = 1; optional Alert details = 2; }