public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Alerting: Configurable queue and batch size in external alert manager notifier (#109831)

Browse Source
This commit is contained in:
Alexander Akhmetov
2025-08-19 09:32:10 +02:00
committed by GitHub
parent dff7050aaa
commit bdd5bf0397
5 changed files with 205 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/ngalert/sender/notifier.go
+12 -12
View File
@@ -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 {
pkg/services/ngalert/sender/notifier_ext.go
+9
View File
@@ -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 {
pkg/services/ngalert/sender/notifier_test.go
+16 -15
View File
@@ -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())
}
pkg/services/ngalert/sender/sender.go
+49 -16
View File
@@ -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(),
pkg/services/ngalert/sender/sender_test.go
+119
View File
@@ -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"))
})
}