Alerting: Support Unified Alerting with Grafana HA (#37920) (#39342)

* Alerting: Support Unified Alerting in Grafana's HA mode.

(cherry picked from commit 7db97097c9)

Co-authored-by: gotjosh <josue@grafana.com>

pkg/services/ngalert/metrics/ngalert.go

@@ -52,6 +52,7 @@ type Scheduler struct {
 }
 
 type MultiOrgAlertmanager struct {
+	Registerer               prometheus.Registerer
 	ActiveConfigurations     prometheus.Gauge
 	DiscoveredConfigurations prometheus.Gauge
 	registries               *OrgRegistries
@@ -178,6 +179,7 @@ func newStateMetrics(r prometheus.Registerer) *State {
 
 func newMultiOrgAlertmanagerMetrics(r prometheus.Registerer) *MultiOrgAlertmanager {
 	return &MultiOrgAlertmanager{
+		Registerer: r,
 		registries: NewOrgRegistries(),
 		DiscoveredConfigurations: promauto.With(r).NewGauge(prometheus.GaugeOpts{
 			Namespace: Namespace,

pkg/services/ngalert/ngalert.go

@@ -84,6 +84,8 @@ type AlertNG struct {
 }
 
 func (ng *AlertNG) init() error {
+	var err error
+
 	baseInterval := ng.Cfg.AlertingBaseInterval
 	if baseInterval <= 0 {
 		baseInterval = defaultBaseIntervalSeconds
@@ -97,7 +99,11 @@ func (ng *AlertNG) init() error {
 		Logger:                 ng.Log,
 	}
 
-	ng.MultiOrgAlertmanager = notifier.NewMultiOrgAlertmanager(ng.Cfg, store, store, ng.KVStore, ng.Metrics.GetMultiOrgAlertmanagerMetrics())
+	multiOrgMetrics := ng.Metrics.GetMultiOrgAlertmanagerMetrics()
+	ng.MultiOrgAlertmanager, err = notifier.NewMultiOrgAlertmanager(ng.Cfg, store, store, ng.KVStore, multiOrgMetrics, log.New("ngalert.multiorg.alertmanager"))
+	if err != nil {
+		return err
+	}
 
 	// Let's make sure we're able to complete an initial sync of Alertmanagers before we start the alerting components.
 	if err := ng.MultiOrgAlertmanager.LoadAndSyncAlertmanagersForOrgs(context.Background()); err != nil {

pkg/services/ngalert/notifier/alertmanager.go

@@ -15,6 +15,7 @@ import (
 
 	gokit_log "github.com/go-kit/kit/log"
 	amv2 "github.com/prometheus/alertmanager/api/v2/models"
+	"github.com/prometheus/alertmanager/cluster"
 	"github.com/prometheus/alertmanager/dispatch"
 	"github.com/prometheus/alertmanager/inhibit"
 	"github.com/prometheus/alertmanager/nflog"
@@ -24,6 +25,7 @@ import (
 	"github.com/prometheus/alertmanager/silence"
 	"github.com/prometheus/alertmanager/template"
 	"github.com/prometheus/alertmanager/types"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
 
 	"github.com/grafana/grafana/pkg/components/securejsondata"
@@ -77,9 +79,16 @@ const (
 `
 )
 
+type ClusterPeer interface {
+	AddState(string, cluster.State, prometheus.Registerer) cluster.ClusterChannel
+	Position() int
+	WaitReady(context.Context) error
+}
+
 type Alertmanager struct {
 	logger      log.Logger
 	gokitLogger gokit_log.Logger
+	OrgID       int64
 
 	Settings  *setting.Cfg
 	Store     store.AlertingStore
@@ -90,6 +99,8 @@ type Alertmanager struct {
 	marker          types.Marker
 	alerts          *mem.Alerts
 	route           *dispatch.Route
+	peer            ClusterPeer
+	peerTimeout     time.Duration
 
 	dispatcher *dispatch.Dispatcher
 	inhibitor  *inhibit.Inhibitor
@@ -111,7 +122,7 @@ type Alertmanager struct {
 	orgID           int64
 }
 
-func newAlertmanager(orgID int64, cfg *setting.Cfg, store store.AlertingStore, kvStore kvstore.KVStore, m *metrics.Alertmanager) (*Alertmanager, error) {
+func newAlertmanager(orgID int64, cfg *setting.Cfg, store store.AlertingStore, kvStore kvstore.KVStore, peer ClusterPeer, m *metrics.Alertmanager) (*Alertmanager, error) {
 	am := &Alertmanager{
 		Settings:          cfg,
 		stopc:             make(chan struct{}),
@@ -120,6 +131,8 @@ func newAlertmanager(orgID int64, cfg *setting.Cfg, store store.AlertingStore, k
 		stageMetrics:      notify.NewMetrics(m.Registerer),
 		dispatcherMetrics: dispatch.NewDispatcherMetrics(false, m.Registerer),
 		Store:             store,
+		peer:              peer,
+		peerTimeout:       cfg.HAPeerTimeout,
 		Metrics:           m,
 		orgID:             orgID,
 	}
@@ -148,6 +161,9 @@ func newAlertmanager(orgID int64, cfg *setting.Cfg, store store.AlertingStore, k
 	if err != nil {
 		return nil, fmt.Errorf("unable to initialize the notification log component of alerting: %w", err)
 	}
+	c := am.peer.AddState(fmt.Sprintf("notificationlog:%d", am.OrgID), am.notificationLog, m.Registerer)
+	am.notificationLog.SetBroadcast(c.Broadcast)
+
 	// Initialize silences
 	am.silences, err = silence.New(silence.Options{
 		Metrics:      m.Registerer,
@@ -158,6 +174,9 @@ func newAlertmanager(orgID int64, cfg *setting.Cfg, store store.AlertingStore, k
 		return nil, fmt.Errorf("unable to initialize the silencing component of alerting: %w", err)
 	}
 
+	c = am.peer.AddState(fmt.Sprintf("silences:%d", am.OrgID), am.silences, m.Registerer)
+	am.silences.SetBroadcast(c.Broadcast)
+
 	am.wg.Add(1)
 	go func() {
 		am.silences.Maintenance(15*time.Minute, silencesFilePath, am.stopc, func() (int64, error) {
@@ -392,15 +411,16 @@ func (am *Alertmanager) applyConfig(cfg *apimodels.PostableUserConfig, rawConfig
 	am.inhibitor = inhibit.NewInhibitor(am.alerts, cfg.AlertmanagerConfig.InhibitRules, am.marker, am.gokitLogger)
 	am.silencer = silence.NewSilencer(am.silences, am.marker, am.gokitLogger)
 
+	meshStage := notify.NewGossipSettleStage(am.peer)
 	inhibitionStage := notify.NewMuteStage(am.inhibitor)
 	silencingStage := notify.NewMuteStage(am.silencer)
 	for name := range integrationsMap {
-		stage := am.createReceiverStage(name, integrationsMap[name], waitFunc, am.notificationLog)
-		routingStage[name] = notify.MultiStage{silencingStage, inhibitionStage, stage}
+		stage := am.createReceiverStage(name, integrationsMap[name], am.waitFunc, am.notificationLog)
+		routingStage[name] = notify.MultiStage{meshStage, silencingStage, inhibitionStage, stage}
 	}
 
 	am.route = dispatch.NewRoute(cfg.AlertmanagerConfig.Route, nil)
-	am.dispatcher = dispatch.NewDispatcher(am.alerts, am.route, routingStage, am.marker, timeoutFunc, &nilLimits{}, am.gokitLogger, am.dispatcherMetrics)
+	am.dispatcher = dispatch.NewDispatcher(am.alerts, am.route, routingStage, am.marker, am.timeoutFunc, &nilLimits{}, am.gokitLogger, am.dispatcherMetrics)
 
 	am.wg.Add(1)
 	go func() {
@@ -701,21 +721,17 @@ func (am *Alertmanager) createReceiverStage(name string, integrations []notify.I
 	return fs
 }
 
-func waitFunc() time.Duration {
-	// When it's a single instance, we don't need additional wait. The routing policies will have their own group wait.
-	// We need >0 wait here in case we have peers to sync the notification state with. 0 wait in that case can result
-	// in duplicate notifications being sent.
-	// TODO: we have setting.AlertingNotificationTimeout in legacy settings. Either use that or separate set of config
-	// for clustering with intuitive name, like "PeerTimeout".
-	return 0
+func (am *Alertmanager) waitFunc() time.Duration {
+	return time.Duration(am.peer.Position()) * am.peerTimeout
 }
 
-func timeoutFunc(d time.Duration) time.Duration {
-	//TODO: What does MinTimeout means here?
+func (am *Alertmanager) timeoutFunc(d time.Duration) time.Duration {
+	// time.Duration d relates to the receiver's group_interval. Even with a group interval of 1s,
+	// we need to make sure (non-position-0) peers in the cluster wait before flushing the notifications.
 	if d < notify.MinTimeout {
 		d = notify.MinTimeout
 	}
-	return d + waitFunc()
+	return d + am.waitFunc()
 }
 
 type nilLimits struct{}

pkg/services/ngalert/notifier/alertmanager_test.go

@@ -48,7 +48,7 @@ func setupAMTest(t *testing.T) *Alertmanager {
 	}
 
 	kvStore := newFakeKVStore(t)
-	am, err := newAlertmanager(1, cfg, s, kvStore, m)
+	am, err := newAlertmanager(1, cfg, s, kvStore, &NilPeer{}, m)
 	require.NoError(t, err)
 	return am
 }

pkg/services/ngalert/notifier/multiorg_alertmanager.go

@@ -6,6 +6,12 @@ import (
 	"sync"
 	"time"
 
+	"github.com/grafana/grafana/pkg/services/ngalert/logging"
+
+	gokit_log "github.com/go-kit/kit/log"
+	"github.com/prometheus/alertmanager/cluster"
+	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/grafana/grafana/pkg/infra/kvstore"
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/services/ngalert/metrics"
@@ -14,7 +20,6 @@ import (
 )
 
 var (
-	SyncOrgsPollInterval    = 1 * time.Minute
 	ErrNoAlertmanagerForOrg = fmt.Errorf("Alertmanager does not exist for this organization")
 	ErrAlertmanagerNotReady = fmt.Errorf("Alertmanager is not ready yet")
 )
@@ -26,6 +31,10 @@ type MultiOrgAlertmanager struct {
 	settings *setting.Cfg
 	logger   log.Logger
 
+	// clusterPeer represents the clustering peers of Alertmanagers between Grafana instances.
+	peer         ClusterPeer
+	settleCancel context.CancelFunc
+
 	configStore store.AlertingStore
 	orgStore    store.OrgStore
 	kvStore     kvstore.KVStore
@@ -33,16 +42,52 @@ type MultiOrgAlertmanager struct {
 	metrics *metrics.MultiOrgAlertmanager
 }
 
-func NewMultiOrgAlertmanager(cfg *setting.Cfg, configStore store.AlertingStore, orgStore store.OrgStore, kvStore kvstore.KVStore, m *metrics.MultiOrgAlertmanager) *MultiOrgAlertmanager {
-	return &MultiOrgAlertmanager{
+func NewMultiOrgAlertmanager(cfg *setting.Cfg, configStore store.AlertingStore, orgStore store.OrgStore, kvStore kvstore.KVStore, m *metrics.MultiOrgAlertmanager, l log.Logger) (*MultiOrgAlertmanager, error) {
+	moa := &MultiOrgAlertmanager{
+		logger:        l,
 		settings:      cfg,
-		logger:        log.New("multiorg.alertmanager"),
 		alertmanagers: map[int64]*Alertmanager{},
 		configStore:   configStore,
 		orgStore:      orgStore,
 		kvStore:       kvStore,
 		metrics:       m,
 	}
+
+	clusterLogger := gokit_log.With(gokit_log.NewLogfmtLogger(logging.NewWrapper(l)), "component", "cluster")
+	moa.peer = &NilPeer{}
+	if len(cfg.HAPeers) > 0 {
+		peer, err := cluster.Create(
+			clusterLogger,
+			m.Registerer,
+			cfg.HAListenAddr,
+			cfg.HAAdvertiseAddr,
+			cfg.HAPeers, // peers
+			true,
+			cfg.HAPushPullInterval,
+			cfg.HAGossipInterval,
+			cluster.DefaultTcpTimeout,
+			cluster.DefaultProbeTimeout,
+			cluster.DefaultProbeInterval,
+			nil,
+		)
+
+		if err != nil {
+			return nil, fmt.Errorf("unable to initialize gossip mesh: %w", err)
+		}
+
+		err = peer.Join(cluster.DefaultReconnectInterval, cluster.DefaultReconnectTimeout)
+		if err != nil {
+			l.Error("msg", "unable to join gossip mesh while initializing cluster for high availability mode", "err", err)
+		}
+		// Attempt to verify the number of peers for 30s every 2s. The risk here is what we send a notification "too soon".
+		// Which should _never_ happen given we share the notification log via the database so the risk of double notification is very low.
+		var ctx context.Context
+		ctx, moa.settleCancel = context.WithTimeout(context.Background(), 30*time.Second)
+		go peer.Settle(ctx, cluster.DefaultGossipInterval*10)
+		moa.peer = peer
+	}
+
+	return moa, nil
 }
 
 func (moa *MultiOrgAlertmanager) Run(ctx context.Context) error {
@@ -53,7 +98,7 @@ func (moa *MultiOrgAlertmanager) Run(ctx context.Context) error {
 		case <-ctx.Done():
 			moa.StopAndWait()
 			return nil
-		case <-time.After(SyncOrgsPollInterval):
+		case <-time.After(moa.settings.AlertmanagerConfigPollInterval):
 			if err := moa.LoadAndSyncAlertmanagersForOrgs(ctx); err != nil {
 				moa.logger.Error("error while synchronizing Alertmanager orgs", "err", err)
 			}
@@ -90,7 +135,7 @@ func (moa *MultiOrgAlertmanager) SyncAlertmanagersForOrgs(orgIDs []int64) {
 			// To export them, we need to translate the metrics from each individual registry and,
 			// then aggregate them on the main registry.
 			m := metrics.NewAlertmanagerMetrics(moa.metrics.GetOrCreateOrgRegistry(orgID))
-			am, err := newAlertmanager(orgID, moa.settings, moa.configStore, moa.kvStore, m)
+			am, err := newAlertmanager(orgID, moa.settings, moa.configStore, moa.kvStore, moa.peer, m)
 			if err != nil {
 				moa.logger.Error("unable to create Alertmanager for org", "org", orgID, "err", err)
 			}
@@ -130,6 +175,14 @@ func (moa *MultiOrgAlertmanager) StopAndWait() {
 	for _, am := range moa.alertmanagers {
 		am.StopAndWait()
 	}
+
+	p, ok := moa.peer.(*cluster.Peer)
+	if ok {
+		moa.settleCancel()
+		if err := p.Leave(10 * time.Second); err != nil {
+			moa.logger.Warn("unable to leave the gossip mesh", "err", err)
+		}
+	}
 }
 
 // AlertmanagerFor returns the Alertmanager instance for the organization provided.
@@ -150,3 +203,16 @@ func (moa *MultiOrgAlertmanager) AlertmanagerFor(orgID int64) (*Alertmanager, er
 
 	return orgAM, nil
 }
+
+// NilPeer and NilChannel implements the Alertmanager clustering interface.
+type NilPeer struct{}
+
+func (p *NilPeer) Position() int                   { return 0 }
+func (p *NilPeer) WaitReady(context.Context) error { return nil }
+func (p *NilPeer) AddState(string, cluster.State, prometheus.Registerer) cluster.ClusterChannel {
+	return &NilChannel{}
+}
+
+type NilChannel struct{}
+
+func (c *NilChannel) Broadcast([]byte) {}

pkg/services/ngalert/notifier/multiorg_alertmanager_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/services/ngalert/metrics"
 	"github.com/grafana/grafana/pkg/services/ngalert/models"
 	"github.com/grafana/grafana/pkg/setting"
@@ -18,7 +19,6 @@ import (
 )
 
 func TestMultiOrgAlertmanager_SyncAlertmanagersForOrgs(t *testing.T) {
-	t.Skipf("Skipping multiorg alertmanager tests for now")
 	configStore := &FakeConfigStore{
 		configs: map[int64]*models.AlertConfiguration{},
 	}
@@ -28,12 +28,15 @@ func TestMultiOrgAlertmanager_SyncAlertmanagersForOrgs(t *testing.T) {
 
 	tmpDir, err := ioutil.TempDir("", "test")
 	require.NoError(t, err)
-
-	SyncOrgsPollInterval = 10 * time.Minute // Don't poll in unit tests.
 	kvStore := newFakeKVStore(t)
 	reg := prometheus.NewPedanticRegistry()
 	m := metrics.NewNGAlert(reg)
-	mam := NewMultiOrgAlertmanager(&setting.Cfg{DataPath: tmpDir}, configStore, orgStore, kvStore, m.GetMultiOrgAlertmanagerMetrics())
+	cfg := &setting.Cfg{
+		DataPath:                       tmpDir,
+		AlertmanagerConfigPollInterval: 3 * time.Minute, // do not poll in tests
+	}
+	mam, err := NewMultiOrgAlertmanager(cfg, configStore, orgStore, kvStore, m.GetMultiOrgAlertmanagerMetrics(), log.New("testlogger"))
+	require.NoError(t, err)
 	ctx := context.Background()
 
 	t.Cleanup(cleanOrgDirectories(tmpDir, t))
@@ -82,22 +85,23 @@ grafana_alerting_discovered_configurations 4
 }
 
 func TestMultiOrgAlertmanager_AlertmanagerFor(t *testing.T) {
-	t.Skipf("Skipping multiorg alertmanager tests for now")
 	configStore := &FakeConfigStore{
 		configs: map[int64]*models.AlertConfiguration{},
 	}
 	orgStore := &FakeOrgStore{
 		orgs: []int64{1, 2, 3},
 	}
-
 	tmpDir, err := ioutil.TempDir("", "test")
 	require.NoError(t, err)
-
-	SyncOrgsPollInterval = 10 * time.Minute // Don't poll in unit tests.
+	cfg := &setting.Cfg{
+		DataPath:                       tmpDir,
+		AlertmanagerConfigPollInterval: 3 * time.Minute, // do not poll in tests
+	}
 	kvStore := newFakeKVStore(t)
 	reg := prometheus.NewPedanticRegistry()
 	m := metrics.NewNGAlert(reg)
-	mam := NewMultiOrgAlertmanager(&setting.Cfg{DataPath: tmpDir}, configStore, orgStore, kvStore, m.GetMultiOrgAlertmanagerMetrics())
+	mam, err := NewMultiOrgAlertmanager(cfg, configStore, orgStore, kvStore, m.GetMultiOrgAlertmanagerMetrics(), log.New("testlogger"))
+	require.NoError(t, err)
 	ctx := context.Background()
 
 	t.Cleanup(cleanOrgDirectories(tmpDir, t))

pkg/services/ngalert/schedule/schedule_unit_test.go

@@ -231,6 +231,8 @@ func setupScheduler(t *testing.T, rs store.RuleStore, is store.InstanceStore, ac
 	mockedClock := clock.NewMock()
 	logger := log.New("ngalert schedule test")
 	m := metrics.NewNGAlert(prometheus.NewPedanticRegistry())
+	moa, err := notifier.NewMultiOrgAlertmanager(&setting.Cfg{}, &notifier.FakeConfigStore{}, &notifier.FakeOrgStore{}, &notifier.FakeKVStore{}, nil, log.New("testlogger"))
+	require.NoError(t, err)
 	schedCfg := SchedulerCfg{
 		C:                       mockedClock,
 		BaseInterval:            time.Second,
@@ -239,7 +241,7 @@ func setupScheduler(t *testing.T, rs store.RuleStore, is store.InstanceStore, ac
 		RuleStore:               rs,
 		InstanceStore:           is,
 		AdminConfigStore:        acs,
-		MultiOrgNotifier:        notifier.NewMultiOrgAlertmanager(&setting.Cfg{}, &notifier.FakeConfigStore{}, &notifier.FakeOrgStore{}, &notifier.FakeKVStore{}, nil),
+		MultiOrgNotifier:        moa,
 		Logger:                  logger,
 		Metrics:                 m.GetSchedulerMetrics(),
 		AdminConfigPollInterval: 10 * time.Minute, // do not poll in unit tests.