Alerting: Update rules version when folder title is updated (grafana#53013)

* remove support for bus from scheduler
* rename event to FolderTitleUpdated and fire only if title has changed
* add method to increase version of all rules that belong to a folder
* update ngalert service to subscribe to folder title change event call data store and update scheduler
* add tests

Author: yuri-tceretian

pkg/events/events.go

@@ -71,7 +71,7 @@ type DataSourceCreated struct {
 	OrgID     int64     `json:"org_id"`
 }
 
-type FolderUpdated struct {
+type FolderTitleUpdated struct {
 	Timestamp time.Time `json:"timestamp"`
 	Title     string    `json:"name"`
 	ID        int64     `json:"id"`

pkg/services/dashboards/service/folder_service.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"strings"
-	"time"
 
 	"github.com/grafana/grafana/pkg/bus"
 	"github.com/grafana/grafana/pkg/events"
@@ -207,6 +206,7 @@ func (f *FolderServiceImpl) UpdateFolder(ctx context.Context, user *models.Signe
 	}
 
 	dashFolder := query.Result
+	currentTitle := dashFolder.Title
 
 	if !dashFolder.IsFolder {
 		return dashboards.ErrFolderNotFound
@@ -238,14 +238,16 @@ func (f *FolderServiceImpl) UpdateFolder(ctx context.Context, user *models.Signe
 	}
 	cmd.Result = folder
 
-	if err := f.bus.Publish(ctx, &events.FolderUpdated{
-		Timestamp: time.Now(),
-		Title:     folder.Title,
-		ID:        dash.Id,
-		UID:       dash.Uid,
-		OrgID:     orgID,
-	}); err != nil {
-		f.log.Error("failed to publish FolderUpdated event", "folder", folder.Title, "user", user.UserId, "error", err)
+	if currentTitle != folder.Title {
+		if err := f.bus.Publish(ctx, &events.FolderTitleUpdated{
+			Timestamp: folder.Updated,
+			Title:     folder.Title,
+			ID:        dash.Id,
+			UID:       dash.Uid,
+			OrgID:     orgID,
+		}); err != nil {
+			f.log.Error("failed to publish FolderTitleUpdated event", "folder", folder.Title, "user", user.UserId, "error", err)
+		}
 	}
 
 	return nil

pkg/services/ngalert/models/alert_rule.go

@@ -188,8 +188,13 @@ func (alertRule *AlertRule) Diff(rule *AlertRule, ignore ...string) cmputil.Diff
 
 // AlertRuleKey is the alert definition identifier
 type AlertRuleKey struct {
-	OrgID int64
-	UID   string
+	OrgID int64  `xorm:"org_id"`
+	UID   string `xorm:"uid"`
+}
+
+type AlertRuleKeyWithVersion struct {
+	Version      int64
+	AlertRuleKey `xorm:"extends"`
 }
 
 // AlertRuleGroupKey is the identifier of a group of alerts

pkg/services/ngalert/models/testing.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/grafana/grafana-plugin-sdk-go/data"
 
+	models2 "github.com/grafana/grafana/pkg/models"
 	"github.com/grafana/grafana/pkg/util"
 )
 
@@ -132,6 +133,18 @@ func WithSequentialGroupIndex() AlertRuleMutator {
 	}
 }
 
+func WithOrgID(orgId int64) AlertRuleMutator {
+	return func(rule *AlertRule) {
+		rule.OrgID = orgId
+	}
+}
+
+func WithNamespace(namespace *models2.Folder) AlertRuleMutator {
+	return func(rule *AlertRule) {
+		rule.NamespaceUID = namespace.Uid
+	}
+}
+
 func GenerateAlertLabels(count int, prefix string) data.Labels {
 	labels := make(data.Labels, count)
 	for i := 0; i < count; i++ {

pkg/services/ngalert/ngalert.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/grafana/grafana/pkg/api/routing"
 	"github.com/grafana/grafana/pkg/bus"
+	"github.com/grafana/grafana/pkg/events"
 	"github.com/grafana/grafana/pkg/expr"
 	"github.com/grafana/grafana/pkg/infra/kvstore"
 	"github.com/grafana/grafana/pkg/infra/log"
@@ -21,6 +22,7 @@ import (
 	"github.com/grafana/grafana/pkg/services/ngalert/eval"
 	"github.com/grafana/grafana/pkg/services/ngalert/image"
 	"github.com/grafana/grafana/pkg/services/ngalert/metrics"
+	"github.com/grafana/grafana/pkg/services/ngalert/models"
 	"github.com/grafana/grafana/pkg/services/ngalert/notifier"
 	"github.com/grafana/grafana/pkg/services/ngalert/provisioning"
 	"github.com/grafana/grafana/pkg/services/ngalert/schedule"
@@ -162,7 +164,12 @@ func (ng *AlertNG) init() error {
 	}
 
 	stateManager := state.NewManager(ng.Log, ng.Metrics.GetStateMetrics(), appUrl, store, store, ng.dashboardService, ng.imageService, clk)
-	scheduler := schedule.NewScheduler(schedCfg, appUrl, stateManager, ng.bus)
+	scheduler := schedule.NewScheduler(schedCfg, appUrl, stateManager)
+
+	// if it is required to include folder title to the alerts, we need to subscribe to changes of alert title
+	if !ng.Cfg.UnifiedAlerting.ReservedLabels.IsReservedLabelDisabled(models.FolderTitleLabel) {
+		subscribeToFolderChanges(ng.Log, ng.bus, store, scheduler)
+	}
 
 	ng.stateManager = stateManager
 	ng.schedule = scheduler
@@ -207,6 +214,31 @@ func (ng *AlertNG) init() error {
 	return DeclareFixedRoles(ng.accesscontrol)
 }
 
+func subscribeToFolderChanges(logger log.Logger, bus bus.Bus, dbStore store.RuleStore, scheduler schedule.ScheduleService) {
+	// if folder title is changed, we update all alert rules in that folder to make sure that all peers (in HA mode) will update folder title and
+	// clean up the current state
+	bus.AddEventListener(func(ctx context.Context, e *events.FolderTitleUpdated) error {
+		// do not block the upstream execution
+		go func(evt *events.FolderTitleUpdated) {
+			logger.Debug("Got folder title updated event. updating rules in the folder", "folder_uid", evt.UID)
+			updated, err := dbStore.IncreaseVersionForAllRulesInNamespace(context.Background(), evt.OrgID, evt.UID)
+			if err != nil {
+				logger.Error("Failed to update alert rules in the folder after its title was changed", "err", err, "folder_uid", evt.UID, "folder", evt.Title)
+				return
+			}
+			if len(updated) > 0 {
+				logger.Debug("rules that belong to the folder have been updated successfully. clearing their status", "updated_rules", len(updated))
+				for _, key := range updated {
+					scheduler.UpdateAlertRule(key.AlertRuleKey, key.Version)
+				}
+			} else {
+				logger.Debug("no alert rules found in the folder. nothing to update", "folder_uid", evt.UID, "folder", evt.Title)
+			}
+		}(e)
+		return nil
+	})
+}
+
 // Run starts the scheduler and Alertmanager.
 func (ng *AlertNG) Run(ctx context.Context) error {
 	ng.Log.Debug("ngalert starting")

pkg/services/ngalert/ngalert_test.go

@@ -0,0 +1,73 @@
+package ngalert
+
+import (
+	"context"
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	busmock "github.com/grafana/grafana/pkg/bus/mock"
+	"github.com/grafana/grafana/pkg/events"
+	"github.com/grafana/grafana/pkg/infra/log"
+	models2 "github.com/grafana/grafana/pkg/models"
+	"github.com/grafana/grafana/pkg/services/ngalert/models"
+	"github.com/grafana/grafana/pkg/services/ngalert/schedule"
+	"github.com/grafana/grafana/pkg/services/ngalert/store"
+	"github.com/grafana/grafana/pkg/util"
+)
+
+func Test_subscribeToFolderChanges(t *testing.T) {
+	orgID := rand.Int63()
+	folder := &models2.Folder{
+		Id:    0,
+		Uid:   util.GenerateShortUID(),
+		Title: "Folder" + util.GenerateShortUID(),
+	}
+	rules := models.GenerateAlertRules(5, models.AlertRuleGen(models.WithOrgID(orgID), models.WithNamespace(folder)))
+
+	bus := busmock.New()
+	db := store.NewFakeRuleStore(t)
+	db.Folders[orgID] = append(db.Folders[orgID], folder)
+	db.PutRule(context.Background(), rules...)
+
+	scheduler := &schedule.FakeScheduleService{}
+	scheduler.EXPECT().UpdateAlertRule(mock.Anything, mock.Anything).Return()
+
+	subscribeToFolderChanges(log.New("test"), bus, db, scheduler)
+
+	err := bus.Publish(context.Background(), &events.FolderTitleUpdated{
+		Timestamp: time.Now(),
+		Title:     "Folder" + util.GenerateShortUID(),
+		ID:        folder.Id,
+		UID:       folder.Uid,
+		OrgID:     orgID,
+	})
+	require.NoError(t, err)
+
+	require.Eventuallyf(t, func() bool {
+		return len(db.GetRecordedCommands(func(cmd interface{}) (interface{}, bool) {
+			c, ok := cmd.(store.GenericRecordedQuery)
+			if !ok || c.Name != "IncreaseVersionForAllRulesInNamespace" {
+				return nil, false
+			}
+			return c, true
+		})) > 0
+	}, time.Second, 10*time.Millisecond, "expected to call db store method but nothing was called")
+
+	var calledTimes int
+	require.Eventuallyf(t, func() bool {
+		for _, call := range scheduler.Calls {
+			if call.Method == "UpdateAlertRule" {
+				calledTimes++
+			}
+		}
+		return calledTimes == len(rules)
+	}, time.Second, 10*time.Millisecond, "scheduler was expected to be called %d times but called %d", len(rules), calledTimes)
+
+	for _, rule := range rules {
+		scheduler.AssertCalled(t, "UpdateAlertRule", rule.GetKey(), rule.Version)
+	}
+}

pkg/services/ngalert/schedule/schedule.go

@@ -8,8 +8,6 @@ import (
 
 	prometheusModel "github.com/prometheus/common/model"
 
-	"github.com/grafana/grafana/pkg/bus"
-	"github.com/grafana/grafana/pkg/events"
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/models"
 	"github.com/grafana/grafana/pkg/services/alerting"
@@ -27,23 +25,19 @@ import (
 
 // ScheduleService is an interface for a service that schedules the evaluation
 // of alert rules.
-//go:generate mockery --name ScheduleService --structname FakeScheduleService --inpackage --filename schedule_mock.go
+//go:generate mockery --name ScheduleService --structname FakeScheduleService --inpackage --filename schedule_mock.go --with-expecter
 type ScheduleService interface {
 	// Run the scheduler until the context is canceled or the scheduler returns
 	// an error. The scheduler is terminated when this function returns.
 	Run(context.Context) error
 	// UpdateAlertRule notifies scheduler that a rule has been changed
 	UpdateAlertRule(key ngmodels.AlertRuleKey, lastVersion int64)
-	// UpdateAlertRulesByNamespaceUID notifies scheduler that all rules in a namespace should be updated.
-	UpdateAlertRulesByNamespaceUID(ctx context.Context, orgID int64, uid string) error
 	// DeleteAlertRule notifies scheduler that a rule has been changed
 	DeleteAlertRule(key ngmodels.AlertRuleKey)
 	// the following are used by tests only used for tests
 	evalApplied(ngmodels.AlertRuleKey, time.Time)
 	stopApplied(ngmodels.AlertRuleKey)
 	overrideCfg(cfg SchedulerCfg)
-
-	folderUpdateHandler(ctx context.Context, evt *events.FolderUpdated) error
 }
 
 //go:generate mockery --name AlertsSender --structname AlertsSenderMock --inpackage --filename alerts_sender_mock.go --with-expecter
@@ -97,9 +91,6 @@ type schedule struct {
 	// current tick depends on its evaluation interval and when it was
 	// last evaluated.
 	schedulableAlertRules alertRulesRegistry
-
-	// bus is used to hook into events that should cause rule updates.
-	bus bus.Bus
 }
 
 // SchedulerCfg is the scheduler configuration.
@@ -117,7 +108,7 @@ type SchedulerCfg struct {
 }
 
 // NewScheduler returns a new schedule.
-func NewScheduler(cfg SchedulerCfg, appURL *url.URL, stateManager *state.Manager, bus bus.Bus) *schedule {
+func NewScheduler(cfg SchedulerCfg, appURL *url.URL, stateManager *state.Manager) *schedule {
 	ticker := alerting.NewTicker(cfg.C, cfg.Cfg.BaseInterval, cfg.Metrics.Ticker)
 
 	sch := schedule{
@@ -138,12 +129,9 @@ func NewScheduler(cfg SchedulerCfg, appURL *url.URL, stateManager *state.Manager
 		stateManager:          stateManager,
 		minRuleInterval:       cfg.Cfg.MinInterval,
 		schedulableAlertRules: alertRulesRegistry{rules: make(map[ngmodels.AlertRuleKey]*ngmodels.AlertRule)},
-		bus:                   bus,
 		alertsSender:          cfg.AlertSender,
 	}
 
-	bus.AddEventListener(sch.folderUpdateHandler)
-
 	return &sch
 }
 
@@ -165,26 +153,6 @@ func (sch *schedule) UpdateAlertRule(key ngmodels.AlertRuleKey, lastVersion int6
 	ruleInfo.update(ruleVersion(lastVersion))
 }
 
-// UpdateAlertRulesByNamespaceUID looks for the active rule evaluation for every rule in the given namespace and commands it to update the rule.
-func (sch *schedule) UpdateAlertRulesByNamespaceUID(ctx context.Context, orgID int64, uid string) error {
-	q := ngmodels.ListAlertRulesQuery{
-		OrgID:         orgID,
-		NamespaceUIDs: []string{uid},
-	}
-	if err := sch.ruleStore.ListAlertRules(ctx, &q); err != nil {
-		return err
-	}
-
-	for _, r := range q.Result {
-		sch.UpdateAlertRule(ngmodels.AlertRuleKey{
-			OrgID: orgID,
-			UID:   r.UID,
-		}, r.Version)
-	}
-
-	return nil
-}
-
 // DeleteAlertRule stops evaluation of the rule, deletes it from active rules, and cleans up state cache.
 func (sch *schedule) DeleteAlertRule(key ngmodels.AlertRuleKey) {
 	// It can happen that the scheduler has deleted the alert rule before the
@@ -465,14 +433,6 @@ func (sch *schedule) saveAlertStates(ctx context.Context, states []*state.State)
 	}
 }
 
-// folderUpdateHandler listens for folder update events and updates all rules in the given folder.
-func (sch *schedule) folderUpdateHandler(ctx context.Context, evt *events.FolderUpdated) error {
-	if sch.disableGrafanaFolder {
-		return nil
-	}
-	return sch.UpdateAlertRulesByNamespaceUID(ctx, evt.OrgID, evt.UID)
-}
-
 // overrideCfg is only used on tests.
 func (sch *schedule) overrideCfg(cfg SchedulerCfg) {
 	sch.clock = cfg.C