Alerting: Add SaveAlertInstancesForRule instance store method (#94505)
Alerting: Add SaveAlertInstancesForRule method to the InstanceStore interface
This commit is contained in:
@@ -86,7 +86,7 @@ func newRuleFactory(
|
||||
}
|
||||
return newAlertRule(
|
||||
ctx,
|
||||
rule.GetKey(),
|
||||
rule.GetKeyWithGroup(),
|
||||
appURL,
|
||||
disableGrafanaFolder,
|
||||
maxAttempts,
|
||||
@@ -112,7 +112,7 @@ type ruleProvider interface {
|
||||
}
|
||||
|
||||
type alertRule struct {
|
||||
key ngmodels.AlertRuleKey
|
||||
key ngmodels.AlertRuleKeyWithGroup
|
||||
|
||||
evalCh chan *Evaluation
|
||||
updateCh chan RuleVersionAndPauseStatus
|
||||
@@ -140,7 +140,7 @@ type alertRule struct {
|
||||
|
||||
func newAlertRule(
|
||||
parent context.Context,
|
||||
key ngmodels.AlertRuleKey,
|
||||
key ngmodels.AlertRuleKeyWithGroup,
|
||||
appURL *url.URL,
|
||||
disableGrafanaFolder bool,
|
||||
maxAttempts int64,
|
||||
@@ -155,7 +155,7 @@ func newAlertRule(
|
||||
evalAppliedHook func(ngmodels.AlertRuleKey, time.Time),
|
||||
stopAppliedHook func(ngmodels.AlertRuleKey),
|
||||
) *alertRule {
|
||||
ctx, stop := util.WithCancelCause(ngmodels.WithRuleKey(parent, key))
|
||||
ctx, stop := util.WithCancelCause(ngmodels.WithRuleKey(parent, key.AlertRuleKey))
|
||||
return &alertRule{
|
||||
key: key,
|
||||
evalCh: make(chan *Evaluation),
|
||||
@@ -194,7 +194,7 @@ func (a *alertRule) Status() ngmodels.RuleStatus {
|
||||
//
|
||||
// the second element contains a dropped message that was sent by a concurrent sender.
|
||||
func (a *alertRule) Eval(eval *Evaluation) (bool, *Evaluation) {
|
||||
if a.key != eval.rule.GetKey() {
|
||||
if a.key != eval.rule.GetKeyWithGroup() {
|
||||
// Make sure that rule has the same key. This should not happen
|
||||
a.logger.Error("Invalid rule sent for evaluating. Skipping", "ruleKeyToEvaluate", eval.rule.GetKey().String())
|
||||
return false, eval
|
||||
@@ -352,7 +352,7 @@ func (a *alertRule) Run() error {
|
||||
// cases.
|
||||
ctx, cancelFunc := context.WithTimeout(context.Background(), time.Minute)
|
||||
defer cancelFunc()
|
||||
states := a.stateManager.DeleteStateByRuleUID(ngmodels.WithRuleKey(ctx, a.key), a.key, ngmodels.StateReasonRuleDeleted)
|
||||
states := a.stateManager.DeleteStateByRuleUID(ngmodels.WithRuleKey(ctx, a.key.AlertRuleKey), a.key, ngmodels.StateReasonRuleDeleted)
|
||||
a.expireAndSend(grafanaCtx, states)
|
||||
}
|
||||
a.logger.Debug("Stopping alert rule routine")
|
||||
@@ -467,7 +467,7 @@ func (a *alertRule) send(ctx context.Context, logger log.Logger, states state.St
|
||||
|
||||
if len(alerts.PostableAlerts) > 0 {
|
||||
logger.Debug("Sending transitions to notifier", "transitions", len(alerts.PostableAlerts))
|
||||
a.sender.Send(ctx, a.key, alerts)
|
||||
a.sender.Send(ctx, a.key.AlertRuleKey, alerts)
|
||||
}
|
||||
return alerts
|
||||
}
|
||||
@@ -476,12 +476,12 @@ func (a *alertRule) send(ctx context.Context, logger log.Logger, states state.St
|
||||
func (a *alertRule) expireAndSend(ctx context.Context, states []state.StateTransition) {
|
||||
expiredAlerts := state.FromAlertsStateToStoppedAlert(states, a.appURL, a.clock)
|
||||
if len(expiredAlerts.PostableAlerts) > 0 {
|
||||
a.sender.Send(ctx, a.key, expiredAlerts)
|
||||
a.sender.Send(ctx, a.key.AlertRuleKey, expiredAlerts)
|
||||
}
|
||||
}
|
||||
|
||||
func (a *alertRule) resetState(ctx context.Context, isPaused bool) {
|
||||
rule := a.ruleProvider.get(a.key)
|
||||
rule := a.ruleProvider.get(a.key.AlertRuleKey)
|
||||
reason := ngmodels.StateReasonUpdated
|
||||
if isPaused {
|
||||
reason = ngmodels.StateReasonPaused
|
||||
@@ -496,7 +496,7 @@ func (a *alertRule) evalApplied(now time.Time) {
|
||||
return
|
||||
}
|
||||
|
||||
a.evalAppliedHook(a.key, now)
|
||||
a.evalAppliedHook(a.key.AlertRuleKey, now)
|
||||
}
|
||||
|
||||
// stopApplied is only used on tests.
|
||||
@@ -505,7 +505,7 @@ func (a *alertRule) stopApplied() {
|
||||
return
|
||||
}
|
||||
|
||||
a.stopAppliedHook(a.key)
|
||||
a.stopAppliedHook(a.key.AlertRuleKey)
|
||||
}
|
||||
|
||||
func SchedulerUserFor(orgID int64) *user.SignedInUser {
|
||||
|
||||
@@ -39,7 +39,7 @@ func TestAlertRule(t *testing.T) {
|
||||
|
||||
t.Run("when rule evaluation is not stopped", func(t *testing.T) {
|
||||
t.Run("update should send to updateCh", func(t *testing.T) {
|
||||
r := blankRuleForTests(context.Background(), models.GenerateRuleKey(1))
|
||||
r := blankRuleForTests(context.Background(), models.GenerateRuleKeyWithGroup(1))
|
||||
resultCh := make(chan bool)
|
||||
go func() {
|
||||
resultCh <- r.Update(RuleVersionAndPauseStatus{fingerprint(rand.Uint64()), false})
|
||||
@@ -52,7 +52,7 @@ func TestAlertRule(t *testing.T) {
|
||||
}
|
||||
})
|
||||
t.Run("update should drop any concurrent sending to updateCh", func(t *testing.T) {
|
||||
r := blankRuleForTests(context.Background(), models.GenerateRuleKey(1))
|
||||
r := blankRuleForTests(context.Background(), models.GenerateRuleKeyWithGroup(1))
|
||||
version1 := RuleVersionAndPauseStatus{fingerprint(rand.Uint64()), false}
|
||||
version2 := RuleVersionAndPauseStatus{fingerprint(rand.Uint64()), false}
|
||||
|
||||
@@ -79,7 +79,7 @@ func TestAlertRule(t *testing.T) {
|
||||
})
|
||||
t.Run("eval should send to evalCh", func(t *testing.T) {
|
||||
ruleSpec := gen.GenerateRef()
|
||||
r := blankRuleForTests(context.Background(), ruleSpec.GetKey())
|
||||
r := blankRuleForTests(context.Background(), ruleSpec.GetKeyWithGroup())
|
||||
expected := time.Now()
|
||||
resultCh := make(chan evalResponse)
|
||||
data := &Evaluation{
|
||||
@@ -103,7 +103,7 @@ func TestAlertRule(t *testing.T) {
|
||||
})
|
||||
t.Run("eval should drop any concurrent sending to evalCh", func(t *testing.T) {
|
||||
ruleSpec := gen.GenerateRef()
|
||||
r := blankRuleForTests(context.Background(), ruleSpec.GetKey())
|
||||
r := blankRuleForTests(context.Background(), ruleSpec.GetKeyWithGroup())
|
||||
time1 := time.UnixMilli(rand.Int63n(math.MaxInt64))
|
||||
time2 := time.UnixMilli(rand.Int63n(math.MaxInt64))
|
||||
resultCh1 := make(chan evalResponse)
|
||||
@@ -150,7 +150,7 @@ func TestAlertRule(t *testing.T) {
|
||||
})
|
||||
t.Run("eval should exit when context is cancelled", func(t *testing.T) {
|
||||
ruleSpec := gen.GenerateRef()
|
||||
r := blankRuleForTests(context.Background(), ruleSpec.GetKey())
|
||||
r := blankRuleForTests(context.Background(), ruleSpec.GetKeyWithGroup())
|
||||
resultCh := make(chan evalResponse)
|
||||
data := &Evaluation{
|
||||
scheduledAt: time.Now(),
|
||||
@@ -174,14 +174,14 @@ func TestAlertRule(t *testing.T) {
|
||||
})
|
||||
t.Run("when rule evaluation is stopped", func(t *testing.T) {
|
||||
t.Run("Update should do nothing", func(t *testing.T) {
|
||||
r := blankRuleForTests(context.Background(), models.GenerateRuleKey(1))
|
||||
r := blankRuleForTests(context.Background(), models.GenerateRuleKeyWithGroup(1))
|
||||
r.Stop(errRuleDeleted)
|
||||
require.ErrorIs(t, r.ctx.Err(), errRuleDeleted)
|
||||
require.False(t, r.Update(RuleVersionAndPauseStatus{fingerprint(rand.Uint64()), false}))
|
||||
})
|
||||
t.Run("eval should do nothing", func(t *testing.T) {
|
||||
ruleSpec := gen.GenerateRef()
|
||||
r := blankRuleForTests(context.Background(), ruleSpec.GetKey())
|
||||
r := blankRuleForTests(context.Background(), ruleSpec.GetKeyWithGroup())
|
||||
r.Stop(nil)
|
||||
data := &Evaluation{
|
||||
scheduledAt: time.Now(),
|
||||
@@ -193,19 +193,19 @@ func TestAlertRule(t *testing.T) {
|
||||
require.Nilf(t, dropped, "expected no dropped evaluations but got one")
|
||||
})
|
||||
t.Run("calling stop multiple times should not panic", func(t *testing.T) {
|
||||
r := blankRuleForTests(context.Background(), models.GenerateRuleKey(1))
|
||||
r := blankRuleForTests(context.Background(), models.GenerateRuleKeyWithGroup(1))
|
||||
r.Stop(nil)
|
||||
r.Stop(nil)
|
||||
})
|
||||
t.Run("stop should not panic if parent context stopped", func(t *testing.T) {
|
||||
ctx, cancelFn := context.WithCancel(context.Background())
|
||||
r := blankRuleForTests(ctx, models.GenerateRuleKey(1))
|
||||
r := blankRuleForTests(ctx, models.GenerateRuleKeyWithGroup(1))
|
||||
cancelFn()
|
||||
r.Stop(nil)
|
||||
})
|
||||
})
|
||||
t.Run("should be thread-safe", func(t *testing.T) {
|
||||
r := blankRuleForTests(context.Background(), models.GenerateRuleKey(1))
|
||||
r := blankRuleForTests(context.Background(), models.GenerateRuleKeyWithGroup(1))
|
||||
wg := sync.WaitGroup{}
|
||||
go func() {
|
||||
for {
|
||||
@@ -249,7 +249,7 @@ func TestAlertRule(t *testing.T) {
|
||||
})
|
||||
|
||||
t.Run("Run should exit if idle when Stop is called", func(t *testing.T) {
|
||||
rule := blankRuleForTests(context.Background(), models.GenerateRuleKey(1))
|
||||
rule := blankRuleForTests(context.Background(), models.GenerateRuleKeyWithGroup(1))
|
||||
runResult := make(chan error)
|
||||
go func() {
|
||||
runResult <- rule.Run()
|
||||
@@ -266,7 +266,7 @@ func TestAlertRule(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func blankRuleForTests(ctx context.Context, key models.AlertRuleKey) *alertRule {
|
||||
func blankRuleForTests(ctx context.Context, key models.AlertRuleKeyWithGroup) *alertRule {
|
||||
return newAlertRule(ctx, key, nil, false, 0, nil, nil, nil, nil, nil, nil, log.NewNopLogger(), nil, nil, nil)
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user