Alerting: Support retry with backoff in alert rule evaluation (#99710)
This commit is contained in:
@@ -321,7 +321,12 @@ func (ng *AlertNG) init() error {
|
||||
ng.RecordingWriter = recordingWriter
|
||||
|
||||
schedCfg := schedule.SchedulerCfg{
|
||||
MaxAttempts: ng.Cfg.UnifiedAlerting.MaxAttempts,
|
||||
RetryConfig: schedule.RetryConfig{
|
||||
MaxAttempts: ng.Cfg.UnifiedAlerting.MaxAttempts,
|
||||
InitialRetryDelay: ng.Cfg.UnifiedAlerting.InitialRetryDelay,
|
||||
MaxRetryDelay: ng.Cfg.UnifiedAlerting.MaxRetryDelay,
|
||||
RandomizationFactor: ng.Cfg.UnifiedAlerting.RandomizationFactor,
|
||||
},
|
||||
C: clk,
|
||||
BaseInterval: ng.Cfg.UnifiedAlerting.BaseInterval,
|
||||
MinRuleInterval: ng.Cfg.UnifiedAlerting.MinInterval,
|
||||
|
||||
@@ -56,7 +56,7 @@ func (f ruleFactoryFunc) new(ctx context.Context, rule *ngmodels.AlertRule) Rule
|
||||
func newRuleFactory(
|
||||
appURL *url.URL,
|
||||
disableGrafanaFolder bool,
|
||||
maxAttempts int64,
|
||||
retryConfig RetryConfig,
|
||||
sender AlertsSender,
|
||||
stateManager *state.Manager,
|
||||
evalFactory eval.EvaluatorFactory,
|
||||
@@ -75,7 +75,7 @@ func newRuleFactory(
|
||||
return newRecordingRule(
|
||||
ctx,
|
||||
rule.GetKeyWithGroup(),
|
||||
maxAttempts,
|
||||
retryConfig,
|
||||
clock,
|
||||
evalFactory,
|
||||
rrCfg,
|
||||
@@ -92,7 +92,7 @@ func newRuleFactory(
|
||||
rule.GetKeyWithGroup(),
|
||||
appURL,
|
||||
disableGrafanaFolder,
|
||||
maxAttempts,
|
||||
retryConfig,
|
||||
sender,
|
||||
stateManager,
|
||||
evalFactory,
|
||||
@@ -120,7 +120,7 @@ type alertRule struct {
|
||||
|
||||
appURL *url.URL
|
||||
disableGrafanaFolder bool
|
||||
maxAttempts int64
|
||||
retryConfig RetryConfig
|
||||
|
||||
clock clock.Clock
|
||||
sender AlertsSender
|
||||
@@ -142,7 +142,7 @@ func newAlertRule(
|
||||
key ngmodels.AlertRuleKeyWithGroup,
|
||||
appURL *url.URL,
|
||||
disableGrafanaFolder bool,
|
||||
maxAttempts int64,
|
||||
retryConfig RetryConfig,
|
||||
sender AlertsSender,
|
||||
stateManager *state.Manager,
|
||||
evalFactory eval.EvaluatorFactory,
|
||||
@@ -155,6 +155,7 @@ func newAlertRule(
|
||||
stopAppliedHook func(ngmodels.AlertRuleKey),
|
||||
) *alertRule {
|
||||
ctx, stop := util.WithCancelCause(ngmodels.WithRuleKey(parent, key.AlertRuleKey))
|
||||
|
||||
return &alertRule{
|
||||
key: key,
|
||||
evalCh: make(chan *Evaluation),
|
||||
@@ -163,7 +164,7 @@ func newAlertRule(
|
||||
stopFn: stop,
|
||||
appURL: appURL,
|
||||
disableGrafanaFolder: disableGrafanaFolder,
|
||||
maxAttempts: maxAttempts,
|
||||
retryConfig: retryConfig,
|
||||
clock: clock,
|
||||
sender: sender,
|
||||
stateManager: stateManager,
|
||||
@@ -271,6 +272,14 @@ func (a *alertRule) Run() error {
|
||||
logger := a.logger.New("version", ctx.rule.Version, "fingerprint", f, "now", ctx.scheduledAt)
|
||||
logger.Debug("Processing tick")
|
||||
|
||||
retryer := newExponentialBackoffRetryer(
|
||||
a.retryConfig.MaxAttempts-1, // First attempt is not a retry.
|
||||
a.retryConfig.InitialRetryDelay,
|
||||
a.retryConfig.MaxRetryDelay,
|
||||
a.retryConfig.RandomizationFactor,
|
||||
a.clock,
|
||||
)
|
||||
|
||||
func() {
|
||||
orgID := fmt.Sprint(a.key.OrgID)
|
||||
evalDuration := a.metrics.EvalDuration.WithLabelValues(orgID)
|
||||
@@ -282,7 +291,8 @@ func (a *alertRule) Run() error {
|
||||
a.evalApplied(ctx.scheduledAt)
|
||||
}()
|
||||
|
||||
for attempt := int64(1); attempt <= a.maxAttempts; attempt++ {
|
||||
attempt := 1
|
||||
for {
|
||||
isPaused := ctx.rule.IsPaused
|
||||
|
||||
// Do not clean up state if the eval loop has just started.
|
||||
@@ -327,8 +337,9 @@ func (a *alertRule) Run() error {
|
||||
logger.Error("Skip evaluation and updating the state because the context has been cancelled", "version", ctx.rule.Version, "fingerprint", f, "attempt", attempt, "now", ctx.scheduledAt)
|
||||
return
|
||||
}
|
||||
retry := attempt < a.maxAttempts
|
||||
err := a.evaluate(tracingCtx, ctx, span, retry, logger)
|
||||
nextDelay := retryer.NextAttemptIn()
|
||||
shouldRetry := nextDelay != retryStop
|
||||
err := a.evaluate(tracingCtx, ctx, span, shouldRetry, logger)
|
||||
// This is extremely confusing - when we exhaust all retry attempts, or we have no retryable errors
|
||||
// we return nil - so technically, this is meaningless to know whether the evaluation has errors or not.
|
||||
span.End()
|
||||
@@ -337,12 +348,14 @@ func (a *alertRule) Run() error {
|
||||
return
|
||||
}
|
||||
|
||||
logger.Error("Failed to evaluate rule", "attempt", attempt, "error", err)
|
||||
logger.Error("Failed to evaluate rule", "attempt", attempt, "max_attempts", a.retryConfig.MaxAttempts, "next_attempt_in", nextDelay, "error", err)
|
||||
attempt++
|
||||
|
||||
select {
|
||||
case <-tracingCtx.Done():
|
||||
logger.Error("Context has been cancelled while backing off", "attempt", attempt)
|
||||
return
|
||||
case <-time.After(retryDelay):
|
||||
case <-a.clock.After(nextDelay):
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/benbjohnson/clock"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/data"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/testutil"
|
||||
@@ -502,7 +503,7 @@ func blankRuleForTests(ctx context.Context, key models.AlertRuleKeyWithGroup) *a
|
||||
Log: log.NewNopLogger(),
|
||||
}
|
||||
st := state.NewManager(managerCfg, state.NewNoopPersister())
|
||||
return newAlertRule(ctx, key, nil, false, 0, nil, st, nil, nil, nil, log.NewNopLogger(), nil, featuremgmt.WithFeatures(), nil, nil)
|
||||
return newAlertRule(ctx, key, nil, false, RetryConfig{}, nil, st, nil, nil, nil, log.NewNopLogger(), nil, featuremgmt.WithFeatures(), nil, nil)
|
||||
}
|
||||
|
||||
func TestRuleRoutine(t *testing.T) {
|
||||
@@ -510,12 +511,13 @@ func TestRuleRoutine(t *testing.T) {
|
||||
createSchedule := func(
|
||||
evalAppliedChan chan time.Time,
|
||||
senderMock *SyncAlertsSenderMock,
|
||||
clk clock.Clock,
|
||||
) (*schedule, *fakeRulesStore, *state.FakeInstanceStore, prometheus.Gatherer) {
|
||||
ruleStore := newFakeRulesStore()
|
||||
instanceStore := &state.FakeInstanceStore{}
|
||||
|
||||
registry := prometheus.NewPedanticRegistry()
|
||||
sch := setupScheduler(t, ruleStore, instanceStore, registry, senderMock, nil, nil)
|
||||
sch := setupScheduler(t, ruleStore, instanceStore, registry, senderMock, nil, nil, withSchedulerClock(clk))
|
||||
sch.evalAppliedFunc = func(key models.AlertRuleKey, t time.Time) {
|
||||
evalAppliedChan <- t
|
||||
}
|
||||
@@ -530,7 +532,7 @@ func TestRuleRoutine(t *testing.T) {
|
||||
// TODO rewrite when we are able to mock/fake state manager
|
||||
t.Run(fmt.Sprintf("when rule evaluation happens (evaluation state %s)", evalState), func(t *testing.T) {
|
||||
evalAppliedChan := make(chan time.Time)
|
||||
sch, ruleStore, instanceStore, reg := createSchedule(evalAppliedChan, nil)
|
||||
sch, ruleStore, instanceStore, reg := createSchedule(evalAppliedChan, nil, clock.NewMock())
|
||||
|
||||
rule := gen.With(withQueryForState(t, evalState)).GenerateRef()
|
||||
ruleStore.PutRule(context.Background(), rule)
|
||||
@@ -718,7 +720,7 @@ func TestRuleRoutine(t *testing.T) {
|
||||
t.Run("and clean up the state if parent context is cancelled", func(t *testing.T) {
|
||||
stoppedChan := make(chan error)
|
||||
sender := NewSyncAlertsSenderMock()
|
||||
sch, _, _, _ := createSchedule(make(chan time.Time), sender)
|
||||
sch, _, _, _ := createSchedule(make(chan time.Time), sender, clock.NewMock())
|
||||
|
||||
_ = sch.stateManager.ProcessEvalResults(context.Background(), sch.clock.Now(), rule, genEvalResults(sch.clock.Now()), nil, nil)
|
||||
expectedStates := sch.stateManager.GetStatesForRuleUID(rule.OrgID, rule.UID)
|
||||
@@ -742,7 +744,7 @@ func TestRuleRoutine(t *testing.T) {
|
||||
t.Run("and clean up the state but not send anything if the reason is not rule deleted", func(t *testing.T) {
|
||||
stoppedChan := make(chan error)
|
||||
sender := NewSyncAlertsSenderMock()
|
||||
sch, _, _, _ := createSchedule(make(chan time.Time), sender)
|
||||
sch, _, _, _ := createSchedule(make(chan time.Time), sender, clock.NewMock())
|
||||
|
||||
_ = sch.stateManager.ProcessEvalResults(context.Background(), sch.clock.Now(), rule, genEvalResults(sch.clock.Now()), nil, nil)
|
||||
require.NotEmpty(t, sch.stateManager.GetStatesForRuleUID(rule.OrgID, rule.UID))
|
||||
@@ -766,7 +768,7 @@ func TestRuleRoutine(t *testing.T) {
|
||||
stoppedChan := make(chan error)
|
||||
sender := NewSyncAlertsSenderMock()
|
||||
sender.EXPECT().Send(mock.Anything, mock.Anything, mock.Anything).Times(1)
|
||||
sch, _, _, _ := createSchedule(make(chan time.Time), sender)
|
||||
sch, _, _, _ := createSchedule(make(chan time.Time), sender, clock.NewMock())
|
||||
|
||||
_ = sch.stateManager.ProcessEvalResults(context.Background(), sch.clock.Now(), rule, genEvalResults(sch.clock.Now()), nil, nil)
|
||||
require.NotEmpty(t, sch.stateManager.GetStatesForRuleUID(rule.OrgID, rule.UID))
|
||||
@@ -796,7 +798,7 @@ func TestRuleRoutine(t *testing.T) {
|
||||
sender := NewSyncAlertsSenderMock()
|
||||
sender.EXPECT().Send(mock.Anything, rule.GetKey(), mock.Anything).Return()
|
||||
|
||||
sch, ruleStore, _, _ := createSchedule(evalAppliedChan, sender)
|
||||
sch, ruleStore, _, _ := createSchedule(evalAppliedChan, sender, clock.NewMock())
|
||||
ruleStore.PutRule(context.Background(), rule)
|
||||
sch.schedulableAlertRules.set([]*models.AlertRule{rule}, map[models.FolderKey]string{rule.GetFolderKey(): folderTitle})
|
||||
factory := ruleFactoryFromScheduler(sch)
|
||||
@@ -878,10 +880,34 @@ func TestRuleRoutine(t *testing.T) {
|
||||
sender := NewSyncAlertsSenderMock()
|
||||
sender.EXPECT().Send(mock.Anything, rule.GetKey(), mock.Anything).Return()
|
||||
|
||||
sch, ruleStore, _, reg := createSchedule(evalAppliedChan, sender)
|
||||
sch.maxAttempts = 3
|
||||
clk := clock.NewMock()
|
||||
sch, ruleStore, _, reg := createSchedule(evalAppliedChan, sender, clk)
|
||||
sch.retryConfig = RetryConfig{
|
||||
MaxAttempts: 3,
|
||||
InitialRetryDelay: 1 * time.Second,
|
||||
MaxRetryDelay: 1 * time.Second,
|
||||
RandomizationFactor: 0,
|
||||
}
|
||||
ruleStore.PutRule(context.Background(), rule)
|
||||
factory := ruleFactoryFromScheduler(sch)
|
||||
|
||||
factory := newRuleFactory(
|
||||
sch.appURL,
|
||||
sch.disableGrafanaFolder,
|
||||
sch.retryConfig,
|
||||
sch.alertsSender,
|
||||
sch.stateManager,
|
||||
sch.evaluatorFactory,
|
||||
sch.clock,
|
||||
sch.rrCfg,
|
||||
sch.metrics,
|
||||
sch.log,
|
||||
sch.tracer,
|
||||
sch.featureToggles,
|
||||
sch.recordingWriter,
|
||||
sch.evalAppliedFunc,
|
||||
sch.stopAppliedFunc,
|
||||
)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
t.Cleanup(cancel)
|
||||
ruleInfo := factory.new(ctx, rule)
|
||||
@@ -895,6 +921,12 @@ func TestRuleRoutine(t *testing.T) {
|
||||
rule: rule,
|
||||
})
|
||||
|
||||
// Because we are using a mock clock, first we need to wait until the rule evaluation
|
||||
// reaches the point where it sleeps for the duration of the retry interval.
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
// Then advance the mock clock to trigger the retry.
|
||||
clk.Add(2 * time.Second)
|
||||
|
||||
waitForTimeChannel(t, evalAppliedChan)
|
||||
|
||||
t.Run("it should increase failure counter by 1 and attempt failure counter by 3", func(t *testing.T) {
|
||||
@@ -902,10 +934,10 @@ func TestRuleRoutine(t *testing.T) {
|
||||
expectedMetric := fmt.Sprintf(
|
||||
`# HELP grafana_alerting_rule_evaluation_duration_seconds The time to evaluate a rule.
|
||||
# TYPE grafana_alerting_rule_evaluation_duration_seconds histogram
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="0.01"} 1
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="0.1"} 1
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="0.5"} 1
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="1"} 1
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="0.01"} 0
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="0.1"} 0
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="0.5"} 0
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="1"} 0
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="5"} 1
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="10"} 1
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="15"} 1
|
||||
@@ -916,7 +948,7 @@ func TestRuleRoutine(t *testing.T) {
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="240"} 1
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="300"} 1
|
||||
grafana_alerting_rule_evaluation_duration_seconds_bucket{org="%[1]d",le="+Inf"} 1
|
||||
grafana_alerting_rule_evaluation_duration_seconds_sum{org="%[1]d"} 0
|
||||
grafana_alerting_rule_evaluation_duration_seconds_sum{org="%[1]d"} 2
|
||||
grafana_alerting_rule_evaluation_duration_seconds_count{org="%[1]d"} 1
|
||||
# HELP grafana_alerting_rule_evaluation_failures_total The total number of rule evaluation failures.
|
||||
# TYPE grafana_alerting_rule_evaluation_failures_total counter
|
||||
@@ -1007,7 +1039,7 @@ func TestRuleRoutine(t *testing.T) {
|
||||
sender := NewSyncAlertsSenderMock()
|
||||
sender.EXPECT().Send(mock.Anything, rule.GetKey(), mock.Anything).Return()
|
||||
|
||||
sch, ruleStore, _, _ := createSchedule(evalAppliedChan, sender)
|
||||
sch, ruleStore, _, _ := createSchedule(evalAppliedChan, sender, clock.NewMock())
|
||||
ruleStore.PutRule(context.Background(), rule)
|
||||
factory := ruleFactoryFromScheduler(sch)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
@@ -1041,7 +1073,7 @@ func TestRuleRoutine(t *testing.T) {
|
||||
sender := NewSyncAlertsSenderMock()
|
||||
sender.EXPECT().Send(mock.Anything, rule.GetKey(), mock.Anything).Return()
|
||||
|
||||
sch, ruleStore, _, _ := createSchedule(evalAppliedChan, sender)
|
||||
sch, ruleStore, _, _ := createSchedule(evalAppliedChan, sender, clock.NewMock())
|
||||
ruleStore.PutRule(context.Background(), rule)
|
||||
factory := ruleFactoryFromScheduler(sch)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
@@ -1076,7 +1108,7 @@ func TestRuleRoutine(t *testing.T) {
|
||||
sender := NewSyncAlertsSenderMock()
|
||||
sender.EXPECT().Send(mock.Anything, rule.GetKey(), mock.Anything).Return()
|
||||
|
||||
sch, ruleStore, _, _ := createSchedule(evalAppliedChan, sender)
|
||||
sch, ruleStore, _, _ := createSchedule(evalAppliedChan, sender, clock.NewMock())
|
||||
sch.stateManager.ResolvedRetention = 4 * time.Second
|
||||
sch.stateManager.ResendDelay = 2 * time.Second
|
||||
sch.stateManager.Put([]*state.State{
|
||||
@@ -1125,8 +1157,147 @@ func TestRuleRoutine(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestAlertRuleRetry(t *testing.T) {
|
||||
gen := models.RuleGen
|
||||
createSchedule := func(
|
||||
evalAppliedChan chan time.Time,
|
||||
senderMock *SyncAlertsSenderMock,
|
||||
) (*schedule, *fakeRulesStore, *state.FakeInstanceStore, prometheus.Gatherer) {
|
||||
ruleStore := newFakeRulesStore()
|
||||
instanceStore := &state.FakeInstanceStore{}
|
||||
|
||||
registry := prometheus.NewPedanticRegistry()
|
||||
sch := setupScheduler(t, ruleStore, instanceStore, registry, senderMock, nil, nil)
|
||||
sch.evalAppliedFunc = func(key models.AlertRuleKey, t time.Time) {
|
||||
evalAppliedChan <- t
|
||||
}
|
||||
return sch, ruleStore, instanceStore, registry
|
||||
}
|
||||
|
||||
evalAppliedChan := make(chan time.Time)
|
||||
|
||||
rule := gen.With(withQueryForState(t, eval.Error)).GenerateRef()
|
||||
rule.ExecErrState = models.ErrorErrState
|
||||
|
||||
sender := NewSyncAlertsSenderMock()
|
||||
sender.EXPECT().Send(mock.Anything, rule.GetKey(), mock.Anything).Return()
|
||||
|
||||
sch, ruleStore, _, reg := createSchedule(evalAppliedChan, sender)
|
||||
fakeClock := sch.clock.(*clock.Mock)
|
||||
|
||||
ruleStore.PutRule(context.Background(), rule)
|
||||
|
||||
maxAttempts := int64(3)
|
||||
backoffDuration := time.Millisecond * 10
|
||||
|
||||
factory := newRuleFactory(
|
||||
sch.appURL,
|
||||
sch.disableGrafanaFolder,
|
||||
RetryConfig{
|
||||
MaxAttempts: maxAttempts,
|
||||
InitialRetryDelay: backoffDuration,
|
||||
MaxRetryDelay: backoffDuration,
|
||||
},
|
||||
sch.alertsSender,
|
||||
sch.stateManager,
|
||||
sch.evaluatorFactory,
|
||||
fakeClock,
|
||||
sch.rrCfg,
|
||||
sch.metrics,
|
||||
sch.log,
|
||||
sch.tracer,
|
||||
sch.featureToggles,
|
||||
sch.recordingWriter,
|
||||
sch.evalAppliedFunc,
|
||||
sch.stopAppliedFunc,
|
||||
)
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
t.Cleanup(cancel)
|
||||
ruleInfo := factory.new(ctx, rule)
|
||||
|
||||
go func() {
|
||||
_ = ruleInfo.Run()
|
||||
}()
|
||||
|
||||
// Run the rule evaluation tick
|
||||
ruleInfo.Eval(&Evaluation{
|
||||
scheduledAt: sch.clock.Now(),
|
||||
rule: rule,
|
||||
})
|
||||
|
||||
compareMetrics := func(c *assert.CollectT, evaluations, expectedFailures int) {
|
||||
expectedMetric := fmt.Sprintf(
|
||||
`# HELP grafana_alerting_rule_evaluation_attempts_total The total number of rule evaluation attempts.
|
||||
# TYPE grafana_alerting_rule_evaluation_attempts_total counter
|
||||
grafana_alerting_rule_evaluation_attempts_total{org="%[1]d"} %[3]d
|
||||
# HELP grafana_alerting_rule_evaluation_attempt_failures_total The total number of rule evaluation attempt failures.
|
||||
# TYPE grafana_alerting_rule_evaluation_attempt_failures_total counter
|
||||
grafana_alerting_rule_evaluation_attempt_failures_total{org="%[1]d"} %[3]d
|
||||
# HELP grafana_alerting_rule_evaluations_total The total number of rule evaluations.
|
||||
# TYPE grafana_alerting_rule_evaluations_total counter
|
||||
grafana_alerting_rule_evaluations_total{org="%[1]d"} %[2]d
|
||||
`, rule.OrgID, evaluations, expectedFailures)
|
||||
|
||||
err := testutil.GatherAndCompare(
|
||||
reg,
|
||||
bytes.NewBufferString(expectedMetric),
|
||||
"grafana_alerting_rule_evaluations_total",
|
||||
"grafana_alerting_rule_evaluation_attempts_total",
|
||||
"grafana_alerting_rule_evaluation_attempt_failures_total",
|
||||
)
|
||||
assert.NoError(c, err)
|
||||
}
|
||||
|
||||
t.Run("first attempt", func(t *testing.T) {
|
||||
require.EventuallyWithT(t, func(c *assert.CollectT) {
|
||||
compareMetrics(c, 1, 1)
|
||||
}, 5*time.Millisecond, 1*time.Millisecond)
|
||||
})
|
||||
|
||||
t.Run("second attempt", func(t *testing.T) {
|
||||
// advance the clock by the backoff duration
|
||||
fakeClock.Add(backoffDuration)
|
||||
require.EventuallyWithT(t, func(c *assert.CollectT) {
|
||||
compareMetrics(c, 1, 2)
|
||||
}, 5*time.Millisecond, 1*time.Millisecond)
|
||||
})
|
||||
|
||||
t.Run("third attempt", func(t *testing.T) {
|
||||
// advance the clock by the backoff duration
|
||||
fakeClock.Add(backoffDuration)
|
||||
require.EventuallyWithT(t, func(c *assert.CollectT) {
|
||||
compareMetrics(c, 1, 3)
|
||||
}, 5*time.Millisecond, 1*time.Millisecond)
|
||||
})
|
||||
|
||||
t.Run("no fourth attempt", func(t *testing.T) {
|
||||
// Wait long enough to ensure no fourth attempt occurs
|
||||
fakeClock.Add(backoffDuration * 10)
|
||||
require.EventuallyWithT(t, func(c *assert.CollectT) {
|
||||
compareMetrics(c, 1, 3)
|
||||
}, 5*time.Millisecond, 1*time.Millisecond)
|
||||
})
|
||||
}
|
||||
|
||||
func ruleFactoryFromScheduler(sch *schedule) ruleFactory {
|
||||
return newRuleFactory(sch.appURL, sch.disableGrafanaFolder, sch.maxAttempts, sch.alertsSender, sch.stateManager, sch.evaluatorFactory, sch.clock, sch.rrCfg, sch.metrics, sch.log, sch.tracer, sch.featureToggles, sch.recordingWriter, sch.evalAppliedFunc, sch.stopAppliedFunc)
|
||||
return newRuleFactory(
|
||||
sch.appURL,
|
||||
sch.disableGrafanaFolder,
|
||||
sch.retryConfig,
|
||||
sch.alertsSender,
|
||||
sch.stateManager,
|
||||
sch.evaluatorFactory,
|
||||
sch.clock,
|
||||
sch.rrCfg,
|
||||
sch.metrics,
|
||||
sch.log,
|
||||
sch.tracer,
|
||||
sch.featureToggles,
|
||||
sch.recordingWriter,
|
||||
sch.evalAppliedFunc,
|
||||
sch.stopAppliedFunc,
|
||||
)
|
||||
}
|
||||
|
||||
func stateForRule(rule *models.AlertRule, ts time.Time, evalState eval.State) *state.State {
|
||||
|
||||
@@ -40,7 +40,7 @@ type recordingRule struct {
|
||||
evaluationTimestamp *atomic.Time
|
||||
evaluationDuration *atomic.Duration
|
||||
|
||||
maxAttempts int64
|
||||
retryConfig RetryConfig
|
||||
|
||||
clock clock.Clock
|
||||
evalFactory eval.EvaluatorFactory
|
||||
@@ -56,7 +56,20 @@ type recordingRule struct {
|
||||
tracer tracing.Tracer
|
||||
}
|
||||
|
||||
func newRecordingRule(parent context.Context, key ngmodels.AlertRuleKeyWithGroup, maxAttempts int64, clock clock.Clock, evalFactory eval.EvaluatorFactory, cfg setting.RecordingRuleSettings, logger log.Logger, metrics *metrics.Scheduler, tracer tracing.Tracer, writer RecordingWriter, evalAppliedHook evalAppliedFunc, stopAppliedHook stopAppliedFunc) *recordingRule {
|
||||
func newRecordingRule(
|
||||
parent context.Context,
|
||||
key ngmodels.AlertRuleKeyWithGroup,
|
||||
retryConfig RetryConfig,
|
||||
clock clock.Clock,
|
||||
evalFactory eval.EvaluatorFactory,
|
||||
cfg setting.RecordingRuleSettings,
|
||||
logger log.Logger,
|
||||
metrics *metrics.Scheduler,
|
||||
tracer tracing.Tracer,
|
||||
writer RecordingWriter,
|
||||
evalAppliedHook evalAppliedFunc,
|
||||
stopAppliedHook stopAppliedFunc,
|
||||
) *recordingRule {
|
||||
ctx, stop := util.WithCancelCause(ngmodels.WithRuleKey(parent, key.AlertRuleKey))
|
||||
return &recordingRule{
|
||||
key: key,
|
||||
@@ -70,7 +83,7 @@ func newRecordingRule(parent context.Context, key ngmodels.AlertRuleKeyWithGroup
|
||||
clock: clock,
|
||||
evalFactory: evalFactory,
|
||||
cfg: cfg,
|
||||
maxAttempts: maxAttempts,
|
||||
retryConfig: retryConfig,
|
||||
evalAppliedHook: evalAppliedHook,
|
||||
stopAppliedHook: stopAppliedHook,
|
||||
logger: logger.FromContext(ctx),
|
||||
@@ -190,8 +203,17 @@ func (r *recordingRule) doEvaluate(ctx context.Context, ev *Evaluation) {
|
||||
))
|
||||
defer span.End()
|
||||
|
||||
retryer := newExponentialBackoffRetryer(
|
||||
r.retryConfig.MaxAttempts-1, // first attempt is not a retry
|
||||
r.retryConfig.InitialRetryDelay,
|
||||
r.retryConfig.MaxRetryDelay,
|
||||
r.retryConfig.RandomizationFactor,
|
||||
r.clock,
|
||||
)
|
||||
attempt := 1
|
||||
|
||||
var latestError error
|
||||
for attempt := int64(1); attempt <= r.maxAttempts; attempt++ {
|
||||
for {
|
||||
logger := logger.New("attempt", attempt)
|
||||
if ctx.Err() != nil {
|
||||
span.SetStatus(codes.Error, "rule evaluation cancelled")
|
||||
@@ -213,14 +235,20 @@ func (r *recordingRule) doEvaluate(ctx context.Context, ev *Evaluation) {
|
||||
break
|
||||
}
|
||||
|
||||
if attempt < r.maxAttempts {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
logger.Error("Context has been cancelled while backing off", "attempt", attempt)
|
||||
return
|
||||
case <-time.After(retryDelay):
|
||||
continue
|
||||
}
|
||||
retryIn := retryer.NextAttemptIn()
|
||||
if retryIn == retryStop {
|
||||
logger.Error("Recording rule evaluation failed after all attempts", "lastError", latestError)
|
||||
break
|
||||
}
|
||||
|
||||
attempt++
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
logger.Error("Context has been cancelled while backing off", "attempt", attempt)
|
||||
return
|
||||
case <-r.clock.After(retryIn):
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
@@ -230,9 +258,6 @@ func (r *recordingRule) doEvaluate(ctx context.Context, ev *Evaluation) {
|
||||
span.RecordError(latestError)
|
||||
r.lastError.Store(latestError)
|
||||
r.health.Store("error")
|
||||
if r.maxAttempts > 0 {
|
||||
logger.Error("Recording rule evaluation failed after all attempts", "lastError", latestError)
|
||||
}
|
||||
return
|
||||
}
|
||||
logger.Debug("Recording rule evaluation succeeded")
|
||||
|
||||
@@ -178,7 +178,8 @@ func blankRecordingRuleForTests(ctx context.Context) *recordingRule {
|
||||
st := setting.RecordingRuleSettings{
|
||||
Enabled: true,
|
||||
}
|
||||
return newRecordingRule(context.Background(), models.AlertRuleKeyWithGroup{}, 0, nil, nil, st, log.NewNopLogger(), nil, nil, writer.FakeWriter{}, nil, nil)
|
||||
|
||||
return newRecordingRule(context.Background(), models.AlertRuleKeyWithGroup{}, RetryConfig{}, nil, nil, st, log.NewNopLogger(), nil, nil, writer.FakeWriter{}, nil, nil)
|
||||
}
|
||||
|
||||
func TestRecordingRule_Integration(t *testing.T) {
|
||||
@@ -431,7 +432,8 @@ func testRecordingRule_Integration(t *testing.T, writeTarget *writer.TestRemoteW
|
||||
gen := models.RuleGen.With(models.RuleGen.WithAllRecordingRules(), models.RuleGen.WithOrgID(123))
|
||||
ruleStore := newFakeRulesStore()
|
||||
reg := prometheus.NewPedanticRegistry()
|
||||
sch := setupScheduler(t, ruleStore, nil, reg, nil, nil, nil)
|
||||
clk := clock.NewMock()
|
||||
sch := setupScheduler(t, ruleStore, nil, reg, nil, nil, nil, withSchedulerClock(clk))
|
||||
sch.recordingWriter = writer
|
||||
|
||||
t.Run("rule that succeeds", func(t *testing.T) {
|
||||
@@ -606,6 +608,13 @@ func testRecordingRule_Integration(t *testing.T, writeTarget *writer.TestRemoteW
|
||||
rule: rule,
|
||||
folderTitle: folderTitle,
|
||||
})
|
||||
|
||||
// Because we are using a mock clock, first we need to wait until the rule evaluation
|
||||
// reaches the point where it sleeps for the duration of the retry interval.
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
// Then advance the mock clock to trigger the retry.
|
||||
clk.Add(2 * time.Second)
|
||||
|
||||
_ = waitForTimeChannel(t, evalDoneChan)
|
||||
|
||||
t.Run("reports basic evaluation metrics", func(t *testing.T) {
|
||||
@@ -744,6 +753,13 @@ func testRecordingRule_Integration(t *testing.T, writeTarget *writer.TestRemoteW
|
||||
rule: rule,
|
||||
folderTitle: folderTitle,
|
||||
})
|
||||
|
||||
// Because we are using a mock clock, first we need to wait until the rule evaluation
|
||||
// reaches the point where it sleeps for the duration of the retry interval.
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
// Then advance the mock clock to trigger the retry.
|
||||
clk.Add(2 * time.Second)
|
||||
|
||||
_ = waitForTimeChannel(t, evalDoneChan)
|
||||
|
||||
t.Run("status shows evaluation", func(t *testing.T) {
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
package schedule
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/benbjohnson/clock"
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
)
|
||||
|
||||
const retryStop = backoff.Stop
|
||||
|
||||
type exponentialBackoffRetryer struct {
|
||||
backoff.BackOff
|
||||
}
|
||||
|
||||
func newExponentialBackoffRetryer(
|
||||
maxRetries int64,
|
||||
initialRetryDelay time.Duration,
|
||||
maxRetryDelay time.Duration,
|
||||
randomizationFactor float64,
|
||||
clock clock.Clock,
|
||||
) *exponentialBackoffRetryer {
|
||||
b := backoff.NewExponentialBackOff(
|
||||
backoff.WithClockProvider(clock),
|
||||
backoff.WithMaxInterval(maxRetryDelay),
|
||||
backoff.WithInitialInterval(initialRetryDelay),
|
||||
backoff.WithRandomizationFactor(randomizationFactor),
|
||||
)
|
||||
|
||||
return &exponentialBackoffRetryer{
|
||||
BackOff: backoff.WithMaxRetries(b, uint64(maxRetries)),
|
||||
}
|
||||
}
|
||||
|
||||
func (b *exponentialBackoffRetryer) NextAttemptIn() time.Duration {
|
||||
return b.NextBackOff()
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
package schedule
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/benbjohnson/clock"
|
||||
"github.com/cenkalti/backoff/v4"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestExponentialBackoffRetryProvider_New(t *testing.T) {
|
||||
testClock := clock.NewMock()
|
||||
|
||||
maxRetries := int64(5)
|
||||
initialDelay := 100 * time.Millisecond
|
||||
maxDelay := 1 * time.Second
|
||||
|
||||
retry := newExponentialBackoffRetryer(maxRetries, initialDelay, maxDelay, 0, testClock)
|
||||
require.NotNil(t, retry, "Retry instance should not be nil")
|
||||
|
||||
for i := int64(0); i < maxRetries; i++ {
|
||||
delay := retry.NextAttemptIn()
|
||||
require.GreaterOrEqual(t, delay, initialDelay, "Delay should be at least the initial delay")
|
||||
require.LessOrEqual(t, delay, maxDelay, "Delay should not exceed the max delay")
|
||||
|
||||
testClock.Add(delay)
|
||||
}
|
||||
|
||||
delay := retry.NextAttemptIn()
|
||||
require.Equal(t, backoff.Stop, delay, "Delay should be backoff.Stop after max retries")
|
||||
}
|
||||
|
||||
func TestExponentialBackoffRetryProvider_MaxRetries(t *testing.T) {
|
||||
testClock := clock.NewMock()
|
||||
|
||||
t.Run("max retries is zero", func(t *testing.T) {
|
||||
retry := newExponentialBackoffRetryer(0, 100*time.Millisecond, 1*time.Second, 0, testClock)
|
||||
require.NotNil(t, retry, "Retry instance should not be nil")
|
||||
delay := retry.NextAttemptIn()
|
||||
require.Equal(t, backoff.Stop, delay, "Should immediately stop when maxRetries is 0")
|
||||
})
|
||||
|
||||
t.Run("max retries is not zero", func(t *testing.T) {
|
||||
maxRetries := int64(10)
|
||||
retry := newExponentialBackoffRetryer(maxRetries, 10*time.Millisecond, 1*time.Second, 0, testClock)
|
||||
for i := int64(0); i < maxRetries; i++ {
|
||||
delay := retry.NextAttemptIn()
|
||||
require.NotEqual(t, backoff.Stop, delay, "Should not stop before reaching max retries")
|
||||
testClock.Add(delay)
|
||||
}
|
||||
|
||||
delay := retry.NextAttemptIn()
|
||||
require.Equal(t, backoff.Stop, delay, "Should stop after reaching maxRetries")
|
||||
})
|
||||
}
|
||||
|
||||
func TestExponentialBackoffRetryProvider_DelaysWithinBounds(t *testing.T) {
|
||||
testClock := clock.NewMock()
|
||||
|
||||
initialDelay := 200 * time.Millisecond
|
||||
maxDelay := 2 * time.Second
|
||||
maxRetries := int64(10)
|
||||
|
||||
retry := newExponentialBackoffRetryer(maxRetries, initialDelay, maxDelay, 0, testClock)
|
||||
|
||||
for i := int64(0); i < maxRetries; i++ {
|
||||
delay := retry.NextAttemptIn()
|
||||
require.GreaterOrEqual(t, delay, initialDelay, "Delay should not be less than initial delay")
|
||||
require.LessOrEqual(t, delay, maxDelay, "Delay should not exceed max delay")
|
||||
testClock.Add(delay)
|
||||
}
|
||||
}
|
||||
@@ -31,9 +31,6 @@ type ScheduleService interface {
|
||||
Run(context.Context) error
|
||||
}
|
||||
|
||||
// retryDelay represents how long to wait between each failed rule evaluation.
|
||||
const retryDelay = 1 * time.Second
|
||||
|
||||
// AlertsSender is an interface for a service that is responsible for sending notifications to the end-user.
|
||||
//
|
||||
//go:generate mockery --name AlertsSender --structname AlertsSenderMock --inpackage --filename alerts_sender_mock.go --with-expecter
|
||||
@@ -67,7 +64,7 @@ type schedule struct {
|
||||
// each rule gets its own channel and routine
|
||||
registry ruleRegistry
|
||||
|
||||
maxAttempts int64
|
||||
retryConfig RetryConfig
|
||||
|
||||
clock clock.Clock
|
||||
|
||||
@@ -112,9 +109,17 @@ type schedule struct {
|
||||
recordingWriter RecordingWriter
|
||||
}
|
||||
|
||||
// RetryConfig configures the exponential backoff for alert rule and recording rule evaluations.
|
||||
type RetryConfig struct {
|
||||
MaxAttempts int64
|
||||
InitialRetryDelay time.Duration
|
||||
MaxRetryDelay time.Duration
|
||||
RandomizationFactor float64
|
||||
}
|
||||
|
||||
// SchedulerCfg is the scheduler configuration.
|
||||
type SchedulerCfg struct {
|
||||
MaxAttempts int64
|
||||
RetryConfig RetryConfig
|
||||
BaseInterval time.Duration
|
||||
C clock.Clock
|
||||
MinRuleInterval time.Duration
|
||||
@@ -136,14 +141,14 @@ type SchedulerCfg struct {
|
||||
// NewScheduler returns a new scheduler.
|
||||
func NewScheduler(cfg SchedulerCfg, stateManager *state.Manager) *schedule {
|
||||
const minMaxAttempts = int64(1)
|
||||
if cfg.MaxAttempts < minMaxAttempts {
|
||||
cfg.Log.Warn("Invalid scheduler maxAttempts, using a safe minimum", "configured", cfg.MaxAttempts, "actual", minMaxAttempts)
|
||||
cfg.MaxAttempts = minMaxAttempts
|
||||
if cfg.RetryConfig.MaxAttempts < minMaxAttempts {
|
||||
cfg.Log.Warn("Invalid scheduler maxAttempts, using a safe minimum", "configured", cfg.RetryConfig.MaxAttempts, "actual", minMaxAttempts)
|
||||
cfg.RetryConfig.MaxAttempts = minMaxAttempts
|
||||
}
|
||||
|
||||
sch := schedule{
|
||||
registry: newRuleRegistry(),
|
||||
maxAttempts: cfg.MaxAttempts,
|
||||
retryConfig: cfg.RetryConfig,
|
||||
clock: cfg.C,
|
||||
baseInterval: cfg.BaseInterval,
|
||||
log: cfg.Log,
|
||||
@@ -168,7 +173,7 @@ func NewScheduler(cfg SchedulerCfg, stateManager *state.Manager) *schedule {
|
||||
}
|
||||
|
||||
func (sch *schedule) Run(ctx context.Context) error {
|
||||
sch.log.Info("Starting scheduler", "tickInterval", sch.baseInterval, "maxAttempts", sch.maxAttempts)
|
||||
sch.log.Info("Starting scheduler", "tickInterval", sch.baseInterval, "maxAttempts", sch.retryConfig.MaxAttempts)
|
||||
t := ticker.New(sch.clock, sch.baseInterval, sch.metrics.Ticker, sch.log)
|
||||
defer t.Stop()
|
||||
|
||||
@@ -296,10 +301,11 @@ func (sch *schedule) processTick(ctx context.Context, dispatcherGroup *errgroup.
|
||||
updatedRules := make([]ngmodels.AlertRuleKeyWithVersion, 0, len(updated)) // this is needed for tests only
|
||||
restartedRules := make([]Rule, 0)
|
||||
missingFolder := make(map[string][]string)
|
||||
|
||||
ruleFactory := newRuleFactory(
|
||||
sch.appURL,
|
||||
sch.disableGrafanaFolder,
|
||||
sch.maxAttempts,
|
||||
sch.retryConfig,
|
||||
sch.alertsSender,
|
||||
sch.stateManager,
|
||||
sch.evaluatorFactory,
|
||||
|
||||
@@ -85,6 +85,12 @@ func TestProcessTicks(t *testing.T) {
|
||||
}
|
||||
|
||||
schedCfg := SchedulerCfg{
|
||||
RetryConfig: RetryConfig{
|
||||
MaxAttempts: 1,
|
||||
InitialRetryDelay: time.Second,
|
||||
MaxRetryDelay: time.Second * 10,
|
||||
RandomizationFactor: 0,
|
||||
},
|
||||
BaseInterval: cfg.BaseInterval,
|
||||
C: mockedClock,
|
||||
AppURL: appUrl,
|
||||
@@ -1189,11 +1195,35 @@ func TestSchedule_deleteAlertRule(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func setupScheduler(t *testing.T, rs *fakeRulesStore, is *state.FakeInstanceStore, registry *prometheus.Registry, senderMock *SyncAlertsSenderMock, evalMock eval.EvaluatorFactory, ruleStopReasonProvider AlertRuleStopReasonProvider) *schedule {
|
||||
type schedulerOpts struct {
|
||||
clock clock.Clock
|
||||
}
|
||||
|
||||
func withSchedulerClock(clock clock.Clock) func(opts *schedulerOpts) {
|
||||
return func(opts *schedulerOpts) {
|
||||
opts.clock = clock
|
||||
}
|
||||
}
|
||||
|
||||
func setupScheduler(
|
||||
t *testing.T,
|
||||
rs *fakeRulesStore,
|
||||
is *state.FakeInstanceStore,
|
||||
registry *prometheus.Registry,
|
||||
senderMock *SyncAlertsSenderMock,
|
||||
evalMock eval.EvaluatorFactory,
|
||||
ruleStopReasonProvider AlertRuleStopReasonProvider,
|
||||
options ...func(opts *schedulerOpts),
|
||||
) *schedule {
|
||||
t.Helper()
|
||||
testTracer := tracing.InitializeTracerForTest()
|
||||
|
||||
mockedClock := clock.NewMock()
|
||||
opts := &schedulerOpts{
|
||||
clock: clock.NewMock(),
|
||||
}
|
||||
for _, o := range options {
|
||||
o(opts)
|
||||
}
|
||||
|
||||
if rs == nil {
|
||||
rs = newFakeRulesStore()
|
||||
@@ -1236,8 +1266,11 @@ func setupScheduler(t *testing.T, rs *fakeRulesStore, is *state.FakeInstanceStor
|
||||
}
|
||||
|
||||
cfg := setting.UnifiedAlertingSettings{
|
||||
BaseInterval: time.Second,
|
||||
MaxAttempts: 1,
|
||||
BaseInterval: time.Second,
|
||||
MaxAttempts: 1,
|
||||
InitialRetryDelay: time.Second * 1,
|
||||
MaxRetryDelay: time.Second * 10,
|
||||
RandomizationFactor: 0,
|
||||
RecordingRules: setting.RecordingRuleSettings{
|
||||
Enabled: true,
|
||||
},
|
||||
@@ -1246,9 +1279,14 @@ func setupScheduler(t *testing.T, rs *fakeRulesStore, is *state.FakeInstanceStor
|
||||
fakeRecordingWriter := writer.FakeWriter{}
|
||||
|
||||
schedCfg := SchedulerCfg{
|
||||
RetryConfig: RetryConfig{
|
||||
MaxAttempts: cfg.MaxAttempts,
|
||||
InitialRetryDelay: cfg.InitialRetryDelay,
|
||||
MaxRetryDelay: cfg.MaxRetryDelay,
|
||||
RandomizationFactor: cfg.RandomizationFactor,
|
||||
},
|
||||
BaseInterval: cfg.BaseInterval,
|
||||
MaxAttempts: cfg.MaxAttempts,
|
||||
C: mockedClock,
|
||||
C: opts.clock,
|
||||
AppURL: appUrl,
|
||||
EvaluatorFactory: evaluator,
|
||||
RuleStore: rs,
|
||||
@@ -1266,7 +1304,7 @@ func setupScheduler(t *testing.T, rs *fakeRulesStore, is *state.FakeInstanceStor
|
||||
ExternalURL: nil,
|
||||
InstanceStore: is,
|
||||
Images: &state.NoopImageService{},
|
||||
Clock: mockedClock,
|
||||
Clock: opts.clock,
|
||||
Historian: &state.FakeHistorian{},
|
||||
Tracer: testTracer,
|
||||
Log: log.New("ngalert.state.manager"),
|
||||
|
||||
Reference in New Issue
Block a user