Alerting: Backport fix database unavailable removes rules from scheduler (#52140)

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

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},
+			},
+		),
 	}
 }
 

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
 }

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))
+}

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
+}

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)
+}