Update for newer upstream alertmanager

Author: bboreham

pkg/alertmanager/alertmanager.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/go-kit/kit/log"
 	"github.com/prometheus/alertmanager/api"
+	"github.com/prometheus/alertmanager/cluster"
 	"github.com/prometheus/alertmanager/config"
 	"github.com/prometheus/alertmanager/dispatch"
 	"github.com/prometheus/alertmanager/inhibit"
@@ -24,7 +25,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/route"
 	"github.com/prometheus/prometheus/pkg/labels"
-	"github.com/weaveworks/mesh"
 )
 
 const notificationLogMaintenancePeriod = 15 * time.Minute
@@ -35,31 +35,33 @@ type Config struct {
 	// Used to persist notification logs and silences on disk.
 	DataDir     string
 	Logger      log.Logger
-	MeshRouter  gossipRouter
 	Retention   time.Duration
 	ExternalURL *url.URL
 }
 
 // An Alertmanager manages the alerts for one user.
 type Alertmanager struct {
-	cfg        *Config
-	api        *api.API
-	logger     log.Logger
-	nflog      nflog.Log
-	silences   *silence.Silences
-	marker     types.Marker
-	alerts     *mem.Alerts
-	dispatcher *dispatch.Dispatcher
-	inhibitor  *inhibit.Inhibitor
-	stop       chan struct{}
-	wg         sync.WaitGroup
-	router     *route.Router
+	cfg         *Config
+	api         *api.API
+	logger      log.Logger
+	nflog       *nflog.Log
+	silences    *silence.Silences
+	marker      types.Marker
+	alerts      *mem.Alerts
+	dispatcher  *dispatch.Dispatcher
+	inhibitor   *inhibit.Inhibitor
+	stop        chan struct{}
+	wg          sync.WaitGroup
+	router      *route.Router
+	peer        *cluster.Peer
+	peerTimeout time.Duration
 }
 
 // New creates a new Alertmanager.
-func New(cfg *Config) (*Alertmanager, error) {
+func New(peer *cluster.Peer, peerTimeout time.Duration, cfg *Config) (*Alertmanager, error) {
 	am := &Alertmanager{
 		cfg:    cfg,
+		peer:   peer,
 		logger: log.With(cfg.Logger, "user", cfg.UserID),
 		stop:   make(chan struct{}),
 	}
@@ -68,9 +70,6 @@ func New(cfg *Config) (*Alertmanager, error) {
 	nflogID := fmt.Sprintf("nflog:%s", cfg.UserID)
 	var err error
 	am.nflog, err = nflog.New(
-		nflog.WithMesh(func(g mesh.Gossiper) mesh.Gossip {
-			return cfg.MeshRouter.newGossip(nflogID, g)
-		}),
 		nflog.WithRetention(cfg.Retention),
 		nflog.WithSnapshot(filepath.Join(cfg.DataDir, nflogID)),
 		nflog.WithMaintenance(notificationLogMaintenancePeriod, am.stop, am.wg.Done),
@@ -86,22 +85,25 @@ func New(cfg *Config) (*Alertmanager, error) {
 
 	am.marker = types.NewMarker()
 
+	// TODO(cortex): Build a registry that can merge metrics from multiple users.
+	// For now, these metrics are ignored, as we can't register the same
+	// metric twice with a single registry.
+	localRegistry := prometheus.NewRegistry()
+
 	silencesID := fmt.Sprintf("silences:%s", cfg.UserID)
 	am.silences, err = silence.New(silence.Options{
 		SnapshotFile: filepath.Join(cfg.DataDir, silencesID),
 		Retention:    cfg.Retention,
 		Logger:       log.With(am.logger, "component", "silences"),
-		// TODO(cortex): Build a registry that can merge metrics from multiple users.
-		// For now, these metrics are ignored, as we can't register the same
-		// metric twice with a single registry.
-		Metrics: prometheus.NewRegistry(),
-		Gossip: func(g mesh.Gossiper) mesh.Gossip {
-			return cfg.MeshRouter.newGossip(silencesID, g)
-		},
+		Metrics:      localRegistry,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to create silences: %v", err)
 	}
+	if peer != nil {
+		c := peer.AddState("sil:"+cfg.UserID, am.silences, localRegistry)
+		am.silences.SetBroadcast(c.Broadcast)
+	}
 
 	am.wg.Add(1)
 	go func() {
@@ -122,7 +124,7 @@ func New(cfg *Config) (*Alertmanager, error) {
 			return am.dispatcher.Groups(matchers)
 		},
 		marker.Status,
-		nil, // Passing a nil mesh router since we don't show mesh router information in Cortex anyway.
+		peer,
 		log.With(am.logger, "component", "api"),
 	)
 
@@ -148,6 +150,14 @@ func New(cfg *Config) (*Alertmanager, error) {
 	return am, nil
 }
 
+// clusterWait returns a function that inspects the current peer state and returns
+// a duration of one base timeout for each peer with a higher ID than ourselves.
+func clusterWait(p *cluster.Peer, timeout time.Duration) func() time.Duration {
+	return func() time.Duration {
+		return time.Duration(p.Position()) * timeout
+	}
+}
+
 // ApplyConfig applies a new configuration to an Alertmanager.
 func (am *Alertmanager) ApplyConfig(conf *config.Config) error {
 	var (
@@ -176,7 +186,7 @@ func (am *Alertmanager) ApplyConfig(conf *config.Config) error {
 
 	am.inhibitor = inhibit.NewInhibitor(am.alerts, conf.InhibitRules, am.marker, log.With(am.logger, "component", "inhibitor"))
 
-	waitFunc := meshWait(am.cfg.MeshRouter, 5*time.Second)
+	waitFunc := clusterWait(am.peer, am.peerTimeout)
 	timeoutFunc := func(d time.Duration) time.Duration {
 		if d < notify.MinTimeout {
 			d = notify.MinTimeout
@@ -192,6 +202,7 @@ func (am *Alertmanager) ApplyConfig(conf *config.Config) error {
 		am.silences,
 		am.nflog,
 		am.marker,
+		am.peer,
 		log.With(am.logger, "component", "pipeline"),
 	)
 	am.dispatcher = dispatch.NewDispatcher(