diff --git a/pkg/services/ngalert/state/cache.go b/pkg/services/ngalert/state/cache.go index a7302fc5518..b2972d93e59 100644 --- a/pkg/services/ngalert/state/cache.go +++ b/pkg/services/ngalert/state/cache.go @@ -166,25 +166,23 @@ func expand(ctx context.Context, log log.Logger, name string, original map[strin return expanded, errs } -func (rs *ruleStates) deleteStates(predicate func(s *State) bool) []*State { - deleted := make([]*State, 0) +func (rs *ruleStates) deleteStates(predicate func(s *State) bool) { for id, state := range rs.states { if predicate(state) { delete(rs.states, id) - deleted = append(deleted, state) } } - return deleted } -func (c *cache) deleteRuleStates(ruleKey ngModels.AlertRuleKey, predicate func(s *State) bool) []*State { +// deleteRuleStates iterates over all states for the given rule and deletes those where predicate returns true. +// The predicate function is called once for each state and should return true if the state should be deleted. +func (c *cache) deleteRuleStates(ruleKey ngModels.AlertRuleKey, predicate func(s *State) bool) { c.mtxStates.Lock() defer c.mtxStates.Unlock() ruleStates, ok := c.states[ruleKey.OrgID][ruleKey.UID] if ok { - return ruleStates.deleteStates(predicate) + ruleStates.deleteStates(predicate) } - return nil } func (c *cache) setRuleStates(ruleKey ngModels.AlertRuleKey, s ruleStates) { diff --git a/pkg/services/ngalert/state/manager.go b/pkg/services/ngalert/state/manager.go index 65af2e40255..c55437c8502 100644 --- a/pkg/services/ngalert/state/manager.go +++ b/pkg/services/ngalert/state/manager.go @@ -351,13 +351,13 @@ func (st *Manager) ProcessEvalResults( logger.Debug("State manager processing evaluation results", "resultCount", len(results)) states := st.setNextStateForRule(ctx, alertRule, results, extraLabels, logger, fn, evaluatedAt) - staleStates := st.deleteStaleStatesFromCache(logger, evaluatedAt, alertRule, fn) + missingSeriesStates, staleCount := st.processMissingSeriesStates(logger, evaluatedAt, alertRule, states, fn) span.AddEvent("results processed", trace.WithAttributes( attribute.Int64("state_transitions", int64(len(states))), - attribute.Int64("stale_states", int64(len(staleStates))), + attribute.Int64("stale_states", staleCount), )) - allChanges := StateTransitions(append(states, staleStates...)) + allChanges := StateTransitions(append(states, missingSeriesStates...)) // It's important that this is done *before* we sync the states to the persister. Otherwise, we will not persist // the LastSentAt field to the store. @@ -385,7 +385,7 @@ func (st *Manager) ProcessEvalResults( func (st *Manager) updateLastSentAt(states StateTransitions, evaluatedAt time.Time) StateTransitions { var result StateTransitions for _, t := range states { - if t.NeedsSending(st.ResendDelay, st.ResolvedRetention) { + if t.NeedsSending(evaluatedAt, st.ResendDelay, st.ResolvedRetention) { t.LastSentAt = &evaluatedAt result = append(result, t) } @@ -516,31 +516,47 @@ func translateInstanceState(state ngModels.InstanceStateType) eval.State { } } -func (st *Manager) deleteStaleStatesFromCache(logger log.Logger, evaluatedAt time.Time, alertRule *ngModels.AlertRule, takeImageFn takeImageFn) []StateTransition { - // If we are removing two or more stale series it makes sense to share the resolved image as the alert rule is the same. - // TODO: We will need to change this when we support images without screenshots as each series will have a different image - staleStates := st.cache.deleteRuleStates(alertRule.GetKey(), func(s *State) bool { - return stateIsStale(evaluatedAt, s.LastEvaluationTime, alertRule.IntervalSeconds, alertRule.GetMissingSeriesEvalsToResolve()) - }) - resolvedStates := make([]StateTransition, 0, len(staleStates)) +// processMissingSeriesStates receives the updated state transitions +// that we got from the alert rule, and checks the cache for any states +// that are not in the current evaluation. The missing states are +// for series that are no longer present in the current evaluation. +// For each missing state, we check if it is stale, and if so, we resolve it. +// At the end we return the missing states so that later they can be sent +// to the alertmanager if needed. +func (st *Manager) processMissingSeriesStates(logger log.Logger, evaluatedAt time.Time, alertRule *ngModels.AlertRule, evalTransitions []StateTransition, takeImageFn takeImageFn) ([]StateTransition, int64) { + missingTransitions := []StateTransition{} + var staleStatesCount int64 = 0 - for _, s := range staleStates { - logger.Info("Detected stale state entry", "cacheID", s.CacheID, "state", s.State, "reason", s.StateReason) + st.cache.deleteRuleStates(alertRule.GetKey(), func(s *State) bool { + // We need only states that are not present in the current evaluation, so + // skip the state if it was just evaluated. + if s.LastEvaluationTime.Equal(evaluatedAt) { + return false + } + // After this point, we know that the state is not in the current evaluation. + // Now we need check if it's stale, and if so, we need to resolve it. oldState := s.State oldReason := s.StateReason + isStale := stateIsStale(evaluatedAt, s.LastEvaluationTime, alertRule.IntervalSeconds, alertRule.GetMissingSeriesEvalsToResolve()) - s.State = eval.Normal - s.StateReason = ngModels.StateReasonMissingSeries - s.EndsAt = evaluatedAt - s.LastEvaluationTime = evaluatedAt + if isStale { + logger.Info("Detected stale state entry", "cacheID", s.CacheID, "state", s.State, "reason", s.StateReason) - // By setting ResolvedAt we trigger the scheduler to send a resolved notification to the Alertmanager. - if s.ShouldBeResolved(oldState) { - s.ResolvedAt = &evaluatedAt - image := takeImageFn("stale state") - if image != nil { - s.Image = image + s.State = eval.Normal + s.StateReason = ngModels.StateReasonMissingSeries + s.LastEvaluationTime = evaluatedAt + s.EndsAt = evaluatedAt + + // By setting ResolvedAt we trigger the scheduler to send a resolved notification to the Alertmanager. + if s.ShouldBeResolved(oldState) { + s.ResolvedAt = &evaluatedAt + image := takeImageFn("stale state") + if image != nil { + s.Image = image + } } + + staleStatesCount++ } record := StateTransition{ @@ -548,9 +564,12 @@ func (st *Manager) deleteStaleStatesFromCache(logger log.Logger, evaluatedAt tim PreviousState: oldState, PreviousStateReason: oldReason, } - resolvedStates = append(resolvedStates, record) - } - return resolvedStates + missingTransitions = append(missingTransitions, record) + + return isStale + }) + + return missingTransitions, staleStatesCount } // stateIsStale determines whether the evaluation state is considered stale. diff --git a/pkg/services/ngalert/state/manager_private_test.go b/pkg/services/ngalert/state/manager_private_test.go index 7c41ee4a859..e4ee2a27422 100644 --- a/pkg/services/ngalert/state/manager_private_test.go +++ b/pkg/services/ngalert/state/manager_private_test.go @@ -653,6 +653,29 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { }, expectedTransitions: map[time.Time][]StateTransition{ t2: { + { + PreviousState: eval.Normal, + State: &State{ + Labels: labels["system + rule + labels1"], + State: eval.Normal, + LatestResult: newEvaluation(t1, eval.Normal), + StartsAt: t1, + EndsAt: t1, + LastEvaluationTime: t1, + }, + }, + { + PreviousState: eval.Alerting, + State: &State{ + Labels: labels["system + rule + labels2"], + State: eval.Alerting, + LatestResult: newEvaluation(t1, eval.Alerting), + StartsAt: t1, + EndsAt: t1.Add(ResendDelay * 4), + LastEvaluationTime: t1, + LastSentAt: &t1, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -1007,6 +1030,18 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { }, }, t2: { + { + PreviousState: eval.Alerting, + State: &State{ + Labels: labels["system + rule + labels1"], + State: eval.Alerting, + LatestResult: newEvaluation(t1, eval.Alerting), + StartsAt: t1, + EndsAt: t1.Add(ResendDelay * 4), + LastEvaluationTime: t1, + LastSentAt: &t1, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -1071,6 +1106,18 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { }, expectedTransitions: map[time.Time][]StateTransition{ t3: { + { + PreviousState: eval.Alerting, + State: &State{ + Labels: labels["system + rule + labels1"], + State: eval.Alerting, + LatestResult: newEvaluation(t1, eval.Alerting), + StartsAt: t1, + EndsAt: t1.Add(ResendDelay * 4), + LastEvaluationTime: t1, + LastSentAt: &t1, + }, + }, { PreviousState: eval.NoData, State: &State{ @@ -1440,6 +1487,19 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { expectedTransitions: map[ngmodels.NoDataState]map[time.Time][]StateTransition{ ngmodels.NoData: { t2: { + { + PreviousState: eval.Normal, + State: &State{ + Labels: labels["system + rule + labels1"], + State: eval.Normal, + LatestResult: newEvaluationWithValues(t1, eval.Normal, map[string]float64{"A": 1}), + StartsAt: t1, + EndsAt: t1, + LastEvaluationTime: t1, + Values: map[string]float64{"A": 1}, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -1457,6 +1517,19 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { }, ngmodels.Alerting: { t2: { + { + PreviousState: eval.Normal, + State: &State{ + Labels: labels["system + rule + labels1"], + State: eval.Normal, + LatestResult: newEvaluationWithValues(t1, eval.Normal, map[string]float64{"A": 1}), + StartsAt: t1, + EndsAt: t1, + LastEvaluationTime: t1, + Values: map[string]float64{"A": 1}, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -1476,6 +1549,19 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { }, ngmodels.OK: { t2: { + { + PreviousState: eval.Normal, + State: &State{ + Labels: labels["system + rule + labels1"], + State: eval.Normal, + LatestResult: newEvaluationWithValues(t1, eval.Normal, map[string]float64{"A": 1}), + StartsAt: t1, + EndsAt: t1, + LastEvaluationTime: t1, + Values: map[string]float64{"A": 1}, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -1494,6 +1580,19 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { }, ngmodels.KeepLast: { t2: { + { + PreviousState: eval.Normal, + State: &State{ + Labels: labels["system + rule + labels1"], + State: eval.Normal, + LatestResult: newEvaluationWithValues(t1, eval.Normal, map[string]float64{"A": 1}), + StartsAt: t1, + EndsAt: t1, + LastEvaluationTime: t1, + Values: map[string]float64{"A": 1}, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -1529,6 +1628,31 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { expectedTransitions: map[ngmodels.NoDataState]map[time.Time][]StateTransition{ ngmodels.NoData: { t2: { + { + PreviousState: eval.Normal, + State: &State{ + Labels: labels["system + rule + labels1"], + State: eval.Normal, + LatestResult: newEvaluation(t1, eval.Normal), + StartsAt: t1, + EndsAt: t1, + LastEvaluationTime: t1, + EvaluationDuration: time.Millisecond * 10, + }, + }, + { + PreviousState: eval.Alerting, + State: &State{ + Labels: labels["system + rule + labels2"], + State: eval.Alerting, + LatestResult: newEvaluation(t1, eval.Alerting), + StartsAt: t1, + EndsAt: t1.Add(ResendDelay * 4), + LastEvaluationTime: t1, + LastSentAt: &t1, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -1585,6 +1709,30 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { }, ngmodels.Alerting: { t2: { + { + PreviousState: eval.Normal, + State: &State{ + Labels: labels["system + rule + labels1"], + State: eval.Normal, + LatestResult: newEvaluation(t1, eval.Normal), + StartsAt: t1, + EndsAt: t1, + LastEvaluationTime: t1, + EvaluationDuration: time.Millisecond * 10, + }, + }, + { + PreviousState: eval.Alerting, + State: &State{ + Labels: labels["system + rule + labels2"], + State: eval.Alerting, + LatestResult: newEvaluation(t1, eval.Alerting), + StartsAt: t1, + EndsAt: t1, + LastEvaluationTime: t1, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -1793,6 +1941,30 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { expectedTransitions: map[ngmodels.NoDataState]map[time.Time][]StateTransition{ ngmodels.NoData: { t2: { + { + PreviousState: eval.Normal, + State: &State{ + Labels: labels["system + rule + labels1"], + State: eval.Normal, + LatestResult: newEvaluation(t1, eval.Normal), + StartsAt: t1, + EndsAt: t1, + LastEvaluationTime: t1, + EvaluationDuration: time.Millisecond * 10, + }, + }, + { + PreviousState: eval.Pending, + State: &State{ + Labels: labels["system + rule + labels2"], + State: eval.Pending, + LatestResult: newEvaluation(t1, eval.Alerting), + StartsAt: t1, + EndsAt: t1.Add(ResendDelay * 4), + LastEvaluationTime: t1, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -2049,6 +2221,19 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { expectedTransitions: map[ngmodels.NoDataState]map[time.Time][]StateTransition{ ngmodels.NoData: { t3: { + { + PreviousState: eval.NoData, + State: &State{ + Labels: labels["system + rule + no-data"], + State: eval.NoData, + LatestResult: newEvaluation(t2, eval.NoData), + StartsAt: t2, + EndsAt: t2.Add(ResendDelay * 4), + LastEvaluationTime: t2, + LastSentAt: &t2, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Pending, State: &State{ @@ -2135,6 +2320,19 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { expectedTransitions: map[ngmodels.NoDataState]map[time.Time][]StateTransition{ ngmodels.NoData: { t2: { + { + PreviousState: eval.Alerting, + State: &State{ + Labels: labels["system + rule"], + State: eval.Alerting, + LatestResult: newEvaluation(t1, eval.Alerting), + StartsAt: t1, + EndsAt: t1.Add(ResendDelay * 4), + LastEvaluationTime: t1, + LastSentAt: &t1, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -2404,6 +2602,20 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { }, }, t4: { + { + PreviousState: eval.NoData, + State: &State{ + Labels: labels["system + rule + no-data"], + Annotations: baseRule.Annotations, + State: eval.NoData, + LatestResult: newEvaluation(t3, eval.NoData), + StartsAt: t2, + EndsAt: t3.Add(ResendDelay * 4), + LastEvaluationTime: t3, + LastSentAt: &t2, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -2430,6 +2642,19 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { }, }, t5: { + { + PreviousState: eval.Normal, + State: &State{ + Labels: labels["system + rule"], + Annotations: baseRule.Annotations, + State: eval.Normal, + LatestResult: newEvaluation(t4, eval.Normal), + StartsAt: t4, + EndsAt: t4, + LastEvaluationTime: t4, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.NoData, State: &State{ @@ -2735,6 +2960,18 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { expectedTransitions: map[ngmodels.NoDataState]map[time.Time][]StateTransition{ ngmodels.NoData: { t2: { + { + PreviousState: eval.Normal, + State: &State{ + Labels: labels["system + rule"], + State: eval.Normal, + LatestResult: newEvaluation(t1, eval.Normal), + StartsAt: t1, + EndsAt: t1, + LastEvaluationTime: t1, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -2819,6 +3056,19 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { expectedTransitions: map[ngmodels.NoDataState]map[time.Time][]StateTransition{ ngmodels.NoData: { t2: { + { + PreviousState: eval.Alerting, + State: &State{ + Labels: labels["system + rule"], + State: eval.Alerting, + LatestResult: newEvaluation(t1, eval.Alerting), + StartsAt: t1, + EndsAt: t1.Add(ResendDelay * 4), + LastEvaluationTime: t1, + LastSentAt: &t1, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -2987,6 +3237,18 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { expectedTransitions: map[ngmodels.NoDataState]map[time.Time][]StateTransition{ ngmodels.NoData: { t2: { + { + PreviousState: eval.Pending, + State: &State{ + Labels: labels["system + rule"], + State: eval.Pending, + LatestResult: newEvaluation(t1, eval.Alerting), + StartsAt: t1, + EndsAt: t1.Add(ResendDelay * 4), + LastEvaluationTime: t1, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -3001,6 +3263,19 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { }, }, t3: { + { + PreviousState: eval.NoData, + State: &State{ + Labels: labels["system + rule + no-data"], + State: eval.NoData, + LatestResult: newEvaluation(t2, eval.NoData), + StartsAt: t2, + EndsAt: t2.Add(ResendDelay * 4), + LastEvaluationTime: t2, + LastSentAt: &t2, + EvaluationDuration: time.Millisecond * 10, + }, + }, { PreviousState: eval.Pending, State: &State{ @@ -3333,6 +3608,18 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { expectedTransitions: map[ngmodels.ExecutionErrorState]map[time.Time][]StateTransition{ ngmodels.ErrorErrState: { t2: { + { + PreviousState: eval.Pending, + State: &State{ + Labels: labels["system + rule + labels1"], + State: eval.Pending, + LatestResult: newEvaluationWithValues(t1, eval.Alerting, map[string]float64{"A": 1.0}), + StartsAt: t1, + EndsAt: t1.Add(ResendDelay * 4), + LastEvaluationTime: t1, + Values: map[string]float64{"A": 1.0}, + }, + }, { PreviousState: eval.Normal, State: &State{ @@ -3424,6 +3711,18 @@ func TestProcessEvalResults_StateTransitions(t *testing.T) { expectedTransitions: map[ngmodels.ExecutionErrorState]map[time.Time][]StateTransition{ ngmodels.ErrorErrState: { t2: { + { + PreviousState: eval.Normal, + State: &State{ + Labels: labels["system + rule + labels1"], + State: eval.Normal, + LatestResult: newEvaluationWithValues(t1, eval.Normal, map[string]float64{"A": 1.0}), + StartsAt: t1, + EndsAt: t1, + LastEvaluationTime: t1, + Values: map[string]float64{"A": 1.0}, + }, + }, { PreviousState: eval.Normal, State: &State{ diff --git a/pkg/services/ngalert/state/manager_test.go b/pkg/services/ngalert/state/manager_test.go index 29f6cab577c..c5206e52355 100644 --- a/pkg/services/ngalert/state/manager_test.go +++ b/pkg/services/ngalert/state/manager_test.go @@ -1401,9 +1401,10 @@ func TestProcessEvalResults(t *testing.T) { statePersister := state.NewSyncStatePersisiter(log.New("ngalert.state.manager.persist"), cfg) st := state.NewManager(cfg, statePersister) rule := models.RuleGen.GenerateRef() - var results = eval.GenerateResults(rand.Intn(4)+1, eval.ResultGen(eval.WithEvaluatedAt(clk.Now()))) + now := clk.Now() + var results = eval.GenerateResults(rand.Intn(4)+1, eval.ResultGen(eval.WithEvaluatedAt(now))) - states := st.ProcessEvalResults(context.Background(), clk.Now(), rule, results, make(data.Labels), nil) + states := st.ProcessEvalResults(context.Background(), now, rule, results, make(data.Labels), nil) require.NotEmpty(t, states) savedStates := make(map[data.Fingerprint]models.AlertInstance) diff --git a/pkg/services/ngalert/state/state.go b/pkg/services/ngalert/state/state.go index 47957428c19..630ff236582 100644 --- a/pkg/services/ngalert/state/state.go +++ b/pkg/services/ngalert/state/state.go @@ -573,7 +573,7 @@ func resultKeepLast(state *State, rule *models.AlertRule, result eval.Result, lo // - The state has been resolved since the last notification. // - The state is firing and the last notification was sent at least resendDelay ago. // - The state was resolved within the resolvedRetention period, and the last notification was sent at least resendDelay ago. -func (a *State) NeedsSending(resendDelay time.Duration, resolvedRetention time.Duration) bool { +func (a *State) NeedsSending(now time.Time, resendDelay time.Duration, resolvedRetention time.Duration) bool { if a.State == eval.Pending { // We do not send notifications for pending states. return false @@ -586,13 +586,13 @@ func (a *State) NeedsSending(resendDelay time.Duration, resolvedRetention time.D // For normal states, we should only be sending if this is a resolved notification or a re-send of the resolved // notification within the resolvedRetention period. - if a.State == eval.Normal && (a.ResolvedAt == nil || a.LastEvaluationTime.Sub(*a.ResolvedAt) > resolvedRetention) { + if a.State == eval.Normal && (a.ResolvedAt == nil || now.Sub(*a.ResolvedAt) > resolvedRetention) { return false } - // We should send, and re-send notifications, each time LastSentAt is <= LastEvaluationTime + resendDelay. + // We should send, and re-send notifications, each time LastSentAt is <= now + resendDelay. // This can include normal->normal transitions that were resolved in recent past evaluations. - return a.LastSentAt == nil || !a.LastSentAt.Add(resendDelay).After(a.LastEvaluationTime) + return a.LastSentAt == nil || !a.LastSentAt.Add(resendDelay).After(now) } func (a *State) Equals(b *State) bool { diff --git a/pkg/services/ngalert/state/state_test.go b/pkg/services/ngalert/state/state_test.go index 8200696d385..d0741d931ad 100644 --- a/pkg/services/ngalert/state/state_test.go +++ b/pkg/services/ngalert/state/state_test.go @@ -510,7 +510,7 @@ func TestNeedsSending(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - assert.Equal(t, tc.expected, tc.testState.NeedsSending(tc.resendDelay, tc.resolvedRetention)) + assert.Equal(t, tc.expected, tc.testState.NeedsSending(evaluationTime, tc.resendDelay, tc.resolvedRetention)) }) } }