Remote Alertmanager: Optionally merge remote state before starting the internal Alertmanager (#107710)

* Remote Alertmanager: Use the same struct for Grafana stat and Mimir full state

* Alertmanager: Add methods to merge nflog and silences

* update grafana/alerting version

* make update-workspace

* update mocks

* remove unnecesary methods from the remote Alertmanager implementation, create separate StateMerger interface

* (WIP) Remote Alertmanager: Optionally merge remote state before starting the internal Alertmanager

* cleanup ngalert.go

* restore defaults.ini

* move state parsing logic to 'remote' package, clean up ngalert.go

* remove GetBase, implement MegeNflog and MergeSilences

* delete fmt.Println

* FetchRemoteState -> GetRemoteState

* UserGrafanaState -> UserState

* remove duplicate clusterpb import

* reorder MimirClient interface

* use general getState() method for Grafana state and Mimir full state

* remove unnecessary state merging methods from the Alertmanager interface

* remove pullState field

* reduce diff

* add info log after merging

* merge silences and nflog entries in the same method

* merge the remote state in the forked AM

* reduce diff

* update remote AM mock

* tests

* make error more specific

* typo

packages/grafana-data/src/types/featureToggles.gen.ts

@@ -1074,4 +1074,9 @@ export interface FeatureToggles {
   * Enables adhoc filtering support for the dashboard datasource
   */
   dashboardDsAdHocFiltering?: boolean;
+  /**
+  * Starts Grafana in remote secondary mode pulling the latest state from the remote Alertmanager to avoid duplicate notifications.
+  * @default false
+  */
+  alertmanagerRemoteSecondaryWithRemoteState?: boolean;
 }

pkg/services/featuremgmt/registry.go

@@ -1859,6 +1859,15 @@ var (
 			Owner:        grafanaDataProSquad,
 			FrontendOnly: true,
 		},
+		{
+			Name:              "alertmanagerRemoteSecondaryWithRemoteState",
+			Description:       "Starts Grafana in remote secondary mode pulling the latest state from the remote Alertmanager to avoid duplicate notifications.",
+			Stage:             FeatureStageExperimental,
+			Owner:             grafanaAlertingSquad,
+			HideFromAdminPage: true,
+			HideFromDocs:      true,
+			Expression:        "false",
+		},
 	}
 )
 

pkg/services/featuremgmt/toggles_gen.csv

@@ -240,3 +240,4 @@ alertingNotificationHistory,experimental,@grafana/alerting-squad,false,false,fal
 pluginAssetProvider,experimental,@grafana/plugins-platform-backend,false,true,false
 unifiedStorageSearchDualReaderEnabled,experimental,@grafana/search-and-storage,false,false,false
 dashboardDsAdHocFiltering,experimental,@grafana/datapro,false,false,true
+alertmanagerRemoteSecondaryWithRemoteState,experimental,@grafana/alerting-squad,false,false,false

pkg/services/featuremgmt/toggles_gen.go

@@ -970,4 +970,8 @@ const (
 	// FlagDashboardDsAdHocFiltering
 	// Enables adhoc filtering support for the dashboard datasource
 	FlagDashboardDsAdHocFiltering = "dashboardDsAdHocFiltering"
+
+	// FlagAlertmanagerRemoteSecondaryWithRemoteState
+	// Starts Grafana in remote secondary mode pulling the latest state from the remote Alertmanager to avoid duplicate notifications.
+	FlagAlertmanagerRemoteSecondaryWithRemoteState = "alertmanagerRemoteSecondaryWithRemoteState"
 )

pkg/services/ngalert/ngalert.go

