From bdd5bf03975e1bf6b11bf4f871956b14d92808ef Mon Sep 17 00:00:00 2001 From: Alexander Akhmetov Date: Tue, 19 Aug 2025 09:32:10 +0200 Subject: [PATCH] Alerting: Configurable queue and batch size in external alert manager notifier (#109831) --- pkg/services/ngalert/sender/notifier.go | 24 ++-- pkg/services/ngalert/sender/notifier_ext.go | 9 ++ pkg/services/ngalert/sender/notifier_test.go | 31 ++--- pkg/services/ngalert/sender/sender.go | 65 +++++++--- pkg/services/ngalert/sender/sender_test.go | 119 +++++++++++++++++++ 5 files changed, 205 insertions(+), 43 deletions(-) diff --git a/pkg/services/ngalert/sender/notifier.go b/pkg/services/ngalert/sender/notifier.go index ef1022921e2..2ba819a8c41 100644 --- a/pkg/services/ngalert/sender/notifier.go +++ b/pkg/services/ngalert/sender/notifier.go @@ -1,6 +1,6 @@ // THIS FILE IS COPIED FROM UPSTREAM // -// https://github.com/prometheus/prometheus/blob/293f0c9185260165fd7dabbf8a9e8758b32abeae/notifier/notifier.go +// https://github.com/prometheus/prometheus/blob/bd5b2ea95ce14fba11db871b4068313408465207/notifier/notifier.go // // Copyright 2013 The Prometheus Authors // Licensed under the Apache License, Version 2.0 (the "License"); @@ -31,7 +31,6 @@ import ( "time" "github.com/go-openapi/strfmt" - "github.com/grafana/grafana/pkg/util/httpclient" "github.com/prometheus/alertmanager/api/v2/models" "github.com/prometheus/client_golang/prometheus" config_util "github.com/prometheus/common/config" @@ -48,6 +47,9 @@ import ( ) const ( + // DefaultMaxBatchSize is the default maximum number of alerts to send in a single request to the alertmanager. + DefaultMaxBatchSize = 256 + contentTypeJSON = "application/json" ) @@ -135,6 +137,9 @@ type Options struct { Do func(ctx context.Context, client *http.Client, req *http.Request) (*http.Response, error) Registerer prometheus.Registerer + + // MaxBatchSize determines the maximum number of alerts to send in a single request to the alertmanager. + MaxBatchSize int } type alertMetrics struct { @@ -215,18 +220,15 @@ func newAlertMetrics(r prometheus.Registerer, queueCap int, queueLen, alertmanag return m } -func do(ctx context.Context, client *http.Client, req *http.Request) (*http.Response, error) { - if client == nil { - client = httpclient.New() - } - return client.Do(req.WithContext(ctx)) -} - // NewManager is the manager constructor. func NewManager(o *Options, logger *slog.Logger) *Manager { if o.Do == nil { o.Do = do } + // Set default MaxBatchSize if not provided. + if o.MaxBatchSize <= 0 { + o.MaxBatchSize = DefaultMaxBatchSize + } if logger == nil { logger = promslog.NewNopLogger() } @@ -253,8 +255,6 @@ func NewManager(o *Options, logger *slog.Logger) *Manager { return n } -const maxBatchSize = 64 - func (n *Manager) queueLen() int { n.mtx.RLock() defer n.mtx.RUnlock() @@ -268,7 +268,7 @@ func (n *Manager) nextBatch() []*Alert { var alerts []*Alert - if len(n.queue) > maxBatchSize { + if maxBatchSize := n.opts.MaxBatchSize; len(n.queue) > maxBatchSize { alerts = append(make([]*Alert, 0, maxBatchSize), n.queue[:maxBatchSize]...) n.queue = n.queue[maxBatchSize:] } else { diff --git a/pkg/services/ngalert/sender/notifier_ext.go b/pkg/services/ngalert/sender/notifier_ext.go index 348189eaf34..e57a85f5a4c 100644 --- a/pkg/services/ngalert/sender/notifier_ext.go +++ b/pkg/services/ngalert/sender/notifier_ext.go @@ -19,8 +19,17 @@ import ( "github.com/prometheus/prometheus/config" "github.com/prometheus/prometheus/model/labels" + + "github.com/grafana/grafana/pkg/util/httpclient" ) +func do(ctx context.Context, client *http.Client, req *http.Request) (*http.Response, error) { + if client == nil { + client = httpclient.New() + } + return client.Do(req.WithContext(ctx)) +} + // ApplyConfig updates the status state as the new config requires. // Extension: add new parameter headers. func (n *Manager) ApplyConfig(conf *config.Config, headers map[string]http.Header) error { diff --git a/pkg/services/ngalert/sender/notifier_test.go b/pkg/services/ngalert/sender/notifier_test.go index d1aec6068ab..20caf7e9aea 100644 --- a/pkg/services/ngalert/sender/notifier_test.go +++ b/pkg/services/ngalert/sender/notifier_test.go @@ -1,6 +1,6 @@ // THIS FILE IS COPIED FROM UPSTREAM // -// https://github.com/prometheus/prometheus/blob/293f0c9185260165fd7dabbf8a9e8758b32abeae/notifier/notifier_test.go +// https://github.com/prometheus/prometheus/blob/bd5b2ea95ce14fba11db871b4068313408465207/notifier/notifier_test.go // // Copyright 2013 The Prometheus Authors // Licensed under the Apache License, Version 2.0 (the "License"); @@ -40,15 +40,16 @@ import ( "go.uber.org/atomic" "gopkg.in/yaml.v2" - "github.com/prometheus/prometheus/discovery" - "github.com/prometheus/prometheus/config" + "github.com/prometheus/prometheus/discovery" _ "github.com/prometheus/prometheus/discovery/file" "github.com/prometheus/prometheus/discovery/targetgroup" "github.com/prometheus/prometheus/model/labels" "github.com/prometheus/prometheus/model/relabel" ) +const maxBatchSize = 256 + func TestPostPath(t *testing.T) { cases := []struct { in, out string @@ -411,7 +412,7 @@ func TestCustomDo(t *testing.T) { }, }, nil) - h.sendOne(context.Background(), nil, testURL, []byte(testBody), http.Header{}) + h.sendOne(context.Background(), nil, testURL, []byte(testBody), nil) require.True(t, received, "Expected to receive an alert, but didn't") } @@ -419,6 +420,7 @@ func TestCustomDo(t *testing.T) { func TestExternalLabels(t *testing.T) { h := NewManager(&Options{ QueueCapacity: 3 * maxBatchSize, + MaxBatchSize: maxBatchSize, ExternalLabels: labels.FromStrings("a", "b"), RelabelConfigs: []*relabel.Config{ { @@ -453,6 +455,7 @@ func TestExternalLabels(t *testing.T) { func TestHandlerRelabel(t *testing.T) { h := NewManager(&Options{ QueueCapacity: 3 * maxBatchSize, + MaxBatchSize: maxBatchSize, RelabelConfigs: []*relabel.Config{ { SourceLabels: model.LabelNames{"alertname"}, @@ -531,6 +534,7 @@ func TestHandlerQueuing(t *testing.T) { h := NewManager( &Options{ QueueCapacity: 3 * maxBatchSize, + MaxBatchSize: maxBatchSize, }, nil, ) @@ -660,7 +664,7 @@ alerting: require.NoError(t, err, "Unable to load YAML config.") require.Len(t, cfg.AlertingConfig.AlertmanagerConfigs, 1) - err = n.ApplyConfig(cfg, map[string]http.Header{}) + err = n.ApplyConfig(cfg, nil) require.NoError(t, err, "Error applying the config.") tgs := make(map[string][]*targetgroup.Group) @@ -711,7 +715,7 @@ alerting: require.NoError(t, err, "Unable to load YAML config.") require.Len(t, cfg.AlertingConfig.AlertmanagerConfigs, 1) - err = n.ApplyConfig(cfg, map[string]http.Header{}) + err = n.ApplyConfig(cfg, nil) require.NoError(t, err, "Error applying the config.") tgs := make(map[string][]*targetgroup.Group) @@ -1066,10 +1070,7 @@ func TestStop_DrainingEnabled(t *testing.T) { require.Equal(t, int64(2), alertsReceived.Load()) } -func TestIntegrationApplyConfig(t *testing.T) { - if testing.Short() { - t.Skip("skipping integration test in short mode") - } +func TestApplyConfig(t *testing.T) { targetURL := "alertmanager:9093" targetGroup := &targetgroup.Group{ Targets: []model.LabelSet{ @@ -1094,14 +1095,14 @@ alerting: require.Len(t, cfg.AlertingConfig.AlertmanagerConfigs, 1) // First, apply the config and reload. - require.NoError(t, n.ApplyConfig(cfg, map[string]http.Header{})) + require.NoError(t, n.ApplyConfig(cfg, nil)) tgs := map[string][]*targetgroup.Group{"config-0": {targetGroup}} n.reload(tgs) require.Len(t, n.Alertmanagers(), 1) require.Equal(t, alertmanagerURL, n.Alertmanagers()[0].String()) // Reapply the config. - require.NoError(t, n.ApplyConfig(cfg, map[string]http.Header{})) + require.NoError(t, n.ApplyConfig(cfg, nil)) // Ensure the known alertmanagers are not dropped. require.Len(t, n.Alertmanagers(), 1) require.Equal(t, alertmanagerURL, n.Alertmanagers()[0].String()) @@ -1119,7 +1120,7 @@ alerting: require.NoError(t, yaml.UnmarshalStrict([]byte(s), cfg)) require.Len(t, cfg.AlertingConfig.AlertmanagerConfigs, 2) - require.NoError(t, n.ApplyConfig(cfg, map[string]http.Header{})) + require.NoError(t, n.ApplyConfig(cfg, nil)) require.Len(t, n.Alertmanagers(), 1) // Ensure no unnecessary alertmanagers are injected. require.Empty(t, n.alertmanagers["config-0"].ams) @@ -1142,7 +1143,7 @@ alerting: require.NoError(t, yaml.UnmarshalStrict([]byte(s), cfg)) require.Len(t, cfg.AlertingConfig.AlertmanagerConfigs, 2) - require.NoError(t, n.ApplyConfig(cfg, map[string]http.Header{})) + require.NoError(t, n.ApplyConfig(cfg, nil)) require.Len(t, n.Alertmanagers(), 2) for cfgIdx := range 2 { ams := n.alertmanagers[fmt.Sprintf("config-%d", cfgIdx)].ams @@ -1169,6 +1170,6 @@ alerting: require.NoError(t, yaml.UnmarshalStrict([]byte(s), cfg)) require.Len(t, cfg.AlertingConfig.AlertmanagerConfigs, 2) - require.NoError(t, n.ApplyConfig(cfg, map[string]http.Header{})) + require.NoError(t, n.ApplyConfig(cfg, nil)) require.Empty(t, n.Alertmanagers()) } diff --git a/pkg/services/ngalert/sender/sender.go b/pkg/services/ngalert/sender/sender.go index e4bb96146ff..eb708a5d810 100644 --- a/pkg/services/ngalert/sender/sender.go +++ b/pkg/services/ngalert/sender/sender.go @@ -38,9 +38,9 @@ type ExternalAlertmanager struct { manager *Manager - sanitizeLabelSetFn func(lbls models.LabelSet) labels.Labels - sdCancel context.CancelFunc - sdManager *discovery.Manager + sdCancel context.CancelFunc + sdManager *discovery.Manager + options *ExternalAMOptions } type ExternalAMcfg struct { @@ -49,22 +49,27 @@ type ExternalAMcfg struct { Timeout time.Duration } -type Option func(*ExternalAlertmanager) +type ExternalAMOptions struct { + Options + sanitizeLabelSetFn func(lbls models.LabelSet) labels.Labels +} + +type Option func(*ExternalAMOptions) type doFunc func(context.Context, *http.Client, *http.Request) (*http.Response, error) // WithDoFunc receives a function to use when making HTTP requests from the Manager. func WithDoFunc(doFunc doFunc) Option { - return func(s *ExternalAlertmanager) { - s.manager.opts.Do = doFunc + return func(opts *ExternalAMOptions) { + opts.Do = doFunc } } // WithUTF8Labels skips sanitizing labels and annotations before sending alerts to the external Alertmanager(s). // It assumes UTF-8 label names are supported by the Alertmanager(s). func WithUTF8Labels() Option { - return func(s *ExternalAlertmanager) { - s.sanitizeLabelSetFn = func(lbls models.LabelSet) labels.Labels { + return func(opts *ExternalAMOptions) { + opts.sanitizeLabelSetFn = func(lbls models.LabelSet) labels.Labels { ls := make(labels.Labels, 0, len(lbls)) for k, v := range lbls { ls = append(ls, labels.Label{Name: k, Value: v}) @@ -74,6 +79,20 @@ func WithUTF8Labels() Option { } } +// WithMaxQueueCapacity sets the maximum capacity of the queue used by the sender. +func WithMaxQueueCapacity(capacity int) Option { + return func(opts *ExternalAMOptions) { + opts.QueueCapacity = capacity + } +} + +// WithMaxBatchSize sets the maximum batch size for sending alerts to the external Alertmanager(s). +func WithMaxBatchSize(size int) Option { + return func(opts *ExternalAMOptions) { + opts.MaxBatchSize = size + } +} + func (cfg *ExternalAMcfg) SHA256() string { return asSHA256([]string{cfg.headerString(), cfg.URL}) } @@ -99,16 +118,34 @@ func (cfg *ExternalAMcfg) headerString() string { func NewExternalAlertmanagerSender(l log.Logger, reg prometheus.Registerer, opts ...Option) (*ExternalAlertmanager, error) { sdCtx, sdCancel := context.WithCancel(context.Background()) + + options := &ExternalAMOptions{ + Options: Options{ + QueueCapacity: defaultMaxQueueCapacity, + MaxBatchSize: DefaultMaxBatchSize, + Registerer: reg, + DrainOnShutdown: defaultDrainOnShutdown, + }, + } + + for _, opt := range opts { + opt(options) + } + s := &ExternalAlertmanager{ logger: l, sdCancel: sdCancel, + options: options, + } + + if options.sanitizeLabelSetFn == nil { + options.sanitizeLabelSetFn = s.sanitizeLabelSet } - s.sanitizeLabelSetFn = s.sanitizeLabelSet s.manager = NewManager( // Injecting a new registry here means these metrics are not exported. // Once we fix the individual Alertmanager metrics we should fix this scenario too. - &Options{QueueCapacity: defaultMaxQueueCapacity, Registerer: reg, DrainOnShutdown: defaultDrainOnShutdown}, + &options.Options, toSlogLogger(s.logger), ) sdMetrics, err := discovery.CreateAndRegisterSDMetrics(prometheus.NewRegistry()) @@ -124,10 +161,6 @@ func NewExternalAlertmanagerSender(l log.Logger, reg prometheus.Registerer, opts return nil, errors.New("failed to create new discovery manager") } - for _, opt := range opts { - opt(s) - } - return s, nil } @@ -274,8 +307,8 @@ func buildNotifierConfig(alertmanagers []ExternalAMcfg) (*config.Config, map[str func (s *ExternalAlertmanager) alertToNotifierAlert(alert models.PostableAlert) *Alert { // Prometheus alertmanager has stricter rules for annotations/labels than grafana's internal alertmanager, so we sanitize invalid keys. return &Alert{ - Labels: s.sanitizeLabelSetFn(alert.Labels), - Annotations: s.sanitizeLabelSetFn(alert.Annotations), + Labels: s.options.sanitizeLabelSetFn(alert.Labels), + Annotations: s.options.sanitizeLabelSetFn(alert.Annotations), StartsAt: time.Time(alert.StartsAt), EndsAt: time.Time(alert.EndsAt), GeneratorURL: alert.GeneratorURL.String(), diff --git a/pkg/services/ngalert/sender/sender_test.go b/pkg/services/ngalert/sender/sender_test.go index b3a22bf51fc..1f51aeaf857 100644 --- a/pkg/services/ngalert/sender/sender_test.go +++ b/pkg/services/ngalert/sender/sender_test.go @@ -1,6 +1,7 @@ package sender import ( + "fmt" "testing" "github.com/prometheus/alertmanager/api/v2/models" @@ -108,3 +109,121 @@ func TestSanitizeLabelSet(t *testing.T) { }) } } + +func TestWithMaxQueueCapacity(t *testing.T) { + logger := log.NewNopLogger() + + t.Run("WithMaxQueueCapacity sets custom capacity", func(t *testing.T) { + customCapacity := 123 + am, err := NewExternalAlertmanagerSender(logger, prometheus.NewRegistry(), WithMaxQueueCapacity(customCapacity)) + require.NoError(t, err) + require.Equal(t, customCapacity, am.options.QueueCapacity) + }) + + t.Run("default capacity when option is not used", func(t *testing.T) { + am, err := NewExternalAlertmanagerSender(logger, prometheus.NewRegistry()) + require.NoError(t, err) + require.Equal(t, defaultMaxQueueCapacity, am.options.QueueCapacity) + }) + + t.Run("custom queue capacity is enforced", func(t *testing.T) { + customCapacity := 5 + am, err := NewExternalAlertmanagerSender(logger, prometheus.NewRegistry(), WithMaxQueueCapacity(customCapacity)) + require.NoError(t, err) + + totalAlerts := customCapacity + 3 + alerts := make([]*Alert, totalAlerts) + for i := range alerts { + alerts[i] = &Alert{ + Labels: labels.FromStrings("alertname", fmt.Sprintf("alert_%d", i)), + } + } + + am.manager.Send(alerts...) + + require.Equal(t, customCapacity, len(am.manager.queue)) + + for i, alert := range am.manager.queue { + expectedLabel := fmt.Sprintf("alert_%d", i+3) + require.Equal(t, expectedLabel, alert.Labels.Get("alertname")) + } + }) +} + +func TestWithMaxBatchSize(t *testing.T) { + logger := log.NewNopLogger() + + t.Run("WithMaxBatchSize sets custom batch size", func(t *testing.T) { + customBatchSize := 5 + am, err := NewExternalAlertmanagerSender(logger, prometheus.NewRegistry(), WithMaxBatchSize(customBatchSize)) + require.NoError(t, err) + require.Equal(t, customBatchSize, am.options.MaxBatchSize) + require.Equal(t, customBatchSize, am.manager.opts.MaxBatchSize) + }) + + t.Run("default batch size when option is not used", func(t *testing.T) { + am, err := NewExternalAlertmanagerSender(logger, prometheus.NewRegistry()) + require.NoError(t, err) + require.Equal(t, DefaultMaxBatchSize, am.options.MaxBatchSize) + require.Equal(t, DefaultMaxBatchSize, am.manager.opts.MaxBatchSize) + }) + + t.Run("custom batch size is enforced", func(t *testing.T) { + customBatchSize := 3 + am, err := NewExternalAlertmanagerSender(logger, prometheus.NewRegistry(), WithMaxBatchSize(customBatchSize)) + require.NoError(t, err) + + totalAlerts := customBatchSize * 2 + alerts := make([]*Alert, totalAlerts) + for i := range alerts { + alerts[i] = &Alert{ + Labels: labels.FromStrings("alertname", fmt.Sprintf("alert_%d", i)), + } + } + + am.manager.Send(alerts...) + require.Equal(t, totalAlerts, len(am.manager.queue)) + + firstBatch := am.manager.nextBatch() + require.Equal(t, customBatchSize, len(firstBatch)) + + secondBatch := am.manager.nextBatch() + require.Equal(t, customBatchSize, len(secondBatch)) + + emptyBatch := am.manager.nextBatch() + require.Equal(t, 0, len(emptyBatch), "No more alerts should remain") + }) +} + +func TestWithUTF8Labels(t *testing.T) { + logger := log.NewNopLogger() + + alert := models.PostableAlert{ + Annotations: models.LabelSet{ + "some-name": "test", + }, + Alert: models.Alert{ + Labels: models.LabelSet{ + "🔥": "fire", + }, + }, + } + + t.Run("WithUTF8Labels preserves UTF-8 characters", func(t *testing.T) { + am, err := NewExternalAlertmanagerSender(logger, prometheus.NewRegistry(), WithUTF8Labels()) + require.NoError(t, err) + + result := am.alertToNotifierAlert(alert) + require.Equal(t, "test", result.Annotations.Get("some-name")) + require.Equal(t, "fire", result.Labels.Get("🔥")) + }) + + t.Run("default sanitizes UTF-8 characters", func(t *testing.T) { + am, err := NewExternalAlertmanagerSender(logger, prometheus.NewRegistry()) + require.NoError(t, err) + + result := am.alertToNotifierAlert(alert) + require.Equal(t, "test", result.Annotations.Get("some_name")) + require.Equal(t, "fire", result.Labels.Get("_0x1f525")) + }) +}