Merge pull request #1699 from prometheus/fabxc-multiam

notifier: dispatch to multiple Alertmanagers

cmd/prometheus/config.go

@@ -19,6 +19,7 @@ import (
 	"net"
 	"net/url"
 	"os"
+	"sort"
 	"strings"
 	"text/template"
 	"time"
@@ -48,9 +49,12 @@ var cfg = struct {
 	web         web.Options
 	remote      remote.Options
 
-	prometheusURL string
+	alertmanagerURLs stringset
-	influxdbURL   string
+	prometheusURL    string
-}{}
+	influxdbURL      string
+}{
+	alertmanagerURLs: stringset{},
+}
 
 func init() {
 	flag.CommandLine.Init(os.Args[0], flag.ContinueOnError)
@@ -206,9 +210,9 @@ func init() {
 	)
 
 	// Alertmanager.
-	cfg.fs.StringVar(
+	cfg.fs.Var(
-		&cfg.notifier.AlertmanagerURL, "alertmanager.url", "",
+		&cfg.alertmanagerURLs, "alertmanager.url",
-		"The URL of the alert manager to send notifications to.",
+		"Comma-separated list of Alertmanager URLs to send notifications to.",
 	)
 	cfg.fs.IntVar(
 		&cfg.notifier.QueueCapacity, "alertmanager.notification-queue-capacity", 10000,
@@ -249,8 +253,11 @@ func parse(args []string) error {
 	if err := parseInfluxdbURL(); err != nil {
 		return err
 	}
-	if err := validateAlertmanagerURL(); err != nil {
+	for u := range cfg.alertmanagerURLs {
-		return err
+		if err := validateAlertmanagerURL(u); err != nil {
+			return err
+		}
+		cfg.notifier.AlertmanagerURLs = cfg.alertmanagerURLs.slice()
 	}
 
 	cfg.remote.InfluxdbPassword = os.Getenv("INFLUXDB_PW")
@@ -307,19 +314,19 @@ func parseInfluxdbURL() error {
 	return nil
 }
 
-func validateAlertmanagerURL() error {
+func validateAlertmanagerURL(u string) error {
-	if cfg.notifier.AlertmanagerURL == "" {
+	if u == "" {
 		return nil
 	}
-	if ok := govalidator.IsURL(cfg.notifier.AlertmanagerURL); !ok {
+	if ok := govalidator.IsURL(u); !ok {
-		return fmt.Errorf("invalid Alertmanager URL: %s", cfg.notifier.AlertmanagerURL)
+		return fmt.Errorf("invalid Alertmanager URL: %s", u)
 	}
-	url, err := url.Parse(cfg.notifier.AlertmanagerURL)
+	url, err := url.Parse(u)
 	if err != nil {
 		return err
 	}
 	if url.Scheme == "" {
-		return fmt.Errorf("missing scheme in Alertmanager URL: %s", cfg.notifier.AlertmanagerURL)
+		return fmt.Errorf("missing scheme in Alertmanager URL: %s", u)
 	}
 	return nil
 }
@@ -384,3 +391,28 @@ func usage() {
 		panic(fmt.Errorf("error executing usage template: %s", err))
 	}
 }
+
+type stringset map[string]struct{}
+
+func (ss stringset) Set(s string) error {
+	for _, v := range strings.Split(s, ",") {
+		v = strings.TrimSpace(v)
+		if v != "" {
+			ss[v] = struct{}{}
+		}
+	}
+	return nil
+}
+
+func (ss stringset) String() string {
+	return strings.Join(ss.slice(), ",")
+}
+
+func (ss stringset) slice() []string {
+	slice := make([]string, 0, len(ss))
+	for k := range ss {
+		slice = append(slice, k)
+	}
+	sort.Strings(slice)
+	return slice
+}

notifier/notifier.go

@@ -20,6 +20,7 @@ import (
 	"net/http"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -38,8 +39,9 @@ const (
 
 // String constants for instrumentation.
 const (
-	namespace = "prometheus"
+	namespace         = "prometheus"
-	subsystem = "notifications"
+	subsystem         = "notifications"
+	alertmanagerLabel = "alertmanager"
 )
 
 // Notifier is responsible for dispatching alert notifications to an
@@ -53,20 +55,20 @@ type Notifier struct {
 	ctx    context.Context
 	cancel func()
 
-	latency       prometheus.Summary
+	latency       *prometheus.SummaryVec
-	errors        prometheus.Counter
+	errors        *prometheus.CounterVec
+	sent          *prometheus.CounterVec
 	dropped       prometheus.Counter
-	sent          prometheus.Counter
 	queueLength   prometheus.Gauge
 	queueCapacity prometheus.Metric
 }
 
 // Options are the configurable parameters of a Handler.
 type Options struct {
-	AlertmanagerURL string
+	AlertmanagerURLs []string
-	QueueCapacity   int
+	QueueCapacity    int
-	Timeout         time.Duration
+	Timeout          time.Duration
-	ExternalLabels  model.LabelSet
+	ExternalLabels   model.LabelSet
 }
 
 // New constructs a neww Notifier.
@@ -80,24 +82,30 @@ func New(o *Options) *Notifier {
 		more:   make(chan struct{}, 1),
 		opts:   o,
 
-		latency: prometheus.NewSummary(prometheus.SummaryOpts{
+		latency: prometheus.NewSummaryVec(prometheus.SummaryOpts{
 			Namespace: namespace,
 			Subsystem: subsystem,
 			Name:      "latency_seconds",
 			Help:      "Latency quantiles for sending alert notifications (not including dropped notifications).",
-		}),
+		},
-		errors: prometheus.NewCounter(prometheus.CounterOpts{
+			[]string{alertmanagerLabel},
+		),
+		errors: prometheus.NewCounterVec(prometheus.CounterOpts{
 			Namespace: namespace,
 			Subsystem: subsystem,
 			Name:      "errors_total",
 			Help:      "Total number of errors sending alert notifications.",
-		}),
+		},
-		sent: prometheus.NewCounter(prometheus.CounterOpts{
+			[]string{alertmanagerLabel},
+		),
+		sent: prometheus.NewCounterVec(prometheus.CounterOpts{
 			Namespace: namespace,
 			Subsystem: subsystem,
 			Name:      "sent_total",
 			Help:      "Total number of alerts successfully sent.",
-		}),
+		},
+			[]string{alertmanagerLabel},
+		),
 		dropped: prometheus.NewCounter(prometheus.CounterOpts{
 			Namespace: namespace,
 			Subsystem: subsystem,
@@ -160,9 +168,11 @@ func (n *Notifier) nextBatch() []*model.Alert {
 
 // Run dispatches notifications continuously.
 func (n *Notifier) Run() {
+	numAMs := len(n.opts.AlertmanagerURLs)
 	// Just warn once in the beginning to prevent noisy logs.
-	if n.opts.AlertmanagerURL == "" {
+	if numAMs == 0 {
-		log.Warnf("No AlertManager configured, not dispatching any alerts")
+		log.Warnf("No AlertManagers configured, not dispatching any alerts")
+		return
 	}
 
 	for {
@@ -171,28 +181,21 @@ func (n *Notifier) Run() {
 			return
 		case <-n.more:
 		}
-
 		alerts := n.nextBatch()
 
-		if len(alerts) == 0 {
+		if numAMs > 0 {
-			continue
-		}
-		if n.opts.AlertmanagerURL == "" {
-			n.dropped.Add(float64(len(alerts)))
-			continue
-		}
 
-		begin := time.Now()
+			if len(alerts) > 0 {
-
+				numErrors := n.sendAll(alerts...)
-		if err := n.send(alerts...); err != nil {
+				// Increment the dropped counter if we could not send
-			log.Errorf("Error sending %d alerts: %s", len(alerts), err)
+				// successfully to a single AlertManager.
-			n.errors.Inc()
+				if numErrors == numAMs {
+					n.dropped.Add(float64(len(alerts)))
+				}
+			}
+		} else {
 			n.dropped.Add(float64(len(alerts)))
 		}
-
-		n.latency.Observe(float64(time.Since(begin)) / float64(time.Second))
-		n.sent.Add(float64(len(alerts)))
-
 		// If the queue still has items left, kick off the next iteration.
 		if n.queueLen() > 0 {
 			n.setMore()
@@ -239,11 +242,15 @@ func (n *Notifier) setMore() {
 	}
 }
 
-func (n *Notifier) postURL() string {
+func postURL(u string) string {
-	return strings.TrimRight(n.opts.AlertmanagerURL, "/") + alertPushEndpoint
+	return strings.TrimRight(u, "/") + alertPushEndpoint
 }
 
-func (n *Notifier) send(alerts ...*model.Alert) error {
+// sendAll sends the alerts to all configured Alertmanagers at concurrently.
+// It returns the number of sends that have failed.
+func (n *Notifier) sendAll(alerts ...*model.Alert) int {
+	begin := time.Now()
+
 	// Attach external labels before sending alerts.
 	for _, a := range alerts {
 		for ln, lv := range n.opts.ExternalLabels {
@@ -253,36 +260,62 @@ func (n *Notifier) send(alerts ...*model.Alert) error {
 		}
 	}
 
-	var buf bytes.Buffer
+	b, err := json.Marshal(alerts)
-	if err := json.NewEncoder(&buf).Encode(alerts); err != nil {
+	if err != nil {
-		return err
+		log.Errorf("Encoding alerts failed: %s", err)
+		return len(n.opts.AlertmanagerURLs)
 	}
 	ctx, _ := context.WithTimeout(context.Background(), n.opts.Timeout)
 
-	resp, err := ctxhttp.Post(ctx, http.DefaultClient, n.postURL(), contentTypeJSON, &buf)
+	send := func(u string) error {
-	if err != nil {
+		resp, err := ctxhttp.Post(ctx, http.DefaultClient, postURL(u), contentTypeJSON, bytes.NewReader(b))
+		if err != nil {
+			return err
+		}
+		defer resp.Body.Close()
+
+		if resp.StatusCode/100 != 2 {
+			return fmt.Errorf("bad response status %v", resp.Status)
+		}
 		return err
 	}
-	defer resp.Body.Close()
 
-	if resp.StatusCode/100 != 2 {
+	var (
-		return fmt.Errorf("bad response status %v", resp.Status)
+		wg        sync.WaitGroup
+		numErrors uint64
+	)
+	for _, u := range n.opts.AlertmanagerURLs {
+		wg.Add(1)
+
+		go func(u string) {
+			if err := send(u); err != nil {
+				log.With("alertmanager", u).With("count", fmt.Sprintf("%d", len(alerts))).Errorf("Error sending alerts: %s", err)
+				n.errors.WithLabelValues(u).Inc()
+				atomic.AddUint64(&numErrors, 1)
+			}
+			n.latency.WithLabelValues(u).Observe(float64(time.Since(begin)) / float64(time.Second))
+			n.sent.WithLabelValues(u).Add(float64(len(alerts)))
+
+			wg.Done()
+		}(u)
 	}
-	return nil
+	wg.Wait()
+
+	return int(numErrors)
 }
 
 // Stop shuts down the notification handler.
 func (n *Notifier) Stop() {
 	log.Info("Stopping notification handler...")
-
 	n.cancel()
 }
 
 // Describe implements prometheus.Collector.
 func (n *Notifier) Describe(ch chan<- *prometheus.Desc) {
-	ch <- n.latency.Desc()
+	n.latency.Describe(ch)
-	ch <- n.errors.Desc()
+	n.errors.Describe(ch)
-	ch <- n.sent.Desc()
+	n.sent.Describe(ch)
+
 	ch <- n.dropped.Desc()
 	ch <- n.queueLength.Desc()
 	ch <- n.queueCapacity.Desc()
@@ -292,9 +325,10 @@ func (n *Notifier) Describe(ch chan<- *prometheus.Desc) {
 func (n *Notifier) Collect(ch chan<- prometheus.Metric) {
 	n.queueLength.Set(float64(n.queueLen()))
 
-	ch <- n.latency
+	n.latency.Collect(ch)
-	ch <- n.errors
+	n.errors.Collect(ch)
-	ch <- n.sent
+	n.sent.Collect(ch)
+
 	ch <- n.dropped
 	ch <- n.queueLength
 	ch <- n.queueCapacity

notifier/notifier_test.go

@@ -25,7 +25,7 @@ import (
 	"github.com/prometheus/common/model"
 )
 
-func TestHandlerPostURL(t *testing.T) {
+func TestPostURL(t *testing.T) {
 	var cases = []struct {
 		in, out string
 	}{
@@ -50,13 +50,8 @@ func TestHandlerPostURL(t *testing.T) {
 			out: "http://localhost:9093/prefix/api/v1/alerts",
 		},
 	}
-	h := &Notifier{
-		opts: &Options{},
-	}
-
 	for _, c := range cases {
-		h.opts.AlertmanagerURL = c.in
+		if res := postURL(c.in); res != c.out {
-		if res := h.postURL(); res != c.out {
 			t.Errorf("Expected post URL %q for %q but got %q", c.out, c.in, res)
 		}
 	}
@@ -119,13 +114,13 @@ func alertsEqual(a, b model.Alerts) bool {
 	return true
 }
 
-func TestHandlerSend(t *testing.T) {
+func TestHandlerSendAll(t *testing.T) {
 	var (
-		expected model.Alerts
+		expected         model.Alerts
-		status   int
+		status1, status2 int
 	)
 
-	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+	f := func(w http.ResponseWriter, r *http.Request) {
 		if r.URL.Path != alertPushEndpoint {
 			t.Fatalf("Bad endpoint %q used, expected %q", r.URL.Path, alertPushEndpoint)
 		}
@@ -140,16 +135,23 @@ func TestHandlerSend(t *testing.T) {
 			t.Errorf("%#v %#v", *alerts[0], *expected[0])
 			t.Fatalf("Unexpected alerts received %v exp %v", alerts, expected)
 		}
-
+	}
-		w.WriteHeader(status)
+	server1 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		f(w, r)
+		w.WriteHeader(status1)
+	}))
+	server2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		f(w, r)
+		w.WriteHeader(status2)
 	}))
 
-	defer server.Close()
+	defer server1.Close()
+	defer server2.Close()
 
 	h := New(&Options{
-		AlertmanagerURL: server.URL,
+		AlertmanagerURLs: []string{server1.URL, server2.URL},
-		Timeout:         time.Minute,
+		Timeout:          time.Minute,
-		ExternalLabels:  model.LabelSet{"a": "b"},
+		ExternalLabels:   model.LabelSet{"a": "b"},
 	})
 
 	for i := range make([]struct{}, maxBatchSize) {
@@ -166,16 +168,20 @@ func TestHandlerSend(t *testing.T) {
 		})
 	}
 
-	status = http.StatusOK
+	status1 = http.StatusOK
-
+	status2 = http.StatusOK
-	if err := h.send(h.queue...); err != nil {
+	if ne := h.sendAll(h.queue...); ne != 0 {
-		t.Fatalf("Unexpected error: %s", err)
+		t.Fatalf("Unexpected number of failed sends: %d", ne)
 	}
 
-	status = 500
+	status1 = http.StatusNotFound
+	if ne := h.sendAll(h.queue...); ne != 1 {
+		t.Fatalf("Unexpected number of failed sends: %d", ne)
+	}
 
-	if err := h.send(h.queue...); err == nil {
+	status2 = http.StatusInternalServerError
-		t.Fatalf("Expected error but got none")
+	if ne := h.sendAll(h.queue...); ne != 2 {
+		t.Fatalf("Unexpected number of failed sends: %d", ne)
 	}
 }
 
@@ -203,9 +209,9 @@ func TestHandlerFull(t *testing.T) {
 	}))
 
 	h := New(&Options{
-		AlertmanagerURL: server.URL,
+		AlertmanagerURLs: []string{server.URL},
-		Timeout:         time.Second,
+		Timeout:          time.Second,
-		QueueCapacity:   3 * maxBatchSize,
+		QueueCapacity:    3 * maxBatchSize,
 	})
 
 	var alerts model.Alerts