diff --git a/pkg/services/ngalert/models/alert_rule.go b/pkg/services/ngalert/models/alert_rule.go index 127e85c1842..f920a3902b5 100644 --- a/pkg/services/ngalert/models/alert_rule.go +++ b/pkg/services/ngalert/models/alert_rule.go @@ -108,8 +108,8 @@ const ( type RuleType string const ( - RuleTypeAlerting = "alerting" - RuleTypeRecording = "recording" + RuleTypeAlerting RuleType = "alerting" + RuleTypeRecording RuleType = "recording" ) func (r RuleType) String() string { diff --git a/pkg/services/ngalert/models/testing.go b/pkg/services/ngalert/models/testing.go index 657d0546644..f3281f83af8 100644 --- a/pkg/services/ngalert/models/testing.go +++ b/pkg/services/ngalert/models/testing.go @@ -495,13 +495,13 @@ func (a *AlertRuleMutators) WithRandomRecordingRules() AlertRuleMutator { if rand.Int63()%2 == 0 { return } - convertToRecordingRule(rule) + ConvertToRecordingRule(rule) } } func (a *AlertRuleMutators) WithAllRecordingRules() AlertRuleMutator { return func(rule *AlertRule) { - convertToRecordingRule(rule) + ConvertToRecordingRule(rule) } } @@ -1092,7 +1092,7 @@ func (n SilenceMutators) WithEmptyId() Mutator[Silence] { } } -func convertToRecordingRule(rule *AlertRule) { +func ConvertToRecordingRule(rule *AlertRule) { if rule.Record == nil { rule.Record = &Record{} } diff --git a/pkg/services/ngalert/schedule/alert_rule.go b/pkg/services/ngalert/schedule/alert_rule.go index c96a1515db8..b19cd4976b7 100644 --- a/pkg/services/ngalert/schedule/alert_rule.go +++ b/pkg/services/ngalert/schedule/alert_rule.go @@ -38,6 +38,8 @@ type Rule interface { Eval(eval *Evaluation) (bool, *Evaluation) // Update sends a singal to change the definition of the rule. Update(lastVersion RuleVersionAndPauseStatus) bool + // Type gives the type of the rule. + Type() ngmodels.RuleType } type ruleFactoryFunc func(context.Context, *ngmodels.AlertRule) Rule @@ -76,6 +78,8 @@ func newRuleFactory( met, tracer, recordingWriter, + evalAppliedHook, + stopAppliedHook, ) } return newAlertRule( @@ -172,6 +176,10 @@ func newAlertRule( } } +func (a *alertRule) Type() ngmodels.RuleType { + return ngmodels.RuleTypeAlerting +} + // eval signals the rule evaluation routine to perform the evaluation of the rule. Does nothing if the loop is stopped. // Before sending a message into the channel, it does non-blocking read to make sure that there is no concurrent send operation. // Returns a tuple where first element is diff --git a/pkg/services/ngalert/schedule/recording_rule.go b/pkg/services/ngalert/schedule/recording_rule.go index cb9878f40ea..74c248f2ff1 100644 --- a/pkg/services/ngalert/schedule/recording_rule.go +++ b/pkg/services/ngalert/schedule/recording_rule.go @@ -13,7 +13,6 @@ import ( "go.opentelemetry.io/otel/trace" "go.uber.org/atomic" - "github.com/grafana/grafana/pkg/cmd/grafana-cli/logger" "github.com/grafana/grafana/pkg/infra/log" "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/services/featuremgmt" @@ -50,13 +49,14 @@ type recordingRule struct { // Event hooks that are only used in tests. evalAppliedHook evalAppliedFunc + stopAppliedHook stopAppliedFunc logger log.Logger metrics *metrics.Scheduler tracer tracing.Tracer } -func newRecordingRule(parent context.Context, key ngmodels.AlertRuleKey, maxAttempts int64, clock clock.Clock, evalFactory eval.EvaluatorFactory, ft featuremgmt.FeatureToggles, logger log.Logger, metrics *metrics.Scheduler, tracer tracing.Tracer, writer RecordingWriter) *recordingRule { +func newRecordingRule(parent context.Context, key ngmodels.AlertRuleKey, maxAttempts int64, clock clock.Clock, evalFactory eval.EvaluatorFactory, ft featuremgmt.FeatureToggles, logger log.Logger, metrics *metrics.Scheduler, tracer tracing.Tracer, writer RecordingWriter, evalAppliedHook evalAppliedFunc, stopAppliedHook stopAppliedFunc) *recordingRule { ctx, stop := util.WithCancelCause(ngmodels.WithRuleKey(parent, key)) return &recordingRule{ key: key, @@ -71,6 +71,8 @@ func newRecordingRule(parent context.Context, key ngmodels.AlertRuleKey, maxAtte evalFactory: evalFactory, featureToggles: ft, maxAttempts: maxAttempts, + evalAppliedHook: evalAppliedHook, + stopAppliedHook: stopAppliedHook, logger: logger.FromContext(ctx), metrics: metrics, tracer: tracer, @@ -78,6 +80,10 @@ func newRecordingRule(parent context.Context, key ngmodels.AlertRuleKey, maxAtte } } +func (r *recordingRule) Type() ngmodels.RuleType { + return ngmodels.RuleTypeRecording +} + func (r *recordingRule) Status() RuleStatus { return RuleStatus{ Health: r.health.Load(), @@ -115,17 +121,19 @@ func (r *recordingRule) Stop(reason error) { func (r *recordingRule) Run() error { ctx := r.ctx - logger.Debug("Recording rule routine started") + r.logger.Debug("Recording rule routine started") + + defer r.stopApplied() for { select { case eval, ok := <-r.evalCh: if !ok { - logger.Debug("Evaluation channel has been closed. Exiting") + r.logger.Debug("Evaluation channel has been closed. Exiting") return nil } if !r.featureToggles.IsEnabled(ctx, featuremgmt.FlagGrafanaManagedRecordingRules) { - logger.Warn("Recording rule scheduled but toggle is not enabled. Skipping") + r.logger.Warn("Recording rule scheduled but toggle is not enabled. Skipping") return nil } // TODO: Skipping the "evalRunning" guard that the alert rule routine does, because it seems to be dead code and impossible to hit. @@ -133,7 +141,7 @@ func (r *recordingRule) Run() error { r.doEvaluate(ctx, eval) case <-ctx.Done(): - logger.Debug("Stopping recording rule routine") + r.logger.Debug("Stopping recording rule routine") return nil } } @@ -313,3 +321,12 @@ func (r *recordingRule) frameRef(refID string, resp *backend.QueryDataResponse) return targetNode.Frames, nil } + +// stopApplied is only used on tests. +func (r *recordingRule) stopApplied() { + if r.stopAppliedHook == nil { + return + } + + r.stopAppliedHook(r.key) +} diff --git a/pkg/services/ngalert/schedule/recording_rule_test.go b/pkg/services/ngalert/schedule/recording_rule_test.go index 8dba179927d..fde2f1d65fc 100644 --- a/pkg/services/ngalert/schedule/recording_rule_test.go +++ b/pkg/services/ngalert/schedule/recording_rule_test.go @@ -154,7 +154,7 @@ func TestRecordingRule(t *testing.T) { func blankRecordingRuleForTests(ctx context.Context) *recordingRule { ft := featuremgmt.WithFeatures(featuremgmt.FlagGrafanaManagedRecordingRules) - return newRecordingRule(context.Background(), models.AlertRuleKey{}, 0, nil, nil, ft, log.NewNopLogger(), nil, nil, writer.FakeWriter{}) + return newRecordingRule(context.Background(), models.AlertRuleKey{}, 0, nil, nil, ft, log.NewNopLogger(), nil, nil, writer.FakeWriter{}, nil, nil) } func TestRecordingRule_Integration(t *testing.T) { diff --git a/pkg/services/ngalert/schedule/registry.go b/pkg/services/ngalert/schedule/registry.go index 576b51045d6..157124373f5 100644 --- a/pkg/services/ngalert/schedule/registry.go +++ b/pkg/services/ngalert/schedule/registry.go @@ -15,7 +15,10 @@ import ( "github.com/grafana/grafana/pkg/services/ngalert/models" ) -var errRuleDeleted = errors.New("rule deleted") +var ( + errRuleDeleted = errors.New("rule deleted") + errRuleRestarted = errors.New("rule restarted") +) type ruleFactory interface { new(context.Context, *models.AlertRule) Rule diff --git a/pkg/services/ngalert/schedule/schedule.go b/pkg/services/ngalert/schedule/schedule.go index 1772cbf8bb5..414d44c1f74 100644 --- a/pkg/services/ngalert/schedule/schedule.go +++ b/pkg/services/ngalert/schedule/schedule.go @@ -255,6 +255,7 @@ func (sch *schedule) processTick(ctx context.Context, dispatcherGroup *errgroup. readyToRun := make([]readyToRunItem, 0) 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, @@ -286,6 +287,14 @@ func (sch *schedule) processTick(ctx context.Context, dispatcherGroup *errgroup. invalidInterval := item.IntervalSeconds%int64(sch.baseInterval.Seconds()) != 0 + if item.Type() != ruleRoutine.Type() { + // Restart rules that need it. For now we just replace them, we'll shut them down at the end of the tick. + logger.Debug("Rule restarted because type changed", "old", ruleRoutine.Type(), "new", item.Type()) + restartedRules = append(restartedRules, ruleRoutine) + sch.registry.del(key) + ruleRoutine, newRoutine = sch.registry.getOrCreate(ctx, item, ruleFactory) + } + if newRoutine && !invalidInterval { dispatcherGroup.Go(func() error { return ruleRoutine.Run() @@ -370,6 +379,11 @@ func (sch *schedule) processTick(ctx context.Context, dispatcherGroup *errgroup. }) } + // Stop old routines for rules that got restarted. + for _, oldRoutine := range restartedRules { + oldRoutine.Stop(errRuleRestarted) + } + // unregister and stop routines of the deleted alert rules toDelete := make([]ngmodels.AlertRuleKey, 0, len(registeredDefinitions)) for key := range registeredDefinitions { diff --git a/pkg/services/ngalert/schedule/schedule_unit_test.go b/pkg/services/ngalert/schedule/schedule_unit_test.go index 7e94bee11a1..d8a6ecd432e 100644 --- a/pkg/services/ngalert/schedule/schedule_unit_test.go +++ b/pkg/services/ngalert/schedule/schedule_unit_test.go @@ -74,6 +74,7 @@ func TestProcessTicks(t *testing.T) { RuleStore: ruleStore, Metrics: testMetrics.GetSchedulerMetrics(), AlertSender: notifier, + FeatureToggles: featuremgmt.WithFeatures(featuremgmt.FlagGrafanaManagedRecordingRules), Tracer: testTracer, Log: log.New("ngalert.scheduler"), } @@ -367,10 +368,101 @@ func TestProcessTicks(t *testing.T) { require.Len(t, updated, 1) require.Equal(t, expectedUpdated, updated[0]) }) - t.Run("on 12th tick all rules should be stopped", func(t *testing.T) { + + // Add a recording rule with 2 * base interval. + recordingRule1 := gen.With(gen.WithOrgID(mainOrgID), gen.WithInterval(2*cfg.BaseInterval), gen.WithTitle("recording-1"), gen.WithAllRecordingRules()).GenerateRef() + ruleStore.PutRule(ctx, recordingRule1) + + t.Run("on 12th tick recording rule and alert rules should be evaluated", func(t *testing.T) { + tick = tick.Add(cfg.BaseInterval) + + scheduled, stopped, updated := sched.processTick(ctx, dispatcherGroup, tick) + + require.Len(t, scheduled, 3) + require.Emptyf(t, stopped, "No rules are expected to be stopped") + require.Emptyf(t, updated, "No rules are expected to be updated") + contains := false + for _, sch := range scheduled { + if sch.rule.Title == recordingRule1.Title { + contains = true + } + } + require.True(t, contains, "Expected a scheduled rule with title %s but didn't get one, scheduled rules were %v", recordingRule1.Title, scheduled) + }) + + // Update the recording rule. + recordingRule1 = models.CopyRule(recordingRule1) + recordingRule1.Version++ + expectedUpdated := models.AlertRuleKeyWithVersion{ + Version: recordingRule1.Version, + AlertRuleKey: recordingRule1.GetKey(), + } + ruleStore.PutRule(context.Background(), recordingRule1) + + t.Run("on 13th tick recording rule should be updated", func(t *testing.T) { + // It has 2 * base interval - so normally it would not have been scheduled for evaluation this tick. + tick = tick.Add(cfg.BaseInterval) + scheduled, stopped, updated := sched.processTick(ctx, dispatcherGroup, tick) + + require.Len(t, scheduled, 1) + require.Emptyf(t, stopped, "No rules are expected to be stopped") + require.Len(t, updated, 1) + require.Equal(t, expectedUpdated, updated[0]) + assertScheduledContains(t, scheduled, alertRule3) + }) + + t.Run("on 14th tick both 1-tick alert rule and 2-tick recording rule should be evaluated", func(t *testing.T) { + tick = tick.Add(cfg.BaseInterval) + + scheduled, stopped, updated := sched.processTick(ctx, dispatcherGroup, tick) + + require.Len(t, scheduled, 2) + require.Emptyf(t, stopped, "No rules are expected to be stopped") + require.Emptyf(t, updated, "No rules are expected to be updated") + assertScheduledContains(t, scheduled, alertRule3) + assertScheduledContains(t, scheduled, recordingRule1) + }) + + // Convert an alerting rule to a recording rule. + models.ConvertToRecordingRule(alertRule3) + alertRule3.Version++ + ruleStore.PutRule(ctx, alertRule3) + + t.Run("prior to 15th tick alertRule3 should still be scheduled as alerting rule", func(t *testing.T) { + require.Equal(t, models.RuleTypeAlerting, sched.registry.rules[alertRule3.GetKey()].Type()) + }) + + t.Run("on 15th tick converted rule and 3-tick alert rule should be evaluated", func(t *testing.T) { + tick = tick.Add(cfg.BaseInterval) + scheduled, stopped, updated := sched.processTick(ctx, dispatcherGroup, tick) + + require.Len(t, scheduled, 2) + require.Emptyf(t, stopped, "No rules are expected to be stopped") + // We never sent the Updated command to the restarted rule, so this should be empty. + require.Emptyf(t, updated, "No rules are expected to be updated") + + assertScheduledContains(t, scheduled, alertRule2) + assertScheduledContains(t, scheduled, alertRule3) // converted + // Rule in registry should be updated to the correct type. + require.Equal(t, models.RuleTypeRecording, sched.registry.rules[alertRule3.GetKey()].Type()) + }) + + t.Run("on 16th tick converted rule and 2-tick recording rule should be evaluated", func(t *testing.T) { + tick = tick.Add(cfg.BaseInterval) + scheduled, stopped, updated := sched.processTick(ctx, dispatcherGroup, tick) + + require.Len(t, scheduled, 2) + require.Emptyf(t, stopped, "No rules are expected to be stopped") + require.Emptyf(t, updated, "No rules are expected to be updated") + assertScheduledContains(t, scheduled, recordingRule1) + assertScheduledContains(t, scheduled, alertRule3) + }) + + t.Run("on 17th tick all rules should be stopped", func(t *testing.T) { expectedToBeStopped, err := ruleStore.GetAlertRulesKeysForScheduling(ctx) require.NoError(t, err) + // Remove all rules from store. ruleStore.rules = map[string]*models.AlertRule{} tick = tick.Add(cfg.BaseInterval) scheduled, stopped, updated := sched.processTick(ctx, dispatcherGroup, tick) @@ -899,3 +991,15 @@ func assertStopRun(t *testing.T, ch <-chan models.AlertRuleKey, keys ...models.A } } } + +func assertScheduledContains(t *testing.T, scheduled []readyToRunItem, rule *models.AlertRule) { + t.Helper() + + contains := false + for _, sch := range scheduled { + if sch.rule.GetKey() == rule.GetKey() { + contains = true + } + } + require.True(t, contains, "Expected a scheduled rule with key %s title %s but didn't get one, scheduled rules were %v", rule.GetKey(), rule.Title, scheduled) +}