feat(provider): add subscriber channel metrics (#4630)

Add `alertmanager_alerts_subscriber_channel_writes_total` metric to
track the number of alerts written to subscriber channels.
A drop in the rate of this metric may indicate a problem with the
ingestion of alerts by subscribers (inhibitor and dispatcher).

Signed-off-by: Siavash Safi <[email protected]>

Author: siavashs

dispatch/dispatch.go

@@ -144,7 +144,7 @@ func (d *Dispatcher) Run() {
 	d.ctx, d.cancel = context.WithCancel(context.Background())
 	d.mtx.Unlock()
 
-	d.run(d.alerts.Subscribe())
+	d.run(d.alerts.Subscribe("dispatcher"))
 	close(d.done)
 }
 

dispatch/dispatch_test.go

@@ -392,16 +392,17 @@ route:
 
 	logger := promslog.NewNopLogger()
 	route := NewRoute(conf.Route, nil)
-	marker := types.NewMarker(prometheus.NewRegistry())
-	alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, nil, logger, nil)
+	reg := prometheus.NewRegistry()
+	marker := types.NewMarker(reg)
+	alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, nil, logger, reg)
 	if err != nil {
 		t.Fatal(err)
 	}
 	defer alerts.Close()
 
 	timeout := func(d time.Duration) time.Duration { return time.Duration(0) }
 	recorder := &recordStage{alerts: make(map[string]map[model.Fingerprint]*types.Alert)}
-	dispatcher := NewDispatcher(alerts, route, recorder, marker, timeout, testMaintenanceInterval, nil, logger, NewDispatcherMetrics(false, prometheus.NewRegistry()))
+	dispatcher := NewDispatcher(alerts, route, recorder, marker, timeout, testMaintenanceInterval, nil, logger, NewDispatcherMetrics(false, reg))
 	go dispatcher.Run()
 	defer dispatcher.Stop()
 
@@ -542,8 +543,9 @@ route:
 
 	logger := promslog.NewNopLogger()
 	route := NewRoute(conf.Route, nil)
-	marker := types.NewMarker(prometheus.NewRegistry())
-	alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, nil, logger, nil)
+	reg := prometheus.NewRegistry()
+	marker := types.NewMarker(reg)
+	alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, nil, logger, reg)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -552,7 +554,7 @@ route:
 	timeout := func(d time.Duration) time.Duration { return time.Duration(0) }
 	recorder := &recordStage{alerts: make(map[string]map[model.Fingerprint]*types.Alert)}
 	lim := limits{groups: 6}
-	m := NewDispatcherMetrics(true, prometheus.NewRegistry())
+	m := NewDispatcherMetrics(true, reg)
 	dispatcher := NewDispatcher(alerts, route, recorder, marker, timeout, testMaintenanceInterval, lim, logger, m)
 	go dispatcher.Run()
 	defer dispatcher.Stop()
@@ -663,15 +665,16 @@ func newAlert(labels model.LabelSet) *types.Alert {
 
 func TestDispatcherRace(t *testing.T) {
 	logger := promslog.NewNopLogger()
-	marker := types.NewMarker(prometheus.NewRegistry())
-	alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, nil, logger, nil)
+	reg := prometheus.NewRegistry()
+	marker := types.NewMarker(reg)
+	alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, nil, logger, reg)
 	if err != nil {
 		t.Fatal(err)
 	}
 	defer alerts.Close()
 
 	timeout := func(d time.Duration) time.Duration { return time.Duration(0) }
-	dispatcher := NewDispatcher(alerts, nil, nil, marker, timeout, testMaintenanceInterval, nil, logger, NewDispatcherMetrics(false, prometheus.NewRegistry()))
+	dispatcher := NewDispatcher(alerts, nil, nil, marker, timeout, testMaintenanceInterval, nil, logger, NewDispatcherMetrics(false, reg))
 	go dispatcher.Run()
 	dispatcher.Stop()
 }
@@ -680,8 +683,9 @@ func TestDispatcherRaceOnFirstAlertNotDeliveredWhenGroupWaitIsZero(t *testing.T)
 	const numAlerts = 5000
 
 	logger := promslog.NewNopLogger()
