diff --git a/packages/grafana-data/src/types/featureToggles.gen.ts b/packages/grafana-data/src/types/featureToggles.gen.ts index aa0581004e1..cdcd4b53092 100644 --- a/packages/grafana-data/src/types/featureToggles.gen.ts +++ b/packages/grafana-data/src/types/featureToggles.gen.ts @@ -1251,4 +1251,8 @@ export interface FeatureToggles { * Enables profiles exemplars support in profiles drilldown */ profilesExemplars?: boolean; + /** + * Use synchronized dispatch timer to minimize duplicate notifications across alertmanager HA pods + */ + alertingSyncDispatchTimer?: boolean; } diff --git a/pkg/services/featuremgmt/registry.go b/pkg/services/featuremgmt/registry.go index 4c0456e9457..6b716789f8c 100644 --- a/pkg/services/featuremgmt/registry.go +++ b/pkg/services/featuremgmt/registry.go @@ -981,7 +981,8 @@ var ( Stage: FeatureStageDeprecated, Owner: grafanaPartnerPluginsSquad, Expression: "true", // Enabled by default for now - }, { + }, + { Name: "alertingFilterV2", Description: "Enable the new alerting search experience", Stage: FeatureStageExperimental, @@ -2069,6 +2070,14 @@ var ( Owner: grafanaObservabilityTracesAndProfilingSquad, FrontendOnly: false, }, + { + Name: "alertingSyncDispatchTimer", + Description: "Use synchronized dispatch timer to minimize duplicate notifications across alertmanager HA pods", + Stage: FeatureStageExperimental, + Owner: grafanaAlertingSquad, + RequiresRestart: true, + HideFromDocs: true, + }, } ) diff --git a/pkg/services/featuremgmt/toggles_gen.csv b/pkg/services/featuremgmt/toggles_gen.csv index caba7cdab90..09c75a82041 100644 --- a/pkg/services/featuremgmt/toggles_gen.csv +++ b/pkg/services/featuremgmt/toggles_gen.csv @@ -280,3 +280,4 @@ multiPropsVariables,experimental,@grafana/dashboards-squad,false,false,true smoothingTransformation,experimental,@grafana/datapro,false,false,true secretsManagementAppPlatformAwsKeeper,experimental,@grafana/grafana-operator-experience-squad,false,false,false profilesExemplars,experimental,@grafana/observability-traces-and-profiling,false,false,false +alertingSyncDispatchTimer,experimental,@grafana/alerting-squad,false,true,false diff --git a/pkg/services/featuremgmt/toggles_gen.go b/pkg/services/featuremgmt/toggles_gen.go index d68fa56ec8c..17498d7afb9 100644 --- a/pkg/services/featuremgmt/toggles_gen.go +++ b/pkg/services/featuremgmt/toggles_gen.go @@ -789,4 +789,8 @@ const ( // FlagProfilesExemplars // Enables profiles exemplars support in profiles drilldown FlagProfilesExemplars = "profilesExemplars" + + // FlagAlertingSyncDispatchTimer + // Use synchronized dispatch timer to minimize duplicate notifications across alertmanager HA pods + FlagAlertingSyncDispatchTimer = "alertingSyncDispatchTimer" ) diff --git a/pkg/services/featuremgmt/toggles_gen.json b/pkg/services/featuremgmt/toggles_gen.json index bfefc20f08b..071f1b0671e 100644 --- a/pkg/services/featuremgmt/toggles_gen.json +++ b/pkg/services/featuremgmt/toggles_gen.json @@ -511,6 +511,20 @@ "frontend": true } }, + { + "metadata": { + "name": "alertingSyncDispatchTimer", + "resourceVersion": "1766161788928", + "creationTimestamp": "2025-12-19T16:29:48Z" + }, + "spec": { + "description": "Use synchronized dispatch timer to minimize duplicate notifications across alertmanager HA pods", + "stage": "experimental", + "codeowner": "@grafana/alerting-squad", + "requiresRestart": true, + "hideFromDocs": true + } + }, { "metadata": { "name": "alertingTriage", @@ -662,7 +676,8 @@ "metadata": { "name": "auditLoggingAppPlatform", "resourceVersion": "1767013056996", - "creationTimestamp": "2025-12-29T12:57:36Z" + "creationTimestamp": "2025-12-29T12:57:36Z", + "deletionTimestamp": "2026-01-06T09:18:36Z" }, "spec": { "description": "Enable audit logging with Kubernetes under app platform", diff --git a/pkg/services/ngalert/ngalert.go b/pkg/services/ngalert/ngalert.go index 53e49621117..5177107a602 100644 --- a/pkg/services/ngalert/ngalert.go +++ b/pkg/services/ngalert/ngalert.go @@ -213,6 +213,9 @@ func (ng *AlertNG) init() error { SkipVerify: ng.Cfg.Smtp.SkipVerify, StaticHeaders: ng.Cfg.Smtp.StaticHeaders, } + runtimeConfig := remoteClient.RuntimeConfig{ + DispatchTimer: notifier.GetDispatchTimer(ng.FeatureToggles).String(), + } cfg := remote.AlertmanagerConfig{ BasicAuthPassword: ng.Cfg.UnifiedAlerting.RemoteAlertmanager.Password, @@ -222,6 +225,7 @@ func (ng *AlertNG) init() error { ExternalURL: ng.Cfg.AppURL, SmtpConfig: smtpCfg, Timeout: ng.Cfg.UnifiedAlerting.RemoteAlertmanager.Timeout, + RuntimeConfig: runtimeConfig, } autogenFn := func(ctx context.Context, logger log.Logger, orgID int64, cfg *definitions.PostableApiAlertingConfig, invalidReceiverAction notifier.InvalidReceiversAction) error { return notifier.AddAutogenConfig(ctx, logger, ng.store, orgID, cfg, invalidReceiverAction, ng.FeatureToggles) diff --git a/pkg/services/ngalert/notifier/alertmanager.go b/pkg/services/ngalert/notifier/alertmanager.go index f192ed88058..6d81d51dc75 100644 --- a/pkg/services/ngalert/notifier/alertmanager.go +++ b/pkg/services/ngalert/notifier/alertmanager.go @@ -33,6 +33,9 @@ const ( // How long we keep silences in the kvstore after they've expired. silenceRetention = 5 * 24 * time.Hour + + // How long we keep flushes in the kvstore after they've expired. + flushRetention = 5 * 24 * time.Hour ) type AlertingStore interface { @@ -44,8 +47,10 @@ type AlertingStore interface { type stateStore interface { SaveSilences(ctx context.Context, st alertingNotify.State) (int64, error) SaveNotificationLog(ctx context.Context, st alertingNotify.State) (int64, error) + SaveFlushLog(ctx context.Context, st alertingNotify.State) (int64, error) GetSilences(ctx context.Context) (string, error) GetNotificationLog(ctx context.Context) (string, error) + GetFlushLog(ctx context.Context) (string, error) } type alertmanager struct { @@ -101,6 +106,10 @@ func NewAlertmanager(ctx context.Context, orgID int64, cfg *setting.Cfg, store A if err != nil { return nil, err } + flushLog, err := stateStore.GetFlushLog(ctx) + if err != nil { + return nil, err + } silencesOptions := maintenanceOptions{ initialState: silences, @@ -123,12 +132,29 @@ func NewAlertmanager(ctx context.Context, orgID int64, cfg *setting.Cfg, store A } l := log.New("ngalert.notifier") + dispatchTimer := GetDispatchTimer(featureToggles) + + var flushLogOptions *maintenanceOptions + if dispatchTimer == alertingNotify.DispatchTimerSync { + flushLogOptions = &maintenanceOptions{ + initialState: flushLog, + retention: flushRetention, + maintenanceFrequency: maintenanceInterval, + maintenanceFunc: func(state alertingNotify.State) (int64, error) { + // Detached context here is to make sure that when the service is shut down the persist operation is executed. + return stateStore.SaveFlushLog(context.Background(), state) + }, + } + } + opts := alertingNotify.GrafanaAlertmanagerOpts{ ExternalURL: cfg.AppURL, AlertStoreCallback: nil, PeerTimeout: cfg.UnifiedAlerting.HAPeerTimeout, Silences: silencesOptions, Nflog: nflogOptions, + FlushLog: flushLogOptions, + DispatchTimer: dispatchTimer, Limits: alertingNotify.Limits{ MaxSilences: cfg.UnifiedAlerting.AlertmanagerMaxSilencesCount, MaxSilenceSizeBytes: cfg.UnifiedAlerting.AlertmanagerMaxSilenceSizeBytes, diff --git a/pkg/services/ngalert/notifier/dispatch_timer.go b/pkg/services/ngalert/notifier/dispatch_timer.go new file mode 100644 index 00000000000..04eaf8cb296 --- /dev/null +++ b/pkg/services/ngalert/notifier/dispatch_timer.go @@ -0,0 +1,16 @@ +package notifier + +import ( + alertingNotify "github.com/grafana/alerting/notify" + "github.com/grafana/grafana/pkg/services/featuremgmt" +) + +// GetDispatchTimer returns the appropriate dispatch timer based on feature toggles. +func GetDispatchTimer(features featuremgmt.FeatureToggles) (dt alertingNotify.DispatchTimer) { + //nolint:staticcheck // not yet migrated to OpenFeature + enabled := features.IsEnabledGlobally(featuremgmt.FlagAlertingSyncDispatchTimer) + if enabled { + dt = alertingNotify.DispatchTimerSync + } + return +} diff --git a/pkg/services/ngalert/notifier/dispatch_timer_test.go b/pkg/services/ngalert/notifier/dispatch_timer_test.go new file mode 100644 index 00000000000..3b42a562a32 --- /dev/null +++ b/pkg/services/ngalert/notifier/dispatch_timer_test.go @@ -0,0 +1,36 @@ +package notifier + +import ( + "testing" + + alertingNotify "github.com/grafana/alerting/notify" + "github.com/grafana/grafana/pkg/services/featuremgmt" + "github.com/stretchr/testify/require" +) + +func TestGetDispatchTimer(t *testing.T) { + tests := []struct { + name string + featureFlagValue bool + expected alertingNotify.DispatchTimer + }{ + { + name: "feature flag enabled returns sync timer", + featureFlagValue: true, + expected: alertingNotify.DispatchTimerSync, + }, + { + name: "feature flag disabled returns default timer", + featureFlagValue: false, + expected: alertingNotify.DispatchTimerDefault, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + features := featuremgmt.WithFeatures(featuremgmt.FlagAlertingSyncDispatchTimer, tt.featureFlagValue) + result := GetDispatchTimer(features) + require.Equal(t, tt.expected, result) + }) + } +} diff --git a/pkg/services/ngalert/notifier/file_store.go b/pkg/services/ngalert/notifier/file_store.go index e9628cb536e..3bb128d692d 100644 --- a/pkg/services/ngalert/notifier/file_store.go +++ b/pkg/services/ngalert/notifier/file_store.go @@ -15,6 +15,7 @@ const ( KVNamespace = "alertmanager" NotificationLogFilename = "notifications" SilencesFilename = "silences" + FlushLogFilename = "flushes" ) // FileStore is in charge of persisting the alertmanager files to the database. @@ -42,6 +43,10 @@ func (fileStore *FileStore) GetNotificationLog(ctx context.Context) (string, err return fileStore.contentFor(ctx, NotificationLogFilename) } +func (fileStore *FileStore) GetFlushLog(ctx context.Context) (string, error) { + return fileStore.contentFor(ctx, FlushLogFilename) +} + // contentFor returns the content for the given Alertmanager kvstore key. func (fileStore *FileStore) contentFor(ctx context.Context, filename string) (string, error) { // Then, let's attempt to read it from the database. @@ -74,6 +79,11 @@ func (fileStore *FileStore) SaveNotificationLog(ctx context.Context, st alerting return fileStore.persist(ctx, NotificationLogFilename, st) } +// SaveFlushLog saves the flush log to the database and returns the size of the unencoded state. +func (fileStore *FileStore) SaveFlushLog(ctx context.Context, st alertingNotify.State) (int64, error) { + return fileStore.persist(ctx, FlushLogFilename, st) +} + // persist takes care of persisting the binary representation of internal state to the database as a base64 encoded string. func (fileStore *FileStore) persist(ctx context.Context, filename string, st alertingNotify.State) (int64, error) { var size int64 diff --git a/pkg/services/ngalert/notifier/file_store_test.go b/pkg/services/ngalert/notifier/file_store_test.go index 1952eb5a0f1..7d4de602a0f 100644 --- a/pkg/services/ngalert/notifier/file_store_test.go +++ b/pkg/services/ngalert/notifier/file_store_test.go @@ -106,3 +106,48 @@ func TestFileStore_NotificationLog(t *testing.T) { t.Errorf("Unexpected Diff: %v", cmp.Diff(newState, decoded)) } } + +func TestFileStore_FlushLog(t *testing.T) { + store := fakes.NewFakeKVStore(t) + ctx := context.Background() + var orgId int64 = 1 + + // Initialize kvstore with empty flush log state. + initialState := flushLogState{} // FlushLog uses the same structure as nflog + decodedState, err := initialState.MarshalBinary() + require.NoError(t, err) + encodedState := base64.StdEncoding.EncodeToString(decodedState) + err = store.Set(ctx, orgId, KVNamespace, FlushLogFilename, encodedState) + require.NoError(t, err) + + fs := NewFileStore(orgId, store) + + // Load initial (empty). + flushLog, err := fs.GetFlushLog(ctx) + require.NoError(t, err) + decoded, err := decodeFlushLogState(strings.NewReader(flushLog)) + require.NoError(t, err) + if !cmp.Equal(initialState, decoded) { + t.Errorf("Unexpected Diff: %v", cmp.Diff(initialState, decoded)) + } + + // Save new flush log state. + now := time.Now() + oneHour := now.Add(time.Hour) + + v1 := createFlushLog(1, now, oneHour) + v2 := createFlushLog(2, now, oneHour) + newState := flushLogState{1: v1, 2: v2} + size, err := fs.SaveFlushLog(ctx, newState) + require.NoError(t, err) + require.Greater(t, size, int64(0)) + + // Load new. + flushLog, err = fs.GetFlushLog(ctx) + require.NoError(t, err) + decoded, err = decodeFlushLogState(strings.NewReader(flushLog)) + require.NoError(t, err) + if !cmp.Equal(newState, decoded) { + t.Errorf("Unexpected Diff: %v", cmp.Diff(newState, decoded)) + } +} diff --git a/pkg/services/ngalert/notifier/multiorg_alertmanager.go b/pkg/services/ngalert/notifier/multiorg_alertmanager.go index 4aa0151d18f..a10aee29ef6 100644 --- a/pkg/services/ngalert/notifier/multiorg_alertmanager.go +++ b/pkg/services/ngalert/notifier/multiorg_alertmanager.go @@ -82,6 +82,7 @@ type Alertmanager interface { type ExternalState struct { Silences []byte Nflog []byte + FlushLog []byte } // StateMerger describes a type that is able to merge external state (nflog, silences) with its own. @@ -378,7 +379,7 @@ func (moa *MultiOrgAlertmanager) SyncAlertmanagersForOrgs(ctx context.Context, o func (moa *MultiOrgAlertmanager) cleanupOrphanLocalOrgState(ctx context.Context, activeOrganizations map[int64]struct{}, ) { - storedFiles := []string{NotificationLogFilename, SilencesFilename} + storedFiles := []string{NotificationLogFilename, SilencesFilename, FlushLogFilename} for _, fileName := range storedFiles { keys, err := moa.kvStore.Keys(ctx, kvstore.AllOrganizations, KVNamespace, fileName) if err != nil { diff --git a/pkg/services/ngalert/notifier/state.go b/pkg/services/ngalert/notifier/state.go index c8551d2ed1a..04ba9d9a31d 100644 --- a/pkg/services/ngalert/notifier/state.go +++ b/pkg/services/ngalert/notifier/state.go @@ -5,5 +5,8 @@ func (am *alertmanager) MergeState(state ExternalState) error { if err := am.Base.MergeNflog(state.Nflog); err != nil { return err } - return am.Base.MergeSilences(state.Silences) + if err := am.Base.MergeSilences(state.Silences); err != nil { + return err + } + return am.Base.MergeFlushLog(state.FlushLog) } diff --git a/pkg/services/ngalert/notifier/testing.go b/pkg/services/ngalert/notifier/testing.go index 9fccf6d2f0d..c2b2190183f 100644 --- a/pkg/services/ngalert/notifier/testing.go +++ b/pkg/services/ngalert/notifier/testing.go @@ -11,6 +11,7 @@ import ( "time" "github.com/matttproud/golang_protobuf_extensions/pbutil" + "github.com/prometheus/alertmanager/flushlog/flushlogpb" "github.com/prometheus/alertmanager/nflog/nflogpb" "github.com/prometheus/alertmanager/silence/silencepb" "github.com/prometheus/common/model" @@ -228,15 +229,13 @@ func (f *FakeOrgStore) FetchOrgIds(_ context.Context) ([]int64, error) { return f.orgs, nil } -type NoValidation struct { -} +type NoValidation struct{} func (n NoValidation) Validate(_ models.NotificationSettings) error { return nil } -type RejectingValidation struct { -} +type RejectingValidation struct{} func (n RejectingValidation) Validate(s models.NotificationSettings) error { return ErrorReceiverDoesNotExist{ErrorReferenceInvalid: ErrorReferenceInvalid{Reference: s.Receiver}} @@ -365,6 +364,51 @@ func createNotificationLog(groupKey string, receiverName string, sentAt, expires } } +// https://github.com/grafana/prometheus-alertmanager/blob/main/flushlog/flushlog.go#L136-L136 +type flushLogState map[uint64]*flushlogpb.MeshFlushLog + +func (s flushLogState) MarshalBinary() ([]byte, error) { + var buf bytes.Buffer + + for _, e := range s { + if _, err := pbutil.WriteDelimited(&buf, e); err != nil { + return nil, err + } + } + return buf.Bytes(), nil +} + +func createFlushLog(groupFingerprint uint64, ts, expiresAt time.Time) *flushlogpb.MeshFlushLog { + return &flushlogpb.MeshFlushLog{ + FlushLog: &flushlogpb.FlushLog{ + GroupFingerprint: groupFingerprint, + Timestamp: ts, + }, + ExpiresAt: expiresAt, + } +} + +// decodeFlushLogState copied from decodeState in prometheus-alertmanager/flushlog/flushlog.go +func decodeFlushLogState(r io.Reader) (flushLogState, error) { + st := flushLogState{} + for { + var e flushlogpb.MeshFlushLog + _, err := pbutil.ReadDelimited(r, &e) + if err == nil { + if e.FlushLog == nil || e.FlushLog.GroupFingerprint == 0 || e.FlushLog.Timestamp.IsZero() { + return nil, errInvalidState + } + st[e.FlushLog.GroupFingerprint] = &e + continue + } + if errors.Is(err, io.EOF) { + break + } + return nil, err + } + return st, nil +} + type call struct { Method string Args []interface{} diff --git a/pkg/services/ngalert/remote/alertmanager.go b/pkg/services/ngalert/remote/alertmanager.go index 60740d935af..07fa5e3138f 100644 --- a/pkg/services/ngalert/remote/alertmanager.go +++ b/pkg/services/ngalert/remote/alertmanager.go @@ -47,6 +47,7 @@ import ( type stateStore interface { GetSilences(ctx context.Context) (string, error) GetNotificationLog(ctx context.Context) (string, error) + GetFlushLog(ctx context.Context) (string, error) } // AutogenFn is a function that adds auto-generated routes to a configuration. @@ -86,6 +87,8 @@ type Alertmanager struct { promoteConfig bool externalURL string + + runtimeConfig remoteClient.RuntimeConfig } type AlertmanagerConfig struct { @@ -111,6 +114,9 @@ type AlertmanagerConfig struct { // Timeout for the HTTP client. Timeout time.Duration + + // RuntimeConfig specifies runtime behavior settings for the remote Alertmanager. + RuntimeConfig remoteClient.RuntimeConfig } func (cfg *AlertmanagerConfig) Validate() error { @@ -203,6 +209,7 @@ func NewAlertmanager(ctx context.Context, cfg AlertmanagerConfig, store stateSto externalURL: cfg.ExternalURL, promoteConfig: cfg.PromoteConfig, smtp: cfg.SmtpConfig, + runtimeConfig: cfg.RuntimeConfig, } // Parse the default configuration once and remember its hash so we can compare it later. @@ -331,10 +338,11 @@ func (am *Alertmanager) buildConfiguration(ctx context.Context, raw []byte, crea AlertmanagerConfig: mergeResult.Config, Templates: templates, }, - CreatedAt: createdAtEpoch, - Promoted: am.promoteConfig, - ExternalURL: am.externalURL, - SmtpConfig: am.smtp, + CreatedAt: createdAtEpoch, + Promoted: am.promoteConfig, + ExternalURL: am.externalURL, + SmtpConfig: am.smtp, + RuntimeConfig: am.runtimeConfig, } cfgHash, err := calculateUserGrafanaConfigHash(payload) @@ -388,6 +396,8 @@ func (am *Alertmanager) GetRemoteState(ctx context.Context) (notifier.ExternalSt rs.Silences = p.Data case "nfl": rs.Nflog = p.Data + case "fls": + rs.FlushLog = p.Data default: return rs, fmt.Errorf("unknown part key %q", p.Key) } @@ -677,6 +687,12 @@ func (am *Alertmanager) getFullState(ctx context.Context) (string, error) { } parts = append(parts, alertingClusterPB.Part{Key: notifier.NotificationLogFilename, Data: []byte(notificationLog)}) + flushLog, err := am.state.GetFlushLog(ctx) + if err != nil { + return "", fmt.Errorf("error getting flush log: %w", err) + } + parts = append(parts, alertingClusterPB.Part{Key: notifier.FlushLogFilename, Data: []byte(flushLog)}) + fs := alertingClusterPB.FullState{ Parts: parts, } diff --git a/pkg/services/ngalert/remote/client/alertmanager_configuration.go b/pkg/services/ngalert/remote/client/alertmanager_configuration.go index a53132a8812..75246a6f32d 100644 --- a/pkg/services/ngalert/remote/client/alertmanager_configuration.go +++ b/pkg/services/ngalert/remote/client/alertmanager_configuration.go @@ -29,6 +29,10 @@ func (u *GrafanaAlertmanagerConfig) MarshalJSON() ([]byte, error) { return definition.MarshalJSONWithSecrets((*cfg)(u)) } +type RuntimeConfig struct { + DispatchTimer string `json:"dispatch_timer"` +} + type UserGrafanaConfig struct { GrafanaAlertmanagerConfig GrafanaAlertmanagerConfig `json:"configuration"` Hash string `json:"configuration_hash"` @@ -37,6 +41,7 @@ type UserGrafanaConfig struct { Promoted bool `json:"promoted"` ExternalURL string `json:"external_url"` SmtpConfig SmtpConfig `json:"smtp_config"` + RuntimeConfig RuntimeConfig `json:"runtime_config"` } func (mc *Mimir) GetGrafanaAlertmanagerConfig(ctx context.Context) (*UserGrafanaConfig, error) {