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

Alerting: Restart rule routines if the type changes (#90867)

Browse Source 
* Restart when types change

* Wire up test hooks correctly

* testing
This commit is contained in:
Alexander Weaver
2024-08-14 14:57:47 -05:00
committed by GitHub
parent f158d52ae4
commit 34ab5fe1f3
8 changed files with 160 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/ngalert/models/alert_rule.go
+2 -2
View File
@@ -108,8 +108,8 @@ const (
type RuleType string
const (
RuleTypeAlerting = "alerting"
RuleTypeRecording = "recording"
RuleTypeAlerting RuleType = "alerting"
RuleTypeRecording RuleType = "recording"
)
func (r RuleType) String() string {
pkg/services/ngalert/models/testing.go
+3 -3
View File
@@ -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{}
}
pkg/services/ngalert/schedule/alert_rule.go
+8
View File
@@ -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
pkg/services/ngalert/schedule/recording_rule.go
+23 -6
View File
@@ -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)
}
pkg/services/ngalert/schedule/recording_rule_test.go
+1 -1
View File
@@ -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) {
pkg/services/ngalert/schedule/registry.go
+4 -1
View File
@@ -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
pkg/services/ngalert/schedule/schedule.go
+14
View File
@@ -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 {
pkg/services/ngalert/schedule/schedule_unit_test.go
+105 -1
View File
@@ -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)
}