@@ -193,7 +193,8 @@ func (ng *AlertNG) init() error {
 	crypto := notifier.NewCrypto(ng.SecretsService, ng.store, moaLogger)
 	remotePrimary := ng.FeatureToggles.IsEnabled(initCtx, featuremgmt.FlagAlertmanagerRemotePrimary)
 	remoteSecondary := ng.FeatureToggles.IsEnabled(initCtx, featuremgmt.FlagAlertmanagerRemoteSecondary)
-	if remotePrimary || remoteSecondary {
+	remoteSecondaryWithRemoteState := ng.FeatureToggles.IsEnabled(initCtx, featuremgmt.FlagAlertmanagerRemoteSecondaryWithRemoteState)
+	if remotePrimary || remoteSecondary || remoteSecondaryWithRemoteState {
 		m := ng.Metrics.GetRemoteAlertmanagerMetrics()
 		smtpCfg := remoteClient.SmtpConfig{
 			FromAddress:    ng.Cfg.Smtp.FromAddress,
@@ -268,6 +269,10 @@ func (ng *AlertNG) init() error {
 					cfg.OrgID = orgID
 					remoteAM, err := createRemoteAlertmanager(ctx, cfg, ng.KVStore, crypto, autogenFn, m, ng.tracer)
 					if err != nil {
+						if remoteSecondaryWithRemoteState {
+							// We can't start the internal Alertmanager without the remote state.
+							return nil, fmt.Errorf("failed to create remote Alertmanager, can't start the internal Alertmanager without the remote state: %w", err)
+						}
 						moaLogger.Error("Failed to create remote Alertmanager, falling back to using only the internal one", "err", err)
 						return internalAM, nil
 					}

pkg/services/ngalert/remote/forked_alertmanager_test.go

@@ -84,6 +84,75 @@ func TestForkedAlertmanager_ModeRemoteSecondary(t *testing.T) {
 		}
 	})
 
+	t.Run("ApplyConfig - with remote state", func(tt *testing.T) {
+		{
+			// During the first ApplyConfig call, we should:
+			// 1. Apply the configuration to the remote Alertmanager
+			// 2. Merge the remote state
+			// 3. Apply the configuration to the internal Alertmanager
+			internal, remote, forked := genTestAlertmanagers(tt, modeRemoteSecondary, withRemoteState)
+			readyCall := remote.EXPECT().Ready().Return(false).Once()
+			remote.EXPECT().ApplyConfig(ctx, mock.Anything).Return(nil).Once().NotBefore(readyCall)
+			remote.EXPECT().Ready().Return(true).Once()
+			remoteStateCall := remote.EXPECT().GetRemoteState(mock.Anything).Return(notifier.ExternalState{}, nil).Once()
+			internal.EXPECT().ApplyConfig(ctx, mock.Anything).Return(nil).Once().NotBefore(remoteStateCall)
+			require.NoError(tt, forked.ApplyConfig(ctx, &models.AlertConfiguration{}))
+			require.True(tt, internal.mergeStateCalled)
+
+			// We shouldn't attempt to merge the remote state again on the next sync loop iteration.
+			internal.mergeStateCalled = false
+			remote.EXPECT().Ready().Return(true).Once()
+			internal.EXPECT().ApplyConfig(ctx, mock.Anything).Return(nil).Once()
+			remote.EXPECT().CompareAndSendConfiguration(ctx, mock.Anything).Return(nil).Once()
+			require.NoError(tt, forked.ApplyConfig(ctx, &models.AlertConfiguration{}))
+			require.False(tt, internal.mergeStateCalled)
+		}
+
+		{
+			// If we fail to apply the configuration in the remote Alertmanager, we should get an error and not start the internal Alertmanager.
+			internal, remote, forked := genTestAlertmanagers(tt, modeRemoteSecondary, withSyncInterval(10*time.Minute), withRemoteState)
+			readyCall := remote.EXPECT().Ready().Return(false).Once()
+			remote.EXPECT().ApplyConfig(ctx, mock.Anything).Return(expErr).Once().NotBefore(readyCall)
+			remote.EXPECT().Ready().Return(false).Once()
+			err := forked.ApplyConfig(ctx, &models.AlertConfiguration{})
+			require.Equal(tt, "remote Alertmanager not ready, can't fetch remote state", err.Error())
+			require.False(tt, internal.mergeStateCalled)
+
+			// Calling ApplyConfig again should result in the forked Alertmanager calling ApplyConfig on both
+			// Alertmanagers and merging the remote state, even if the sync interval has not elapsed.
+			remote.EXPECT().Ready().Return(true).Twice()
+			remote.EXPECT().CompareAndSendConfiguration(ctx, mock.Anything).Return(nil).Once()
+			remoteStateCall := remote.EXPECT().GetRemoteState(mock.Anything).Return(notifier.ExternalState{}, nil).Once()
+			internal.EXPECT().ApplyConfig(ctx, mock.Anything).Return(nil).Once().NotBefore(remoteStateCall)
+			require.NoError(tt, forked.ApplyConfig(ctx, &models.AlertConfiguration{}))
+			require.True(tt, internal.mergeStateCalled)
+		}
+
+		{
+			// An error in the remote Alertmanager should be returned.
+			// The internal Alertmanager shouldn't be started.
+			internal, remote, forked := genTestAlertmanagers(tt, modeRemotePrimary)
+			remote.EXPECT().ApplyConfig(ctx, mock.Anything).Return(expErr).Once()
+			require.ErrorIs(tt, forked.ApplyConfig(ctx, &models.AlertConfiguration{}), expErr)
+			require.False(t, internal.mergeStateCalled)
+		}
+
+		{
+			// An error in the internal Alertmanager should be returned.
+			internal, remote, forked := genTestAlertmanagers(tt, modeRemoteSecondary, withRemoteState)
+			internal.EXPECT().ApplyConfig(ctx, mock.Anything).Return(expErr).Once()
+
+			// Simulate starting the remote Alertmanager and merging the remote state.
+			readyCall := remote.EXPECT().Ready().Return(false).Once()
+			remote.EXPECT().ApplyConfig(ctx, mock.Anything).Return(nil).Once().NotBefore(readyCall)
+			remote.EXPECT().Ready().Return(true).Once()
+			remote.EXPECT().GetRemoteState(mock.Anything).Return(notifier.ExternalState{}, nil).Once()
+
+			require.ErrorIs(tt, forked.ApplyConfig(ctx, &models.AlertConfiguration{}), expErr)
+			require.True(t, internal.mergeStateCalled)
+		}
+	})
+
 	t.Run("SaveAndApplyConfig", func(tt *testing.T) {
 		// SaveAndApplyConfig should only be called on the remote Alertmanager.
 		// State and configuration are updated on an interval.
@@ -707,6 +776,11 @@ func (m *internalAlertmanagerMock) MergeState(notifier.ExternalState) error {
 	return nil
 }
 
+func withRemoteState(rsc RemoteSecondaryConfig) RemoteSecondaryConfig {
+	rsc.WithRemoteState = true
+	return rsc
+}
+
 func withSyncInterval(syncInterval time.Duration) func(RemoteSecondaryConfig) RemoteSecondaryConfig {
 	return func(rsc RemoteSecondaryConfig) RemoteSecondaryConfig {
 		rsc.SyncInterval = syncInterval

pkg/services/ngalert/remote/mock/remoteAlertmanager.go

@@ -13,6 +13,8 @@ import (
 
 	models "github.com/grafana/grafana/pkg/services/ngalert/models"
 
+	notifier "github.com/grafana/grafana/pkg/services/ngalert/notifier"
+
 	notify "github.com/grafana/alerting/notify"
 
 	v2models "github.com/prometheus/alertmanager/api/v2/models"
@@ -413,6 +415,62 @@ func (_c *RemoteAlertmanagerMock_GetReceivers_Call) RunAndReturn(run func(contex
 	return _c
 }
 
+// GetRemoteState provides a mock function with given fields: _a0
+func (_m *RemoteAlertmanagerMock) GetRemoteState(_a0 context.Context) (notifier.ExternalState, error) {
+	ret := _m.Called(_a0)
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetRemoteState")
+	}
+
+	var r0 notifier.ExternalState
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context) (notifier.ExternalState, error)); ok {
+		return rf(_a0)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context) notifier.ExternalState); ok {
+		r0 = rf(_a0)
+	} else {
+		r0 = ret.Get(0).(notifier.ExternalState)
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context) error); ok {
+		r1 = rf(_a0)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// RemoteAlertmanagerMock_GetRemoteState_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetRemoteState'
+type RemoteAlertmanagerMock_GetRemoteState_Call struct {
+	*mock.Call
+}
+
+// GetRemoteState is a helper method to define mock.On call
+//   - _a0 context.Context
+func (_e *RemoteAlertmanagerMock_Expecter) GetRemoteState(_a0 interface{}) *RemoteAlertmanagerMock_GetRemoteState_Call {
+	return &RemoteAlertmanagerMock_GetRemoteState_Call{Call: _e.mock.On("GetRemoteState", _a0)}
+}
+
+func (_c *RemoteAlertmanagerMock_GetRemoteState_Call) Run(run func(_a0 context.Context)) *RemoteAlertmanagerMock_GetRemoteState_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context))
+	})
+	return _c
+}
+
+func (_c *RemoteAlertmanagerMock_GetRemoteState_Call) Return(_a0 notifier.ExternalState, _a1 error) *RemoteAlertmanagerMock_GetRemoteState_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *RemoteAlertmanagerMock_GetRemoteState_Call) RunAndReturn(run func(context.Context) (notifier.ExternalState, error)) *RemoteAlertmanagerMock_GetRemoteState_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // GetSilence provides a mock function with given fields: _a0, _a1
 func (_m *RemoteAlertmanagerMock) GetSilence(_a0 context.Context, _a1 string) (v2models.GettableSilence, error) {
 	ret := _m.Called(_a0, _a1)

pkg/services/ngalert/remote/remote_secondary_forked_alertmanager.go

@@ -22,6 +22,7 @@ type configStore interface {
 type remoteAlertmanager interface {
 	notifier.Alertmanager
 	CompareAndSendConfiguration(context.Context, *models.AlertConfiguration) error
+	GetRemoteState(context.Context) (notifier.ExternalState, error)
 	SendState(context.Context) error
 }
 
@@ -35,6 +36,8 @@ type RemoteSecondaryForkedAlertmanager struct {
 
 	lastSync     time.Time
 	syncInterval time.Duration
+
+	shouldFetchRemoteState bool
 }
 
 type RemoteSecondaryConfig struct {
@@ -45,6 +48,9 @@ type RemoteSecondaryConfig struct {
 	// SyncInterval determines how often we should attempt to synchronize
 	// the configuration on the remote Alertmanager.
 	SyncInterval time.Duration
+
+	// WithRemoteState is used to fetch and merge the state from the remote Alertmanager before starting the internal one.
+	WithRemoteState bool
 }
 
 func (c *RemoteSecondaryConfig) Validate() error {
@@ -59,12 +65,13 @@ func NewRemoteSecondaryForkedAlertmanager(cfg RemoteSecondaryConfig, internal no
 		return nil, err
 	}
 	return &RemoteSecondaryForkedAlertmanager{
-		log:          cfg.Logger,
-		orgID:        cfg.OrgID,
-		store:        cfg.Store,
-		internal:     internal,
-		remote:       remote,
-		syncInterval: cfg.SyncInterval,
+		log:                    cfg.Logger,
+		orgID:                  cfg.OrgID,
+		store:                  cfg.Store,
+		internal:               internal,
+		remote:                 remote,
+		syncInterval:           cfg.SyncInterval,
+		shouldFetchRemoteState: cfg.WithRemoteState,
 	}, nil
 }
 
@@ -99,6 +106,28 @@ func (fam *RemoteSecondaryForkedAlertmanager) ApplyConfig(ctx context.Context, c
 		}
 	}()
 
+	if fam.shouldFetchRemoteState {
+		wg.Wait()
+		if !fam.remote.Ready() {
+			return fmt.Errorf("remote Alertmanager not ready, can't fetch remote state")
+		}
+		// Pull and merge the remote Alertmanager state.
+		rs, err := fam.remote.GetRemoteState(ctx)
+		if err != nil {
+			return fmt.Errorf("failed to fetch remote state: %w", err)
+		}
+
+		// The internal Alertmanager should implement the StateMerger interface.
+		sm := fam.internal.(notifier.StateMerger)
+		if err := sm.MergeState(rs); err != nil {
+			return fmt.Errorf("failed to merge remote state: %w", err)
+		}
+		fam.log.Info("Successfully merged remote silences and nflog entries")
+
+		// This operation should only be performed at startup.
+		fam.shouldFetchRemoteState = false
+	}
+
 	// Call ApplyConfig on the internal Alertmanager - we only care about errors for this one.
 	err := fam.internal.ApplyConfig(ctx, config)
 	wg.Wait()