-	marker := types.NewMarker(prometheus.NewRegistry())
-	alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, nil, logger, nil)
+	reg := prometheus.NewRegistry()
+	marker := types.NewMarker(reg)
+	alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, nil, logger, reg)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -699,7 +703,7 @@ func TestDispatcherRaceOnFirstAlertNotDeliveredWhenGroupWaitIsZero(t *testing.T)
 
 	timeout := func(d time.Duration) time.Duration { return d }
 	recorder := &recordStage{alerts: make(map[string]map[model.Fingerprint]*types.Alert)}
-	dispatcher := NewDispatcher(alerts, route, recorder, marker, timeout, testMaintenanceInterval, nil, logger, NewDispatcherMetrics(false, prometheus.NewRegistry()))
+	dispatcher := NewDispatcher(alerts, route, recorder, marker, timeout, testMaintenanceInterval, nil, logger, NewDispatcherMetrics(false, reg))
 	go dispatcher.Run()
 	defer dispatcher.Stop()
 
@@ -735,7 +739,7 @@ func TestDispatcher_DoMaintenance(t *testing.T) {
 	r := prometheus.NewRegistry()
 	marker := types.NewMarker(r)
 
-	alerts, err := mem.NewAlerts(context.Background(), marker, time.Minute, nil, promslog.NewNopLogger(), nil)
+	alerts, err := mem.NewAlerts(context.Background(), marker, time.Minute, nil, promslog.NewNopLogger(), r)
 	if err != nil {
 		t.Fatal(err)
 	}

inhibit/inhibit.go

@@ -71,7 +71,7 @@ func NewInhibitor(ap provider.Alerts, rs []config.InhibitRule, mk types.AlertMar
 }
 
 func (ih *Inhibitor) run(ctx context.Context) {
-	it := ih.alerts.Subscribe()
+	it := ih.alerts.Subscribe("inhibitor")
 	defer it.Close()
 
 	for {

inhibit/inhibit_test.go

@@ -392,7 +392,7 @@ func newFakeAlerts(alerts []*types.Alert) *fakeAlerts {
 func (f *fakeAlerts) GetPending() provider.AlertIterator          { return nil }
 func (f *fakeAlerts) Get(model.Fingerprint) (*types.Alert, error) { return nil, nil }
 func (f *fakeAlerts) Put(...*types.Alert) error                   { return nil }
-func (f *fakeAlerts) Subscribe() provider.AlertIterator {
+func (f *fakeAlerts) Subscribe(name string) provider.AlertIterator {
 	ch := make(chan *types.Alert)
 	done := make(chan struct{})
 	go func() {

provider/mem/mem.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/common/model"
 
 	"github.com/prometheus/alertmanager/provider"
@@ -44,6 +45,8 @@ type Alerts struct {
 	callback AlertStoreCallback
 
 	logger *slog.Logger
+
+	subscriberChannelWrites *prometheus.CounterVec
 }
 
 type AlertStoreCallback interface {
@@ -61,13 +64,14 @@ type AlertStoreCallback interface {
 }
 
 type listeningAlerts struct {
+	name   string
 	alerts chan *types.Alert
 	done   chan struct{}
 }
 
 func (a *Alerts) registerMetrics(r prometheus.Registerer) {
 	newMemAlertByStatus := func(s types.AlertState) prometheus.GaugeFunc {
-		return prometheus.NewGaugeFunc(
+		return promauto.With(r).NewGaugeFunc(
 			prometheus.GaugeOpts{
 				Name:        "alertmanager_alerts",
 				Help:        "How many alerts by state.",
@@ -79,9 +83,17 @@ func (a *Alerts) registerMetrics(r prometheus.Registerer) {
 		)
 	}
 
-	r.MustRegister(newMemAlertByStatus(types.AlertStateActive))
-	r.MustRegister(newMemAlertByStatus(types.AlertStateSuppressed))
-	r.MustRegister(newMemAlertByStatus(types.AlertStateUnprocessed))
+	newMemAlertByStatus(types.AlertStateActive)
+	newMemAlertByStatus(types.AlertStateSuppressed)
+	newMemAlertByStatus(types.AlertStateUnprocessed)
+
+	a.subscriberChannelWrites = promauto.With(r).NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "alertmanager_alerts_subscriber_channel_writes_total",
+			Help: "Number of times alerts were written to subscriber channels",
+		},
+		[]string{"subscriber"},
+	)
 }
 
 // NewAlerts returns a new alert provider.
@@ -164,7 +176,7 @@ func max(a, b int) int {
 // Subscribe returns an iterator over active alerts that have not been
 // resolved and successfully notified about.
 // They are not guaranteed to be in chronological order.
-func (a *Alerts) Subscribe() provider.AlertIterator {
+func (a *Alerts) Subscribe(name string) provider.AlertIterator {
 	a.mtx.Lock()
 	defer a.mtx.Unlock()
 	var (
@@ -177,7 +189,7 @@ func (a *Alerts) Subscribe() provider.AlertIterator {
 		ch <- a
 	}
 
-	a.listeners[a.next] = listeningAlerts{alerts: ch, done: done}
+	a.listeners[a.next] = listeningAlerts{name: name, alerts: ch, done: done}
 	a.next++
 
 	return provider.NewAlertIterator(ch, done, nil)
@@ -252,6 +264,7 @@ func (a *Alerts) Put(alerts ...*types.Alert) error {
 		for _, l := range a.listeners {
 			select {
 			case l.alerts <- alert:
+				a.subscriberChannelWrites.WithLabelValues(l.name).Inc()
 			case <-l.done:
 			}
 		}

provider/mem/mem_test.go

@@ -87,12 +87,12 @@ func init() {
 // a listener can not unsubscribe as the lock is hold by `alerts.Lock`.
 func TestAlertsSubscribePutStarvation(t *testing.T) {
 	marker := types.NewMarker(prometheus.NewRegistry())
-	alerts, err := NewAlerts(context.Background(), marker, 30*time.Minute, noopCallback{}, promslog.NewNopLogger(), nil)
+	alerts, err := NewAlerts(context.Background(), marker, 30*time.Minute, noopCallback{}, promslog.NewNopLogger(), prometheus.NewRegistry())
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	iterator := alerts.Subscribe()
+	iterator := alerts.Subscribe("test")
 
 	alertsToInsert := []*types.Alert{}
 	// Exhaust alert channel
@@ -142,7 +142,7 @@ func TestDeadLock(t *testing.T) {
 
 	marker := types.NewMarker(prometheus.NewRegistry())
 	// Run gc every 5 milliseconds to increase the possibility of a deadlock with Subscribe()
-	alerts, err := NewAlerts(context.Background(), marker, 5*time.Millisecond, noopCallback{}, promslog.NewNopLogger(), nil)
+	alerts, err := NewAlerts(context.Background(), marker, 5*time.Millisecond, noopCallback{}, promslog.NewNopLogger(), prometheus.NewRegistry())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -176,7 +176,7 @@ func TestDeadLock(t *testing.T) {
 		for {
 			select {
 			case <-tick.C:
-				alerts.Subscribe()
+				alerts.Subscribe("test")
 			case <-stopAfter:
 				done <- true
 				break
@@ -195,7 +195,7 @@ func TestDeadLock(t *testing.T) {
 
 func TestAlertsPut(t *testing.T) {
 	marker := types.NewMarker(prometheus.NewRegistry())
-	alerts, err := NewAlerts(context.Background(), marker, 30*time.Minute, noopCallback{}, promslog.NewNopLogger(), nil)
+	alerts, err := NewAlerts(context.Background(), marker, 30*time.Minute, noopCallback{}, promslog.NewNopLogger(), prometheus.NewRegistry())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -223,7 +223,7 @@ func TestAlertsSubscribe(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	alerts, err := NewAlerts(ctx, marker, 30*time.Minute, noopCallback{}, promslog.NewNopLogger(), nil)
+	alerts, err := NewAlerts(ctx, marker, 30*time.Minute, noopCallback{}, promslog.NewNopLogger(), prometheus.NewRegistry())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -250,7 +250,7 @@ func TestAlertsSubscribe(t *testing.T) {
 		go func(i int) {
 			defer wg.Done()
 
-			it := alerts.Subscribe()
+			it := alerts.Subscribe("test")
 			defer it.Close()
 
 			received := make(map[model.Fingerprint]struct{})
@@ -621,3 +621,92 @@ func TestAlertsConcurrently(t *testing.T) {
 	}, 2*expire, expire)
 	require.Equal(t, int32(0), callback.alerts.Load())
 }
+
+func TestSubscriberChannelMetrics(t *testing.T) {
+	marker := types.NewMarker(prometheus.NewRegistry())
+	reg := prometheus.NewRegistry()
+	alerts, err := NewAlerts(context.Background(), marker, 30*time.Minute, noopCallback{}, promslog.NewNopLogger(), reg)
+	require.NoError(t, err)
+
+	subscriberName := "test_subscriber"
+
+	// Subscribe to alerts
+	iterator := alerts.Subscribe(subscriberName)
+	defer iterator.Close()
+
+	// Consume alerts in the background
+	go func() {
+		for range iterator.Next() {
+			// Just drain the channel
+		}
+	}()
+
+	// Helper function to get counter value
+	getCounterValue := func(name, labelName, labelValue string) float64 {
+		metrics, err := reg.Gather()
+		require.NoError(t, err)
+		for _, mf := range metrics {
+			if mf.GetName() == name {
+				for _, m := range mf.GetMetric() {
+					for _, label := range m.GetLabel() {
+						if label.GetName() == labelName && label.GetValue() == labelValue {
+							return m.GetCounter().GetValue()
+						}
+					}
+				}
+			}
+		}
+		return 0
+	}
+
+	// Initially, the counter should be 0
+	writeCount := getCounterValue("alertmanager_alerts_subscriber_channel_writes_total", "subscriber", subscriberName)
+	require.Equal(t, 0.0, writeCount, "subscriberChannelWrites should start at 0")
+
+	// Put some alerts
+	now := time.Now()
+	alertsToSend := []*types.Alert{
+		{
+			Alert: model.Alert{
+				Labels:       model.LabelSet{"test": "1"},
+				Annotations:  model.LabelSet{"foo": "bar"},
+				StartsAt:     now,
+				EndsAt:       now.Add(1 * time.Hour),
+				GeneratorURL: "http://example.com/prometheus",
+			},
+			UpdatedAt: now,
+			Timeout:   false,
+		},
+		{
+			Alert: model.Alert{
+				Labels:       model.LabelSet{"test": "2"},
+				Annotations:  model.LabelSet{"foo": "bar"},
+				StartsAt:     now,
+				EndsAt:       now.Add(1 * time.Hour),
+				GeneratorURL: "http://example.com/prometheus",
+			},
+			UpdatedAt: now,
+			Timeout:   false,
+		},
+		{
+			Alert: model.Alert{
+				Labels:       model.LabelSet{"test": "3"},
+				Annotations:  model.LabelSet{"foo": "bar"},
+				StartsAt:     now,
+				EndsAt:       now.Add(1 * time.Hour),
+				GeneratorURL: "http://example.com/prometheus",
+			},
+			UpdatedAt: now,
+			Timeout:   false,
+		},
+	}
+
+	err = alerts.Put(alertsToSend...)
+	require.NoError(t, err)
+
+	// Verify the counter incremented for each successful write
+	require.Eventually(t, func() bool {
+		writeCount := getCounterValue("alertmanager_alerts_subscriber_channel_writes_total", "subscriber", subscriberName)
+		return writeCount == float64(len(alertsToSend))
+	}, 1*time.Second, 10*time.Millisecond, "subscriberChannelWrites should equal the number of alerts sent")
+}

provider/provider.go

@@ -75,7 +75,7 @@ type Alerts interface {
 	// Subscribe returns an iterator over active alerts that have not been
 	// resolved and successfully notified about.
 	// They are not guaranteed to be in chronological order.
-	Subscribe() AlertIterator
+	Subscribe(name string) AlertIterator
 	// GetPending returns an iterator over all alerts that have
 	// pending notifications.
 	GetPending() AlertIterator