From 787940f32eae9eea703b5a57009579b7ea705c69 Mon Sep 17 00:00:00 2001 From: "Grot (@grafanabot)" <43478413+grafanabot@users.noreply.github.com> Date: Fri, 25 Feb 2022 04:20:35 -0600 Subject: [PATCH] Rename evalCtx to avoid confusion with context.Context (#45144) (#45893) (cherry picked from commit 2ca79ca0c79f628d8ce414f2a6ee6f9b79e214f1) Co-authored-by: George Robinson --- pkg/services/ngalert/schedule/schedule.go | 46 ++++++------- .../ngalert/schedule/schedule_unit_test.go | 66 +++++++++---------- 2 files changed, 56 insertions(+), 56 deletions(-) diff --git a/pkg/services/ngalert/schedule/schedule.go b/pkg/services/ngalert/schedule/schedule.go index dd72ee0db9f..eaec8aefbb4 100644 --- a/pkg/services/ngalert/schedule/schedule.go +++ b/pkg/services/ngalert/schedule/schedule.go @@ -61,7 +61,7 @@ type schedule struct { clock clock.Clock - heartbeat *alerting.Ticker + ticker *alerting.Ticker // evalApplied is only used for tests: test code can set it to non-nil // function, and then it'll be called from the event loop whenever the @@ -130,7 +130,7 @@ func NewScheduler(cfg SchedulerCfg, expressionService *expr.Service, appURL *url clock: cfg.C, baseInterval: cfg.BaseInterval, log: cfg.Logger, - heartbeat: ticker, + ticker: ticker, evalAppliedFunc: cfg.EvalAppliedFunc, stopAppliedFunc: cfg.StopAppliedFunc, evaluator: cfg.Evaluator, @@ -157,7 +157,7 @@ func (sch *schedule) Pause() error { if sch == nil { return fmt.Errorf("scheduler is not initialised") } - sch.heartbeat.Pause() + sch.ticker.Pause() sch.log.Info("alert rule scheduler paused", "now", sch.clock.Now()) return nil } @@ -166,7 +166,7 @@ func (sch *schedule) Unpause() error { if sch == nil { return fmt.Errorf("scheduler is not initialised") } - sch.heartbeat.Unpause() + sch.ticker.Unpause() sch.log.Info("alert rule scheduler unpaused", "now", sch.clock.Now()) return nil } @@ -367,7 +367,7 @@ func (sch *schedule) schedulePeriodic(ctx context.Context) error { dispatcherGroup, ctx := errgroup.WithContext(ctx) for { select { - case tick := <-sch.heartbeat.C: + case tick := <-sch.ticker.C: start := time.Now() sch.metrics.BehindSeconds.Set(start.Sub(tick).Seconds()) @@ -468,7 +468,7 @@ func (sch *schedule) schedulePeriodic(ctx context.Context) error { } } -func (sch *schedule) ruleRoutine(grafanaCtx context.Context, key models.AlertRuleKey, evalCh <-chan *evalContext, updateCh <-chan struct{}) error { +func (sch *schedule) ruleRoutine(grafanaCtx context.Context, key models.AlertRuleKey, evalCh <-chan *evaluation, updateCh <-chan struct{}) error { logger := sch.log.New("uid", key.UID, "org", key.OrgID) logger.Debug("alert rule routine started") @@ -541,16 +541,16 @@ func (sch *schedule) ruleRoutine(grafanaCtx context.Context, key models.AlertRul return q.Result, nil } - evaluate := func(ctx context.Context, alertRule *models.AlertRule, attempt int64, evalCtx *evalContext) error { - logger := logger.New("version", alertRule.Version, "attempt", attempt, "now", evalCtx.now) + evaluate := func(ctx context.Context, r *models.AlertRule, attempt int64, e *evaluation) error { + logger := logger.New("version", r.Version, "attempt", attempt, "now", e.scheduledAt) start := sch.clock.Now() condition := models.Condition{ - Condition: alertRule.Condition, - OrgID: alertRule.OrgID, - Data: alertRule.Data, + Condition: r.Condition, + OrgID: r.OrgID, + Data: r.Data, } - results, err := sch.evaluator.ConditionEval(&condition, evalCtx.now, sch.expressionService) + results, err := sch.evaluator.ConditionEval(&condition, e.scheduledAt, sch.expressionService) dur := sch.clock.Now().Sub(start) evalTotal.Inc() evalDuration.Observe(dur.Seconds()) @@ -562,7 +562,7 @@ func (sch *schedule) ruleRoutine(grafanaCtx context.Context, key models.AlertRul } logger.Debug("alert rule evaluated", "results", results, "duration", dur) - processedStates := sch.stateManager.ProcessEvalResults(ctx, alertRule, results) + processedStates := sch.stateManager.ProcessEvalResults(ctx, r, results) sch.saveAlertStates(ctx, processedStates) alerts := FromAlertStateToPostableAlerts(processedStates, sch.stateManager, sch.appURL) @@ -616,7 +616,7 @@ func (sch *schedule) ruleRoutine(grafanaCtx context.Context, key models.AlertRul evalRunning = true defer func() { evalRunning = false - sch.evalApplied(key, ctx.now) + sch.evalApplied(key, ctx.scheduledAt) }() err := retryIfError(func(attempt int64) error { @@ -741,7 +741,7 @@ func (r *alertRuleRegistry) keyMap() map[models.AlertRuleKey]struct{} { } type alertRuleInfo struct { - evalCh chan *evalContext + evalCh chan *evaluation updateCh chan struct{} ctx context.Context stop context.CancelFunc @@ -749,15 +749,15 @@ type alertRuleInfo struct { func newAlertRuleInfo(parent context.Context) *alertRuleInfo { ctx, cancel := context.WithCancel(parent) - return &alertRuleInfo{evalCh: make(chan *evalContext), updateCh: make(chan struct{}), ctx: ctx, stop: cancel} + return &alertRuleInfo{evalCh: make(chan *evaluation), updateCh: make(chan struct{}), ctx: ctx, stop: cancel} } // eval signals the rule evaluation routine to perform the evaluation of the rule. Does nothing if the loop is stopped func (a *alertRuleInfo) eval(t time.Time, version int64) bool { select { - case a.evalCh <- &evalContext{ - now: t, - version: version, + case a.evalCh <- &evaluation{ + scheduledAt: t, + version: version, }: return true case <-a.ctx.Done(): @@ -775,16 +775,16 @@ func (a *alertRuleInfo) update() bool { } } -type evalContext struct { - now time.Time - version int64 +type evaluation struct { + scheduledAt time.Time + version int64 } // overrideCfg is only used on tests. func (sch *schedule) overrideCfg(cfg SchedulerCfg) { sch.clock = cfg.C sch.baseInterval = cfg.BaseInterval - sch.heartbeat = alerting.NewTicker(cfg.C.Now(), time.Second*0, cfg.C, int64(cfg.BaseInterval.Seconds())) + sch.ticker = alerting.NewTicker(cfg.C.Now(), time.Second*0, cfg.C, int64(cfg.BaseInterval.Seconds())) sch.evalAppliedFunc = cfg.EvalAppliedFunc sch.stopAppliedFunc = cfg.StopAppliedFunc } diff --git a/pkg/services/ngalert/schedule/schedule_unit_test.go b/pkg/services/ngalert/schedule/schedule_unit_test.go index 69b2f6bf961..6555f11da95 100644 --- a/pkg/services/ngalert/schedule/schedule_unit_test.go +++ b/pkg/services/ngalert/schedule/schedule_unit_test.go @@ -361,7 +361,7 @@ func TestSchedule_ruleRoutine(t *testing.T) { for _, evalState := range normalStates { // 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) { - evalChan := make(chan *evalContext) + evalChan := make(chan *evaluation) evalAppliedChan := make(chan time.Time) sch, ruleStore, instanceStore, _, reg := createSchedule(evalAppliedChan) @@ -375,9 +375,9 @@ func TestSchedule_ruleRoutine(t *testing.T) { expectedTime := time.UnixMicro(rand.Int63()) - evalChan <- &evalContext{ - now: expectedTime, - version: rule.Version, + evalChan <- &evaluation{ + scheduledAt: expectedTime, + version: rule.Version, } actualTime := waitForTimeChannel(t, evalAppliedChan) @@ -467,7 +467,7 @@ func TestSchedule_ruleRoutine(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) go func() { - err := sch.ruleRoutine(ctx, models.AlertRuleKey{}, make(chan *evalContext), make(chan struct{})) + err := sch.ruleRoutine(ctx, models.AlertRuleKey{}, make(chan *evaluation), make(chan struct{})) stoppedChan <- err }() @@ -478,7 +478,7 @@ func TestSchedule_ruleRoutine(t *testing.T) { }) t.Run("should fetch rule from database only if new version is greater than current", func(t *testing.T) { - evalChan := make(chan *evalContext) + evalChan := make(chan *evaluation) evalAppliedChan := make(chan time.Time) ctx := context.Background() @@ -493,9 +493,9 @@ func TestSchedule_ruleRoutine(t *testing.T) { }() expectedTime := time.UnixMicro(rand.Int63()) - evalChan <- &evalContext{ - now: expectedTime, - version: rule.Version, + evalChan <- &evaluation{ + scheduledAt: expectedTime, + version: rule.Version, } actualTime := waitForTimeChannel(t, evalAppliedChan) @@ -508,9 +508,9 @@ func TestSchedule_ruleRoutine(t *testing.T) { // and call with new version expectedTime = expectedTime.Add(time.Duration(rand.Intn(10)) * time.Second) - evalChan <- &evalContext{ - now: expectedTime, - version: newRule.Version, + evalChan <- &evaluation{ + scheduledAt: expectedTime, + version: newRule.Version, } actualTime = waitForTimeChannel(t, evalAppliedChan) @@ -531,7 +531,7 @@ func TestSchedule_ruleRoutine(t *testing.T) { }) t.Run("should not fetch rule if version is equal or less than current", func(t *testing.T) { - evalChan := make(chan *evalContext) + evalChan := make(chan *evaluation) evalAppliedChan := make(chan time.Time) sch, ruleStore, _, _, _ := createSchedule(evalAppliedChan) @@ -545,9 +545,9 @@ func TestSchedule_ruleRoutine(t *testing.T) { }() expectedTime := time.UnixMicro(rand.Int63()) - evalChan <- &evalContext{ - now: expectedTime, - version: rule.Version, + evalChan <- &evaluation{ + scheduledAt: expectedTime, + version: rule.Version, } actualTime := waitForTimeChannel(t, evalAppliedChan) @@ -555,17 +555,17 @@ func TestSchedule_ruleRoutine(t *testing.T) { // try again with the same version expectedTime = expectedTime.Add(time.Duration(rand.Intn(10)) * time.Second) - evalChan <- &evalContext{ - now: expectedTime, - version: rule.Version, + evalChan <- &evaluation{ + scheduledAt: expectedTime, + version: rule.Version, } actualTime = waitForTimeChannel(t, evalAppliedChan) require.Equal(t, expectedTime, actualTime) expectedTime = expectedTime.Add(time.Duration(rand.Intn(10)) * time.Second) - evalChan <- &evalContext{ - now: expectedTime, - version: rule.Version - 1, + evalChan <- &evaluation{ + scheduledAt: expectedTime, + version: rule.Version - 1, } actualTime = waitForTimeChannel(t, evalAppliedChan) require.Equal(t, expectedTime, actualTime) @@ -582,7 +582,7 @@ func TestSchedule_ruleRoutine(t *testing.T) { t.Run("when update channel is not empty", func(t *testing.T) { t.Run("should fetch the alert rule from database", func(t *testing.T) { - evalChan := make(chan *evalContext) + evalChan := make(chan *evaluation) evalAppliedChan := make(chan time.Time) updateChan := make(chan struct{}) @@ -612,9 +612,9 @@ func TestSchedule_ruleRoutine(t *testing.T) { require.Equal(t, rule.OrgID, m.OrgID) // now call evaluation loop to make sure that the rule was persisted - evalChan <- &evalContext{ - now: time.UnixMicro(rand.Int63()), - version: rule.Version, + evalChan <- &evaluation{ + scheduledAt: time.UnixMicro(rand.Int63()), + version: rule.Version, } waitForTimeChannel(t, evalAppliedChan) @@ -637,7 +637,7 @@ func TestSchedule_ruleRoutine(t *testing.T) { go func() { ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) - _ = sch.ruleRoutine(ctx, rule.GetKey(), make(chan *evalContext), updateChan) + _ = sch.ruleRoutine(ctx, rule.GetKey(), make(chan *evaluation), updateChan) }() ruleStore.hook = func(cmd interface{}) error { @@ -677,7 +677,7 @@ func TestSchedule_ruleRoutine(t *testing.T) { return len(s.Alertmanagers()) == 1 }, 20*time.Second, 200*time.Millisecond, "external Alertmanager was not discovered.") - evalChan := make(chan *evalContext) + evalChan := make(chan *evaluation) evalAppliedChan := make(chan time.Time) updateChan := make(chan struct{}) @@ -783,7 +783,7 @@ func TestSchedule_ruleRoutine(t *testing.T) { return len(s.Alertmanagers()) == 1 }, 20*time.Second, 200*time.Millisecond, "external Alertmanager was not discovered.") - evalChan := make(chan *evalContext) + evalChan := make(chan *evaluation) evalAppliedChan := make(chan time.Time) sch, ruleStore, _, _, _ := createSchedule(evalAppliedChan) @@ -797,9 +797,9 @@ func TestSchedule_ruleRoutine(t *testing.T) { _ = sch.ruleRoutine(ctx, rule.GetKey(), evalChan, make(chan struct{})) }() - evalChan <- &evalContext{ - now: time.Now(), - version: rule.Version, + evalChan <- &evaluation{ + scheduledAt: time.Now(), + version: rule.Version, } waitForTimeChannel(t, evalAppliedChan) @@ -843,7 +843,7 @@ func TestSchedule_alertRuleInfo(t *testing.T) { select { case ctx := <-r.evalCh: require.Equal(t, version, ctx.version) - require.Equal(t, expected, ctx.now) + require.Equal(t, expected, ctx.scheduledAt) require.True(t, <-resultCh) case <-time.After(5 * time.Second): t.Fatal("No message was received on eval channel")