diff --git a/pkg/services/ngalert/CHANGELOG.md b/pkg/services/ngalert/CHANGELOG.md index f68e9cb5a43..e7b0b27b337 100644 --- a/pkg/services/ngalert/CHANGELOG.md +++ b/pkg/services/ngalert/CHANGELOG.md @@ -72,6 +72,10 @@ Scopes must have an order to ensure consistency and ease of search, this helps u - [ENHANCEMENT] Scheduler: ticker to support stopping #48142 - [ENHANCEMENT] Migration: Don't stop the migration when failing to parse alert rule tags #51253 +## 8.5.9 + +- [ENHANCEMENT] Scheduler: Adds new metrics to track rules that might be scheduled. + ## 8.5.5 - [BUGFIX] Alerting: Remove double quotes from double quoted matchers #50038 diff --git a/pkg/services/ngalert/metrics/ngalert.go b/pkg/services/ngalert/metrics/ngalert.go index bb89cfc2b2b..1b750ba14bf 100644 --- a/pkg/services/ngalert/metrics/ngalert.go +++ b/pkg/services/ngalert/metrics/ngalert.go @@ -50,8 +50,10 @@ type Scheduler struct { EvalTotal *prometheus.CounterVec EvalFailures *prometheus.CounterVec EvalDuration *prometheus.SummaryVec - GetAlertRulesDuration prometheus.Histogram SchedulePeriodicDuration prometheus.Histogram + AlertRules prometheus.Gauge + AlertRulesHash prometheus.Gauge + UpdateAlertRulesDuration prometheus.Histogram } type MultiOrgAlertmanager struct { @@ -161,15 +163,6 @@ func newSchedulerMetrics(r prometheus.Registerer) *Scheduler { }, []string{"org"}, ), - GetAlertRulesDuration: promauto.With(r).NewHistogram( - prometheus.HistogramOpts{ - Namespace: Namespace, - Subsystem: Subsystem, - Name: "get_alert_rules_duration_seconds", - Help: "The time taken to get all alert rules.", - Buckets: []float64{0.1, 0.25, 0.5, 1, 2, 5, 10}, - }, - ), SchedulePeriodicDuration: promauto.With(r).NewHistogram( prometheus.HistogramOpts{ Namespace: Namespace, @@ -179,6 +172,30 @@ func newSchedulerMetrics(r prometheus.Registerer) *Scheduler { Buckets: []float64{0.1, 0.25, 0.5, 1, 2, 5, 10}, }, ), + AlertRules: promauto.With(r).NewGauge( + prometheus.GaugeOpts{ + Namespace: Namespace, + Subsystem: Subsystem, + Name: "schedule_alert_rules", + Help: "The number of alert rules being considered for evaluation each tick.", + }, + ), + AlertRulesHash: promauto.With(r).NewGauge( + prometheus.GaugeOpts{ + Namespace: Namespace, + Subsystem: Subsystem, + Name: "schedule_alert_rules_hash", + Help: "A hash of the alert rules over time.", + }), + UpdateAlertRulesDuration: promauto.With(r).NewHistogram( + prometheus.HistogramOpts{ + Namespace: Namespace, + Subsystem: Subsystem, + Name: "schedule_query_alert_rules_duration_seconds", + Help: "The time taken to fetch alert rules from the database.", + Buckets: []float64{0.1, 0.25, 0.5, 1, 2, 5, 10}, + }, + ), } } diff --git a/pkg/services/ngalert/schedule/fetcher.go b/pkg/services/ngalert/schedule/fetcher.go index 9f20f4ac23d..d7f3a384e68 100644 --- a/pkg/services/ngalert/schedule/fetcher.go +++ b/pkg/services/ngalert/schedule/fetcher.go @@ -2,24 +2,54 @@ package schedule import ( "context" + "fmt" + "hash/fnv" + "sort" "time" "github.com/grafana/grafana/pkg/services/ngalert/models" ) -func (sch *schedule) getAlertRules(ctx context.Context, disabledOrgs []int64) []*models.AlertRule { +// hashUIDs returns a fnv64 hash of the UIDs for all alert rules. +// The order of the alert rules does not matter as hashUIDs sorts +// the UIDs in increasing order. +func hashUIDs(alertRules []*models.AlertRule) uint64 { + h := fnv.New64() + for _, uid := range sortedUIDs(alertRules) { + // We can ignore err as fnv64 does not return an error + // nolint:errcheck,gosec + h.Write([]byte(uid)) + } + return h.Sum64() +} + +// sortedUIDs returns a slice of sorted UIDs. +func sortedUIDs(alertRules []*models.AlertRule) []string { + uids := make([]string, 0, len(alertRules)) + for _, alertRule := range alertRules { + uids = append(uids, alertRule.UID) + } + sort.Strings(uids) + return uids +} + +// updateAlertRules updates the alert rules for the scheduler. It returns an error +// if the database is unavailable or the query returned an error. +func (sch *schedule) updateAlertRules(ctx context.Context, disabledOrgs []int64) error { start := time.Now() defer func() { - sch.metrics.GetAlertRulesDuration.Observe(time.Since(start).Seconds()) + sch.metrics.UpdateAlertRulesDuration.Observe( + time.Since(start).Seconds()) }() q := models.ListAlertRulesQuery{ ExcludeOrgs: disabledOrgs, } - err := sch.ruleStore.GetAlertRulesForScheduling(ctx, &q) - if err != nil { - sch.log.Error("failed to fetch alert definitions", "err", err) - return nil + if err := sch.ruleStore.GetAlertRulesForScheduling(ctx, &q); err != nil { + return fmt.Errorf("failed to get alert rules: %w", err) } - return q.Result + sch.alertRules.set(q.Result) + sch.metrics.AlertRules.Set(float64(len(q.Result))) + sch.metrics.AlertRulesHash.Set(float64(hashUIDs(q.Result))) + return nil } diff --git a/pkg/services/ngalert/schedule/fetcher_test.go b/pkg/services/ngalert/schedule/fetcher_test.go new file mode 100644 index 00000000000..6d93c4237af --- /dev/null +++ b/pkg/services/ngalert/schedule/fetcher_test.go @@ -0,0 +1,26 @@ +package schedule + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/grafana/grafana/pkg/services/ngalert/models" +) + +func TestHashUIDs(t *testing.T) { + r := []*models.AlertRule{{UID: "foo"}, {UID: "bar"}} + assert.Equal(t, uint64(0xade76f55c76a1c48), hashUIDs(r)) + // expect the same hash irrespective of order + r = []*models.AlertRule{{UID: "bar"}, {UID: "foo"}} + assert.Equal(t, uint64(0xade76f55c76a1c48), hashUIDs(r)) + // expect a different hash + r = []*models.AlertRule{{UID: "bar"}} + assert.Equal(t, uint64(0xd8d9a5186bad3880), hashUIDs(r)) + // slice with no items + r = []*models.AlertRule{} + assert.Equal(t, uint64(0xcbf29ce484222325), hashUIDs(r)) + // a different slice with no items should have the same hash + r = []*models.AlertRule{} + assert.Equal(t, uint64(0xcbf29ce484222325), hashUIDs(r)) +} diff --git a/pkg/services/ngalert/schedule/schedule.go b/pkg/services/ngalert/schedule/schedule.go index a15438d5f8d..1ab9659a39a 100644 --- a/pkg/services/ngalert/schedule/schedule.go +++ b/pkg/services/ngalert/schedule/schedule.go @@ -56,7 +56,7 @@ type schedule struct { baseInterval time.Duration // each alert rule gets its own channel and routine - registry alertRuleRegistry + registry alertRuleInfoRegistry maxAttempts int64 @@ -99,6 +99,12 @@ type schedule struct { adminConfigPollInterval time.Duration disabledOrgs map[int64]struct{} minRuleInterval time.Duration + + // alertRules contains the alert rules that are considered for + // evaluation in the current tick. The evaluation of an alert rule in the + // current tick depends on its evaluation interval and when it was + // last evaluated. + alertRules alertRulesRegistry } // SchedulerCfg is the scheduler configuration. @@ -126,7 +132,7 @@ func NewScheduler(cfg SchedulerCfg, expressionService *expr.Service, appURL *url ticker := alerting.NewTicker(cfg.C.Now(), time.Second*0, cfg.C, int64(cfg.BaseInterval.Seconds())) sch := schedule{ - registry: alertRuleRegistry{alertRuleInfo: make(map[models.AlertRuleKey]*alertRuleInfo)}, + registry: alertRuleInfoRegistry{alertRuleInfo: make(map[models.AlertRuleKey]*alertRuleInfo)}, maxAttempts: cfg.MaxAttempts, clock: cfg.C, baseInterval: cfg.BaseInterval, @@ -150,6 +156,7 @@ func NewScheduler(cfg SchedulerCfg, expressionService *expr.Service, appURL *url adminConfigPollInterval: cfg.AdminConfigPollInterval, disabledOrgs: cfg.DisabledOrgs, minRuleInterval: cfg.MinRuleInterval, + alertRules: alertRulesRegistry{rules: make(map[models.AlertRuleKey]*models.AlertRule)}, } return &sch } @@ -334,9 +341,17 @@ func (sch *schedule) UpdateAlertRule(key models.AlertRuleKey) { // DeleteAlertRule stops evaluation of the rule, deletes it from active rules, and cleans up state cache. func (sch *schedule) DeleteAlertRule(key models.AlertRuleKey) { + // It can happen that the scheduler has deleted the alert rule before the + // Ruler API has called DeleteAlertRule. This can happen as requests to + // the Ruler API do not hold an exclusive lock over all scheduler operations. + if _, ok := sch.alertRules.del(key); !ok { + sch.log.Info("alert rule cannot be removed from the scheduler as it is not scheduled", "uid", key.UID, "org_id", key.OrgID) + } + + // Delete the rule routine ruleInfo, ok := sch.registry.del(key) if !ok { - sch.log.Info("unable to delete alert rule routine information by key", "uid", key.UID, "org_id", key.OrgID) + sch.log.Info("alert rule cannot be stopped as it is not running", "uid", key.UID, "org_id", key.OrgID) return } // stop rule evaluation @@ -382,7 +397,11 @@ func (sch *schedule) schedulePeriodic(ctx context.Context) error { disabledOrgs = append(disabledOrgs, disabledOrg) } - alertRules := sch.getAlertRules(ctx, disabledOrgs) + if err := sch.updateAlertRules(ctx, disabledOrgs); err != nil { + sch.log.Error("scheduler failed to update alert rules", "err", err) + } + alertRules := sch.alertRules.all() + sch.log.Debug("alert rules fetched", "count", len(alertRules), "disabled_orgs", disabledOrgs) // registeredDefinitions is a map used for finding deleted alert rules @@ -667,14 +686,14 @@ func (sch *schedule) saveAlertStates(ctx context.Context, states []*state.State) } } -type alertRuleRegistry struct { +type alertRuleInfoRegistry struct { mu sync.Mutex alertRuleInfo map[models.AlertRuleKey]*alertRuleInfo } // getOrCreateInfo gets rule routine information from registry by the key. If it does not exist, it creates a new one. // Returns a pointer to the rule routine information and a flag that indicates whether it is a new struct or not. -func (r *alertRuleRegistry) getOrCreateInfo(context context.Context, key models.AlertRuleKey) (*alertRuleInfo, bool) { +func (r *alertRuleInfoRegistry) getOrCreateInfo(context context.Context, key models.AlertRuleKey) (*alertRuleInfo, bool) { r.mu.Lock() defer r.mu.Unlock() @@ -688,7 +707,7 @@ func (r *alertRuleRegistry) getOrCreateInfo(context context.Context, key models. // get returns the channel for the specific alert rule // if the key does not exist returns an error -func (r *alertRuleRegistry) get(key models.AlertRuleKey) (*alertRuleInfo, error) { +func (r *alertRuleInfoRegistry) get(key models.AlertRuleKey) (*alertRuleInfo, error) { r.mu.Lock() defer r.mu.Unlock() @@ -699,7 +718,7 @@ func (r *alertRuleRegistry) get(key models.AlertRuleKey) (*alertRuleInfo, error) return info, nil } -func (r *alertRuleRegistry) exists(key models.AlertRuleKey) bool { +func (r *alertRuleInfoRegistry) exists(key models.AlertRuleKey) bool { r.mu.Lock() defer r.mu.Unlock() @@ -710,7 +729,7 @@ func (r *alertRuleRegistry) exists(key models.AlertRuleKey) bool { // del removes pair that has specific key from alertRuleInfo. // Returns 2-tuple where the first element is value of the removed pair // and the second element indicates whether element with the specified key existed. -func (r *alertRuleRegistry) del(key models.AlertRuleKey) (*alertRuleInfo, bool) { +func (r *alertRuleInfoRegistry) del(key models.AlertRuleKey) (*alertRuleInfo, bool) { r.mu.Lock() defer r.mu.Unlock() info, ok := r.alertRuleInfo[key] @@ -720,7 +739,7 @@ func (r *alertRuleRegistry) del(key models.AlertRuleKey) (*alertRuleInfo, bool) return info, ok } -func (r *alertRuleRegistry) iter() <-chan models.AlertRuleKey { +func (r *alertRuleInfoRegistry) iter() <-chan models.AlertRuleKey { c := make(chan models.AlertRuleKey) f := func() { @@ -737,7 +756,7 @@ func (r *alertRuleRegistry) iter() <-chan models.AlertRuleKey { return c } -func (r *alertRuleRegistry) keyMap() map[models.AlertRuleKey]struct{} { +func (r *alertRuleInfoRegistry) keyMap() map[models.AlertRuleKey]struct{} { definitionsIDs := make(map[models.AlertRuleKey]struct{}) for k := range r.iter() { definitionsIDs[k] = struct{}{} @@ -811,3 +830,55 @@ func (sch *schedule) stopApplied(alertDefKey models.AlertRuleKey) { sch.stopAppliedFunc(alertDefKey) } + +type alertRulesRegistry struct { + rules map[models.AlertRuleKey]*models.AlertRule + mu sync.Mutex +} + +// all returns all rules in the registry. +func (r *alertRulesRegistry) all() []*models.AlertRule { + r.mu.Lock() + defer r.mu.Unlock() + result := make([]*models.AlertRule, 0, len(r.rules)) + for _, rule := range r.rules { + result = append(result, rule) + } + return result +} + +func (r *alertRulesRegistry) get(k models.AlertRuleKey) *models.AlertRule { + r.mu.Lock() + defer r.mu.Unlock() + return r.rules[k] +} + +// set replaces all rules in the registry. +func (r *alertRulesRegistry) set(rules []*models.AlertRule) { + r.mu.Lock() + defer r.mu.Unlock() + r.rules = make(map[models.AlertRuleKey]*models.AlertRule) + for _, rule := range rules { + r.rules[rule.GetKey()] = rule + } +} + +// update inserts or replaces a rule in the registry. +func (r *alertRulesRegistry) update(rule *models.AlertRule) { + r.mu.Lock() + defer r.mu.Unlock() + r.rules[rule.GetKey()] = rule +} + +// del removes pair that has specific key from alertRulesRegistry. +// Returns 2-tuple where the first element is value of the removed pair +// and the second element indicates whether element with the specified key existed. +func (r *alertRulesRegistry) del(k models.AlertRuleKey) (*models.AlertRule, bool) { + r.mu.Lock() + defer r.mu.Unlock() + rule, ok := r.rules[k] + if ok { + delete(r.rules, k) + } + return rule, ok +} diff --git a/pkg/services/ngalert/schedule/schedule_unit_test.go b/pkg/services/ngalert/schedule/schedule_unit_test.go index 5882a49dc12..5f8d9498f7a 100644 --- a/pkg/services/ngalert/schedule/schedule_unit_test.go +++ b/pkg/services/ngalert/schedule/schedule_unit_test.go @@ -17,6 +17,7 @@ import ( "github.com/grafana/grafana-plugin-sdk-go/data" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/grafana/grafana/pkg/expr" @@ -1117,3 +1118,54 @@ func CreateTestAlertRule(t *testing.T, dbstore *store.FakeRuleStore, intervalSec t.Logf("alert definition: %v with interval: %d created", rule.GetKey(), rule.IntervalSeconds) return rule } + +func TestSchedulableAlertRulesRegistry(t *testing.T) { + r := alertRulesRegistry{rules: make(map[models.AlertRuleKey]*models.AlertRule)} + assert.Len(t, r.all(), 0) + + // replace all rules in the registry with foo + r.set([]*models.AlertRule{{OrgID: 1, UID: "foo", Version: 1}}) + assert.Len(t, r.all(), 1) + foo := r.get(models.AlertRuleKey{OrgID: 1, UID: "foo"}) + require.NotNil(t, foo) + assert.Equal(t, models.AlertRule{OrgID: 1, UID: "foo", Version: 1}, *foo) + + // update foo to a newer version + r.update(&models.AlertRule{OrgID: 1, UID: "foo", Version: 2}) + assert.Len(t, r.all(), 1) + foo = r.get(models.AlertRuleKey{OrgID: 1, UID: "foo"}) + require.NotNil(t, foo) + assert.Equal(t, models.AlertRule{OrgID: 1, UID: "foo", Version: 2}, *foo) + + // update bar which does not exist in the registry + r.update(&models.AlertRule{OrgID: 1, UID: "bar", Version: 1}) + assert.Len(t, r.all(), 2) + foo = r.get(models.AlertRuleKey{OrgID: 1, UID: "foo"}) + require.NotNil(t, foo) + assert.Equal(t, models.AlertRule{OrgID: 1, UID: "foo", Version: 2}, *foo) + bar := r.get(models.AlertRuleKey{OrgID: 1, UID: "bar"}) + require.NotNil(t, foo) + assert.Equal(t, models.AlertRule{OrgID: 1, UID: "bar", Version: 1}, *bar) + + // replace all rules in the registry with baz + r.set([]*models.AlertRule{{OrgID: 1, UID: "baz", Version: 1}}) + assert.Len(t, r.all(), 1) + baz := r.get(models.AlertRuleKey{OrgID: 1, UID: "baz"}) + require.NotNil(t, baz) + assert.Equal(t, models.AlertRule{OrgID: 1, UID: "baz", Version: 1}, *baz) + assert.Nil(t, r.get(models.AlertRuleKey{OrgID: 1, UID: "foo"})) + assert.Nil(t, r.get(models.AlertRuleKey{OrgID: 1, UID: "bar"})) + + // delete baz + deleted, ok := r.del(models.AlertRuleKey{OrgID: 1, UID: "baz"}) + assert.True(t, ok) + require.NotNil(t, deleted) + assert.Equal(t, *deleted, *baz) + assert.Len(t, r.all(), 0) + assert.Nil(t, r.get(models.AlertRuleKey{OrgID: 1, UID: "baz"})) + + // baz cannot be deleted twice + deleted, ok = r.del(models.AlertRuleKey{OrgID: 1, UID: "baz"}) + assert.False(t, ok) + assert.Nil(t, deleted) +}