Alerting: add sync timer support (#114602)

- add new feature flag to support enabling the dispatcher sync timer on the alertmanager
- this attempts to synchronize the flushes across HA nodes to decrease amount of duplicate notifications

---------

Co-authored-by: Yuri Tseretyan <yuriy.tseretyan@grafana.com>

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;
 }

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,
+		},
 	}
 )
 

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

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"
 )

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",

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)

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,

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
+}

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)
+		})
+	}
+}

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

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))
+	}
+}

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 {

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)
 }

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{}

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,
 	}

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) {