feat: configurable provider channel buffer length

This change makes the memory provider channel buffer length configurable.
The previously hard-coded value of 200 is used as default.
This should result in better performance and less timeouts during sudden alert spikes,
specially if the pipeline speed is affected by slow receivers.

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

Author: siavashs

cmd/alertmanager/main.go

@@ -143,13 +143,14 @@ func run() int {
 	}
 
 	var (
-		configFile          = kingpin.Flag("config.file", "Alertmanager configuration file name.").Default("alertmanager.yml").String()
-		dataDir             = kingpin.Flag("storage.path", "Base path for data storage.").Default("data/").String()
-		retention           = kingpin.Flag("data.retention", "How long to keep data for.").Default("120h").Duration()
-		maintenanceInterval = kingpin.Flag("data.maintenance-interval", "Interval between garbage collection and snapshotting to disk of the silences and the notification logs.").Default("15m").Duration()
-		maxSilences         = kingpin.Flag("silences.max-silences", "Maximum number of silences, including expired silences. If negative or zero, no limit is set.").Default("0").Int()
-		maxSilenceSizeBytes = kingpin.Flag("silences.max-silence-size-bytes", "Maximum silence size in bytes. If negative or zero, no limit is set.").Default("0").Int()
-		alertGCInterval     = kingpin.Flag("alerts.gc-interval", "Interval between alert GC.").Default("30m").Duration()
+		configFile               = kingpin.Flag("config.file", "Alertmanager configuration file name.").Default("alertmanager.yml").String()
+		dataDir                  = kingpin.Flag("storage.path", "Base path for data storage.").Default("data/").String()
+		retention                = kingpin.Flag("data.retention", "How long to keep data for.").Default("120h").Duration()
+		maintenanceInterval      = kingpin.Flag("data.maintenance-interval", "Interval between garbage collection and snapshotting to disk of the silences and the notification logs.").Default("15m").Duration()
+		maxSilences              = kingpin.Flag("silences.max-silences", "Maximum number of silences, including expired silences. If negative or zero, no limit is set.").Default("0").Int()
+		maxSilenceSizeBytes      = kingpin.Flag("silences.max-silence-size-bytes", "Maximum silence size in bytes. If negative or zero, no limit is set.").Default("0").Int()
+		alertChannelBufferLength = kingpin.Flag("alerts.channel-buffer-length", "Alert channel buffer length").Default("200").Int()
+		alertGCInterval          = kingpin.Flag("alerts.gc-interval", "Interval between alert GC.").Default("30m").Duration()
 
 		webConfig      = webflag.AddFlags(kingpin.CommandLine, ":9093")
 		externalURL    = kingpin.Flag("web.external-url", "The URL under which Alertmanager is externally reachable (for example, if Alertmanager is served via a reverse proxy). Used for generating relative and absolute links back to Alertmanager itself. If the URL has a path portion, it will be used to prefix all HTTP endpoints served by Alertmanager. If omitted, relevant URL components will be derived automatically.").String()
@@ -342,7 +343,7 @@ func run() int {
 		go peer.Settle(ctx, *gossipInterval*10)
 	}
 
-	alerts, err := mem.NewAlerts(context.Background(), marker, *alertGCInterval, nil, logger, prometheus.DefaultRegisterer)
+	alerts, err := mem.NewAlerts(context.Background(), marker, *alertChannelBufferLength, *alertGCInterval, nil, logger, prometheus.DefaultRegisterer)
 	if err != nil {
 		logger.Error("error creating memory provider", "err", err)
 		return 1

provider/mem/mem.go

@@ -15,6 +15,7 @@ package mem
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
 	"sync"
 	"time"
@@ -27,8 +28,6 @@ import (
 	"github.com/prometheus/alertmanager/types"
 )
 
-const alertChannelLength = 200
-
 // Alerts gives access to a set of alerts. All methods are goroutine-safe.
 type Alerts struct {
 	cancel context.CancelFunc
@@ -38,8 +37,9 @@ type Alerts struct {
 	alerts *store.Alerts
 	marker types.AlertMarker
 
-	listeners map[int]listeningAlerts
-	next      int
+	channelLength int
+	listeners     map[int]listeningAlerts
+	next          int
 
 	callback AlertStoreCallback
 
@@ -85,20 +85,25 @@ func (a *Alerts) registerMetrics(r prometheus.Registerer) {
 }
 
 // NewAlerts returns a new alert provider.
-func NewAlerts(ctx context.Context, m types.AlertMarker, intervalGC time.Duration, alertCallback AlertStoreCallback, l *slog.Logger, r prometheus.Registerer) (*Alerts, error) {
+func NewAlerts(ctx context.Context, m types.AlertMarker, channelLength int, intervalGC time.Duration, alertCallback AlertStoreCallback, l *slog.Logger, r prometheus.Registerer) (*Alerts, error) {
 	if alertCallback == nil {
 		alertCallback = noopCallback{}
 	}
 
+	if channelLength < 1 {
+		return nil, fmt.Errorf("channel length has to be > zero, provided %d", channelLength)
+	}
+
 	ctx, cancel := context.WithCancel(ctx)
 	a := &Alerts{
-		marker:    m,
-		alerts:    store.NewAlerts(),
-		cancel:    cancel,
-		listeners: map[int]listeningAlerts{},
-		next:      0,
-		logger:    l.With("component", "provider"),
-		callback:  alertCallback,
+		marker:        m,
+		alerts:        store.NewAlerts(),
+		cancel:        cancel,
+		channelLength: channelLength,
+		listeners:     map[int]listeningAlerts{},
+		next:          0,
+		logger:        l.With("component", "provider"),
+		callback:      alertCallback,
 	}
 
 	if r != nil {
@@ -170,7 +175,7 @@ func (a *Alerts) Subscribe() provider.AlertIterator {
 	var (
 		done   = make(chan struct{})
 		alerts = a.alerts.List()
-		ch     = make(chan *types.Alert, max(len(alerts), alertChannelLength))
+		ch     = make(chan *types.Alert, max(len(alerts), a.channelLength))
 	)
 
 	for _, a := range alerts {
@@ -187,7 +192,7 @@ func (a *Alerts) Subscribe() provider.AlertIterator {
 // pending notifications.
 func (a *Alerts) GetPending() provider.AlertIterator {
 	var (
-		ch   = make(chan *types.Alert, alertChannelLength)
+		ch   = make(chan *types.Alert, a.channelLength)
 		done = make(chan struct{})
 	)
 	a.mtx.Lock()