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

[v9.5.x] Alerting: Support concurrent queries for saving alert instances (#70921)

Browse Source 
* Alerting: Support concurrent queries for saving alert instances (#70525)

This commit adds support for concurrent queries when saving alert
instances to the database. This is an experimental feature in
response to some customers experiencing delays between rule evaluation
and sending alerts to Alertmanager, resulting in flapping. It is
disabled by default.

(cherry picked from commit 7edbe72483)

* Trigger PR automation

---------

Co-authored-by: George Robinson <george.robinson@grafana.com>
This commit is contained in:
Santiago
2023-08-16 08:17:52 +01:00
committed by GitHub
co-authored by George Robinson
parent c1ef527b0b
commit 759dd42abf
9 changed files with 124 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
conf/defaults.ini
+5
View File
@@ -1005,6 +1005,11 @@ max_attempts = 3
# The interval string is a possibly signed sequence of decimal numbers, followed by a unit suffix (ms, s, m, h, d), e.g. 30s or 1m.
min_interval = 10s
# This is an experimental option to add parallelization to saving alert states in the database.
# It configures the maximum number of concurrent queries per rule evaluated. The default value is 1
# (concurrent queries per rule disabled).
max_state_save_concurrency = 1
[unified_alerting.screenshots]
# Enable screenshots in notifications. You must have either installed the Grafana image rendering
# plugin, or set up Grafana to use a remote rendering service.
pkg/services/ngalert/backtesting/engine.go
+7 -6
View File
@@ -46,12 +46,13 @@ func NewEngine(appUrl *url.URL, evalFactory eval.EvaluatorFactory) *Engine {
evalFactory: evalFactory,
createStateManager: func() stateManager {
cfg := state.ManagerCfg{
Metrics: nil,
ExternalURL: appUrl,
InstanceStore: nil,
Images: &NoopImageService{},
Clock: clock.New(),
Historian: nil,
Metrics: nil,
ExternalURL: appUrl,
InstanceStore: nil,
Images: &NoopImageService{},
Clock: clock.New(),
Historian: nil,
MaxStateSaveConcurrency: 1,
}
return state.NewManager(cfg)
},
pkg/services/ngalert/ngalert.go
+8 -7
View File
@@ -219,13 +219,14 @@ func (ng *AlertNG) init() error {
return err
}
cfg := state.ManagerCfg{
Metrics: ng.Metrics.GetStateMetrics(),
ExternalURL: appUrl,
InstanceStore: store,
Images: ng.imageService,
Clock: clk,
Historian: history,
DoNotSaveNormalState: ng.FeatureToggles.IsEnabled(featuremgmt.FlagAlertingNoNormalState),
Metrics: ng.Metrics.GetStateMetrics(),
ExternalURL: appUrl,
InstanceStore: ng.store,
Images: ng.imageService,
Clock: clk,
Historian: history,
DoNotSaveNormalState: ng.FeatureToggles.IsEnabled(featuremgmt.FlagAlertingNoNormalState),
MaxStateSaveConcurrency: ng.Cfg.UnifiedAlerting.MaxStateSaveConcurrency,
}
stateManager := state.NewManager(cfg)
scheduler := schedule.NewScheduler(schedCfg, stateManager)
pkg/services/ngalert/schedule/schedule_unit_test.go
+14 -12
View File
@@ -74,12 +74,13 @@ func TestProcessTicks(t *testing.T) {
Tracer: testTracer,
}
managerCfg := state.ManagerCfg{
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: nil,
Images: &state.NoopImageService{},
Clock: mockedClock,
Historian: &state.FakeHistorian{},
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: nil,
Images: &state.NoopImageService{},
Clock: mockedClock,
Historian: &state.FakeHistorian{},
MaxStateSaveConcurrency: 1,
}
st := state.NewManager(managerCfg)
@@ -817,12 +818,13 @@ func setupScheduler(t *testing.T, rs *fakeRulesStore, is *state.FakeInstanceStor
Tracer: testTracer,
}
managerCfg := state.ManagerCfg{
Metrics: m.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: is,
Images: &state.NoopImageService{},
Clock: mockedClock,
Historian: &state.FakeHistorian{},
Metrics: m.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: is,
Images: &state.NoopImageService{},
Clock: mockedClock,
Historian: &state.FakeHistorian{},
MaxStateSaveConcurrency: 1,
}
st := state.NewManager(managerCfg)
pkg/services/ngalert/state/manager.go
+26 -16
View File
@@ -6,6 +6,7 @@ import (
"time"
"github.com/benbjohnson/clock"
"github.com/grafana/dskit/concurrency"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana/pkg/infra/log"
@@ -39,7 +40,8 @@ type Manager struct {
historian Historian
externalURL *url.URL
doNotSaveNormalState bool
doNotSaveNormalState bool
maxStateSaveConcurrency int
}
type ManagerCfg struct {
@@ -51,20 +53,23 @@ type ManagerCfg struct {
Historian Historian
// DoNotSaveNormalState controls whether eval.Normal state is persisted to the database and returned by get methods
DoNotSaveNormalState bool
// MaxStateSaveConcurrency controls the number of goroutines (per rule) that can save alert state in parallel.
MaxStateSaveConcurrency int
}
func NewManager(cfg ManagerCfg) *Manager {
return &Manager{
cache: newCache(),
ResendDelay: ResendDelay, // TODO: make this configurable
log: log.New("ngalert.state.manager"),
metrics: cfg.Metrics,
instanceStore: cfg.InstanceStore,
images: cfg.Images,
historian: cfg.Historian,
clock: cfg.Clock,
externalURL: cfg.ExternalURL,
doNotSaveNormalState: cfg.DoNotSaveNormalState,
cache: newCache(),
ResendDelay: ResendDelay, // TODO: make this configurable
log: log.New("ngalert.state.manager"),
metrics: cfg.Metrics,
instanceStore: cfg.InstanceStore,
images: cfg.Images,
historian: cfg.Historian,
clock: cfg.Clock,
externalURL: cfg.ExternalURL,
doNotSaveNormalState: cfg.DoNotSaveNormalState,
maxStateSaveConcurrency: cfg.MaxStateSaveConcurrency,
}
}
@@ -350,17 +355,17 @@ func (st *Manager) saveAlertStates(ctx context.Context, logger log.Logger, state
return
}
logger.Debug("Saving alert states", "count", len(states))
for _, s := range states {
saveState := func(ctx context.Context, idx int) error {
s := states[idx]
// Do not save normal state to database and remove transition to Normal state but keep mapped states
if st.doNotSaveNormalState && IsNormalStateWithNoReason(s.State) && !s.Changed() {
continue
return nil
}
key, err := s.GetAlertInstanceKey()
if err != nil {
logger.Error("Failed to create a key for alert state to save it to database. The state will be ignored ", "cacheID", s.CacheID, "error", err, "labels", s.Labels.String())
continue
return nil
}
instance := ngModels.AlertInstance{
AlertInstanceKey: key,
@@ -375,9 +380,14 @@ func (st *Manager) saveAlertStates(ctx context.Context, logger log.Logger, state
err = st.instanceStore.SaveAlertInstance(ctx, instance)
if err != nil {
logger.Error("Failed to save alert state", "labels", s.Labels.String(), "state", s.State, "error", err)
return nil
}
return nil
}
logger.Debug("Saving alert states done", "count", len(states))
logger.Debug("Saving alert states", "count", len(states), "max_state_save_concurrency", st.maxStateSaveConcurrency)
_ = concurrency.ForEachJob(ctx, len(states), st.maxStateSaveConcurrency, saveState)
logger.Debug("Saving alert states done", "count", len(states), "max_state_save_concurrency", st.maxStateSaveConcurrency)
}
func (st *Manager) deleteAlertStates(ctx context.Context, logger log.Logger, states []StateTransition) {
pkg/services/ngalert/state/manager_bench_test.go
+2 -1
View File
@@ -22,7 +22,8 @@ func BenchmarkProcessEvalResults(b *testing.B) {
metrics := metrics.NewHistorianMetrics(prometheus.NewRegistry())
hist := historian.NewAnnotationBackend(&as, nil, nil, metrics)
cfg := state.ManagerCfg{
Historian: hist,
Historian: hist,
MaxStateSaveConcurrency: 1,
}
sut := state.NewManager(cfg)
now := time.Now().UTC()
pkg/services/ngalert/state/manager_private_test.go
+2 -2
View File
@@ -114,7 +114,7 @@ func TestManager_saveAlertStates(t *testing.T) {
t.Run("should save all transitions if doNotSaveNormalState is false", func(t *testing.T) {
st := &FakeInstanceStore{}
m := Manager{instanceStore: st, doNotSaveNormalState: false}
m := Manager{instanceStore: st, doNotSaveNormalState: false, maxStateSaveConcurrency: 1}
m.saveAlertStates(context.Background(), &logtest.Fake{}, transitions...)
savedKeys := map[ngmodels.AlertInstanceKey]ngmodels.AlertInstance{}
@@ -131,7 +131,7 @@ func TestManager_saveAlertStates(t *testing.T) {
t.Run("should not save Normal->Normal if doNotSaveNormalState is true", func(t *testing.T) {
st := &FakeInstanceStore{}
m := Manager{instanceStore: st, doNotSaveNormalState: true}
m := Manager{instanceStore: st, doNotSaveNormalState: true, maxStateSaveConcurrency: 1}
m.saveAlertStates(context.Background(), &logtest.Fake{}, transitions...)
savedKeys := map[ngmodels.AlertInstanceKey]ngmodels.AlertInstance{}
pkg/services/ngalert/state/manager_test.go
+56 -48
View File
@@ -194,12 +194,13 @@ func TestWarmStateCache(t *testing.T) {
}
cfg := state.ManagerCfg{
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: dbstore,
Images: &state.NoopImageService{},
Clock: clock.NewMock(),
Historian: &state.FakeHistorian{},
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: dbstore,
Images: &state.NoopImageService{},
Clock: clock.NewMock(),
Historian: &state.FakeHistorian{},
MaxStateSaveConcurrency: 1,
}
st := state.NewManager(cfg)
st.Warm(ctx, dbstore)
@@ -227,12 +228,13 @@ func TestDashboardAnnotations(t *testing.T) {
metrics := metrics.NewHistorianMetrics(prometheus.NewRegistry())
hist := historian.NewAnnotationBackend(fakeAnnoRepo, &dashboards.FakeDashboardService{}, nil, metrics)
cfg := state.ManagerCfg{
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: dbstore,
Images: &state.NoopImageService{},
Clock: clock.New(),
Historian: hist,
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: dbstore,
Images: &state.NoopImageService{},
Clock: clock.New(),
Historian: hist,
MaxStateSaveConcurrency: 1,
}
st := state.NewManager(cfg)
@@ -2256,12 +2258,13 @@ func TestProcessEvalResults(t *testing.T) {
metrics := metrics.NewHistorianMetrics(prometheus.NewRegistry())
hist := historian.NewAnnotationBackend(fakeAnnoRepo, &dashboards.FakeDashboardService{}, nil, metrics)
cfg := state.ManagerCfg{
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: &state.FakeInstanceStore{},
Images: &state.NotAvailableImageService{},
Clock: clock.New(),
Historian: hist,
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: &state.FakeInstanceStore{},
Images: &state.NotAvailableImageService{},
Clock: clock.New(),
Historian: hist,
MaxStateSaveConcurrency: 1,
}
st := state.NewManager(cfg)
t.Run(tc.desc, func(t *testing.T) {
@@ -2291,12 +2294,13 @@ func TestProcessEvalResults(t *testing.T) {
instanceStore := &state.FakeInstanceStore{}
clk := clock.New()
cfg := state.ManagerCfg{
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: instanceStore,
Images: &state.NotAvailableImageService{},
Clock: clk,
Historian: &state.FakeHistorian{},
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: instanceStore,
Images: &state.NotAvailableImageService{},
Clock: clk,
Historian: &state.FakeHistorian{},
MaxStateSaveConcurrency: 1,
}
st := state.NewManager(cfg)
rule := models.AlertRuleGen()()
@@ -2432,12 +2436,13 @@ func TestStaleResultsHandler(t *testing.T) {
for _, tc := range testCases {
ctx := context.Background()
cfg := state.ManagerCfg{
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: dbstore,
Images: &state.NoopImageService{},
Clock: clock.New(),
Historian: &state.FakeHistorian{},
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: dbstore,
Images: &state.NoopImageService{},
Clock: clock.New(),
Historian: &state.FakeHistorian{},
MaxStateSaveConcurrency: 1,
}
st := state.NewManager(cfg)
st.Warm(ctx, dbstore)
@@ -2511,12 +2516,13 @@ func TestStaleResults(t *testing.T) {
store := &state.FakeInstanceStore{}
cfg := state.ManagerCfg{
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: store,
Images: &state.NoopImageService{},
Clock: clk,
Historian: &state.FakeHistorian{},
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: store,
Images: &state.NoopImageService{},
Clock: clk,
Historian: &state.FakeHistorian{},
MaxStateSaveConcurrency: 1,
}
st := state.NewManager(cfg)
@@ -2678,12 +2684,13 @@ func TestDeleteStateByRuleUID(t *testing.T) {
clk := clock.NewMock()
clk.Set(time.Now())
cfg := state.ManagerCfg{
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: dbstore,
Images: &state.NoopImageService{},
Clock: clk,
Historian: &state.FakeHistorian{},
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: dbstore,
Images: &state.NoopImageService{},
Clock: clk,
Historian: &state.FakeHistorian{},
MaxStateSaveConcurrency: 1,
}
st := state.NewManager(cfg)
st.Warm(ctx, dbstore)
@@ -2817,12 +2824,13 @@ func TestResetStateByRuleUID(t *testing.T) {
clk := clock.NewMock()
clk.Set(time.Now())
cfg := state.ManagerCfg{
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: dbstore,
Images: &state.NoopImageService{},
Clock: clk,
Historian: fakeHistorian,
Metrics: testMetrics.GetStateMetrics(),
ExternalURL: nil,
InstanceStore: dbstore,
Images: &state.NoopImageService{},
Clock: clk,
Historian: fakeHistorian,
MaxStateSaveConcurrency: 1,
}
st := state.NewManager(cfg)
st.Warm(ctx, dbstore)
pkg/setting/setting_unified_alerting.go
+4
View File
@@ -86,6 +86,8 @@ type UnifiedAlertingSettings struct {
Screenshots UnifiedAlertingScreenshotSettings
ReservedLabels UnifiedAlertingReservedLabelSettings
StateHistory UnifiedAlertingStateHistorySettings
// MaxStateSaveConcurrency controls the number of goroutines (per rule) that can save alert state in parallel.
MaxStateSaveConcurrency int
}
type UnifiedAlertingScreenshotSettings struct {
@@ -338,6 +340,8 @@ func (cfg *Cfg) ReadUnifiedAlertingSettings(iniFile *ini.File) error {
}
uaCfg.StateHistory = uaCfgStateHistory
uaCfg.MaxStateSaveConcurrency = ua.Key("max_state_save_concurrency").MustInt(1)
cfg.UnifiedAlerting = uaCfg
return nil
}