Alerting: update rule versions on folder move (#88376)
* Alerting: update rule versions on folder move (#88361) * Add tracing to folder.Move and folder.Update
This commit is contained in:
Alexander Akhmetov
@@ -1819,7 +1819,7 @@ func createTestEnv(t *testing.T, testConfig string) testEnvironment {
|
||||
require.NoError(t, err)
|
||||
|
||||
folderStore := folderimpl.ProvideDashboardFolderStore(sqlStore)
|
||||
folderService := folderimpl.ProvideService(actest.FakeAccessControl{ExpectedEvaluate: true}, bus.ProvideBus(tracing.InitializeTracerForTest()), dashboardStore, folderStore, sqlStore, featuremgmt.WithFeatures(), supportbundlestest.NewFakeBundleService(), nil)
|
||||
folderService := folderimpl.ProvideService(actest.FakeAccessControl{ExpectedEvaluate: true}, bus.ProvideBus(tracing.InitializeTracerForTest()), dashboardStore, folderStore, sqlStore, featuremgmt.WithFeatures(), supportbundlestest.NewFakeBundleService(), nil, tracing.InitializeTracerForTest())
|
||||
store := store.DBstore{
|
||||
Logger: log,
|
||||
SQLStore: sqlStore,
|
||||
|
||||
@@ -26,8 +26,8 @@ type RuleStore interface {
|
||||
UpdateAlertRules(ctx context.Context, rule []ngmodels.UpdateRule) error
|
||||
DeleteAlertRulesByUID(ctx context.Context, orgID int64, ruleUID ...string) error
|
||||
|
||||
// IncreaseVersionForAllRulesInNamespace Increases version for all rules that have specified namespace. Returns all rules that belong to the namespace
|
||||
IncreaseVersionForAllRulesInNamespace(ctx context.Context, orgID int64, namespaceUID string) ([]ngmodels.AlertRuleKeyWithVersion, error)
|
||||
// IncreaseVersionForAllRulesInNamespaces Increases version for all rules that have specified namespace uids
|
||||
IncreaseVersionForAllRulesInNamespaces(ctx context.Context, orgID int64, namespaceUIDs []string) ([]ngmodels.AlertRuleKeyWithVersion, error)
|
||||
|
||||
accesscontrol.RuleUIDToNamespaceStore
|
||||
}
|
||||
|
||||
@@ -487,15 +487,16 @@ func (ng *AlertNG) init() error {
|
||||
}
|
||||
|
||||
func subscribeToFolderChanges(logger log.Logger, bus bus.Bus, dbStore api.RuleStore) {
|
||||
// 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
|
||||
// if full path to the folder 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, evt *events.FolderTitleUpdated) error {
|
||||
logger.Info("Got folder title updated event. updating rules in the folder", "folderUID", evt.UID)
|
||||
_, err := dbStore.IncreaseVersionForAllRulesInNamespace(ctx, evt.OrgID, evt.UID)
|
||||
bus.AddEventListener(func(ctx context.Context, evt *events.FolderFullPathUpdated) error {
|
||||
logger.Info("Got folder full path updated event. updating rules in the folders", "folderUIDs", evt.UIDs)
|
||||
updatedKeys, err := dbStore.IncreaseVersionForAllRulesInNamespaces(ctx, evt.OrgID, evt.UIDs)
|
||||
if err != nil {
|
||||
logger.Error("Failed to update alert rules in the folder after its title was changed", "error", err, "folderUID", evt.UID, "folder", evt.Title)
|
||||
logger.Error("Failed to update alert rules in the folders after their full paths were changed", "error", err, "folderUIDs", evt.UIDs, "orgID", evt.OrgID)
|
||||
return err
|
||||
}
|
||||
logger.Info("Updated version for alert rules", "keys", updatedKeys)
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
|
||||
"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/bus"
|
||||
@@ -25,37 +26,52 @@ import (
|
||||
)
|
||||
|
||||
func Test_subscribeToFolderChanges(t *testing.T) {
|
||||
getRecordedCommand := func(ruleStore *fakes.RuleStore) []fakes.GenericRecordedQuery {
|
||||
results := ruleStore.GetRecordedCommands(func(cmd any) (any, bool) {
|
||||
c, ok := cmd.(fakes.GenericRecordedQuery)
|
||||
if !ok || c.Name != "IncreaseVersionForAllRulesInNamespaces" {
|
||||
return nil, false
|
||||
}
|
||||
return c, ok
|
||||
})
|
||||
var result []fakes.GenericRecordedQuery
|
||||
for _, cmd := range results {
|
||||
result = append(result, cmd.(fakes.GenericRecordedQuery))
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
orgID := rand.Int63()
|
||||
folder := &folder.Folder{
|
||||
folder1 := &folder.Folder{
|
||||
UID: util.GenerateShortUID(),
|
||||
Title: "Folder" + util.GenerateShortUID(),
|
||||
}
|
||||
folder2 := &folder.Folder{
|
||||
UID: util.GenerateShortUID(),
|
||||
Title: "Folder" + util.GenerateShortUID(),
|
||||
}
|
||||
gen := models.RuleGen
|
||||
rules := gen.With(gen.WithOrgID(orgID), gen.WithNamespace(folder)).GenerateManyRef(5)
|
||||
rules := gen.With(gen.WithOrgID(orgID), gen.WithNamespace(folder1)).GenerateManyRef(5)
|
||||
|
||||
bus := bus.ProvideBus(tracing.InitializeTracerForTest())
|
||||
db := fakes.NewRuleStore(t)
|
||||
db.Folders[orgID] = append(db.Folders[orgID], folder)
|
||||
db.Folders[orgID] = append(db.Folders[orgID], folder1)
|
||||
db.PutRule(context.Background(), rules...)
|
||||
|
||||
subscribeToFolderChanges(log.New("test"), bus, db)
|
||||
|
||||
err := bus.Publish(context.Background(), &events.FolderTitleUpdated{
|
||||
err := bus.Publish(context.Background(), &events.FolderFullPathUpdated{
|
||||
Timestamp: time.Now(),
|
||||
Title: "Folder" + util.GenerateShortUID(),
|
||||
UID: folder.UID,
|
||||
UIDs: []string{folder1.UID, folder2.UID},
|
||||
OrgID: orgID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Eventuallyf(t, func() bool {
|
||||
return len(db.GetRecordedCommands(func(cmd any) (any, bool) {
|
||||
c, ok := cmd.(fakes.GenericRecordedQuery)
|
||||
if !ok || c.Name != "IncreaseVersionForAllRulesInNamespace" {
|
||||
return nil, false
|
||||
}
|
||||
return c, true
|
||||
})) > 0
|
||||
require.EventuallyWithT(t, func(c *assert.CollectT) {
|
||||
recordedCommands := getRecordedCommand(db)
|
||||
require.Len(c, recordedCommands, 1)
|
||||
require.Equal(c, recordedCommands[0].Params[0].(int64), orgID)
|
||||
require.ElementsMatch(c, recordedCommands[0].Params[1].([]string), []string{folder1.UID, folder2.UID})
|
||||
}, time.Second, 10*time.Millisecond, "expected to call db store method but nothing was called")
|
||||
}
|
||||
|
||||
|
||||
@@ -1479,7 +1479,7 @@ func TestProvisiongWithFullpath(t *testing.T) {
|
||||
_, dashboardStore := testutil.SetupDashboardService(t, sqlStore, folderStore, cfg)
|
||||
ac := acmock.New()
|
||||
features := featuremgmt.WithFeatures(featuremgmt.FlagNestedFolders)
|
||||
folderService := folderimpl.ProvideService(ac, inProcBus, dashboardStore, folderStore, sqlStore, features, supportbundlestest.NewFakeBundleService(), nil)
|
||||
folderService := folderimpl.ProvideService(ac, inProcBus, dashboardStore, folderStore, sqlStore, features, supportbundlestest.NewFakeBundleService(), nil, tracing.InitializeTracerForTest())
|
||||
|
||||
ruleService := createAlertRuleService(t, folderService)
|
||||
var orgID int64 = 1
|
||||
|
||||
@@ -75,16 +75,26 @@ func (st DBstore) DeleteAlertRulesByUID(ctx context.Context, orgID int64, ruleUI
|
||||
})
|
||||
}
|
||||
|
||||
// IncreaseVersionForAllRulesInNamespace Increases version for all rules that have specified namespace. Returns all rules that belong to the namespace
|
||||
func (st DBstore) IncreaseVersionForAllRulesInNamespace(ctx context.Context, orgID int64, namespaceUID string) ([]ngmodels.AlertRuleKeyWithVersion, error) {
|
||||
// IncreaseVersionForAllRulesInNamespaces Increases version for all rules that have specified namespace. Returns all rules that belong to the namespaces
|
||||
func (st DBstore) IncreaseVersionForAllRulesInNamespaces(ctx context.Context, orgID int64, namespaceUIDs []string) ([]ngmodels.AlertRuleKeyWithVersion, error) {
|
||||
var keys []ngmodels.AlertRuleKeyWithVersion
|
||||
err := st.SQLStore.WithTransactionalDbSession(ctx, func(sess *db.Session) error {
|
||||
now := TimeNow()
|
||||
_, err := sess.Exec("UPDATE alert_rule SET version = version + 1, updated = ? WHERE namespace_uid = ? AND org_id = ?", now, namespaceUID, orgID)
|
||||
namespaceUIDsArgs, in := getINSubQueryArgs(namespaceUIDs)
|
||||
sql := fmt.Sprintf(
|
||||
"UPDATE alert_rule SET version = version + 1, updated = ? WHERE org_id = ? AND namespace_uid IN (%s)",
|
||||
strings.Join(in, ","),
|
||||
)
|
||||
args := make([]interface{}, 0, 3+len(namespaceUIDsArgs))
|
||||
args = append(args, sql, now, orgID)
|
||||
args = append(args, namespaceUIDsArgs...)
|
||||
|
||||
_, err := sess.Exec(args...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return sess.Table(ngmodels.AlertRule{}).Where("namespace_uid = ? AND org_id = ?", namespaceUID, orgID).Find(&keys)
|
||||
|
||||
return sess.Table(ngmodels.AlertRule{}).Where("org_id = ?", orgID).In("namespace_uid", namespaceUIDs).Find(&keys)
|
||||
})
|
||||
return keys, err
|
||||
}
|
||||
|
||||
@@ -1184,6 +1184,58 @@ func TestIntegrationRuleGroupsCaseSensitive(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestIncreaseVersionForAllRulesInNamespaces(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
cfg := setting.NewCfg()
|
||||
cfg.UnifiedAlerting = setting.UnifiedAlertingSettings{BaseInterval: time.Duration(rand.Int63n(100)+1) * time.Second}
|
||||
sqlStore := db.InitTestReplDB(t)
|
||||
store := &DBstore{
|
||||
SQLStore: sqlStore,
|
||||
Cfg: cfg.UnifiedAlerting,
|
||||
FolderService: setupFolderService(t, sqlStore, cfg, featuremgmt.WithFeatures()),
|
||||
Logger: &logtest.Fake{},
|
||||
}
|
||||
orgID := int64(1)
|
||||
gen := models.RuleGen
|
||||
gen = gen.With(gen.WithIntervalMatching(store.Cfg.BaseInterval)).With(gen.WithOrgID(orgID))
|
||||
|
||||
alertRules := []*models.AlertRule{}
|
||||
for i := 0; i < 5; i++ {
|
||||
alertRules = append(alertRules, createRule(t, store, gen))
|
||||
}
|
||||
alertRuleNamespaceUIDs := make([]string, 0, len(alertRules))
|
||||
for _, rule := range alertRules {
|
||||
alertRuleNamespaceUIDs = append(alertRuleNamespaceUIDs, rule.NamespaceUID)
|
||||
}
|
||||
alertRuleInAnotherNamespace := createRule(t, store, gen)
|
||||
|
||||
requireAlertRuleVersion := func(t *testing.T, ruleID int64, orgID int64, expectedVersion int64) {
|
||||
t.Helper()
|
||||
dbrule := &models.AlertRule{}
|
||||
err := sqlStore.WithDbSession(context.Background(), func(sess *db.Session) error {
|
||||
exist, err := sess.Table(models.AlertRule{}).ID(ruleID).Get(dbrule)
|
||||
require.Truef(t, exist, fmt.Sprintf("rule with ID %d does not exist", ruleID))
|
||||
return err
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, expectedVersion, dbrule.Version)
|
||||
}
|
||||
|
||||
t.Run("should increase version for all rules", func(t *testing.T) {
|
||||
_, err := store.IncreaseVersionForAllRulesInNamespaces(context.Background(), orgID, alertRuleNamespaceUIDs)
|
||||
require.NoError(t, err)
|
||||
|
||||
for _, rule := range alertRules {
|
||||
requireAlertRuleVersion(t, rule.ID, orgID, rule.Version+1)
|
||||
}
|
||||
|
||||
// this rule's version should not be changed
|
||||
requireAlertRuleVersion(t, alertRuleInAnotherNamespace.ID, orgID, alertRuleInAnotherNamespace.Version)
|
||||
})
|
||||
}
|
||||
|
||||
// createAlertRule creates an alert rule in the database and returns it.
|
||||
// If a generator is not specified, uniqueness of primary key is not guaranteed.
|
||||
func createRule(t *testing.T, store *DBstore, generator *models.AlertRuleGenerator) *models.AlertRule {
|
||||
|
||||
@@ -315,19 +315,24 @@ func (f *RuleStore) UpdateRuleGroup(ctx context.Context, orgID int64, namespaceU
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *RuleStore) IncreaseVersionForAllRulesInNamespace(_ context.Context, orgID int64, namespaceUID string) ([]models.AlertRuleKeyWithVersion, error) {
|
||||
func (f *RuleStore) IncreaseVersionForAllRulesInNamespaces(_ context.Context, orgID int64, namespaceUIDs []string) ([]models.AlertRuleKeyWithVersion, error) {
|
||||
f.mtx.Lock()
|
||||
defer f.mtx.Unlock()
|
||||
|
||||
f.RecordedOps = append(f.RecordedOps, GenericRecordedQuery{
|
||||
Name: "IncreaseVersionForAllRulesInNamespace",
|
||||
Params: []any{orgID, namespaceUID},
|
||||
Name: "IncreaseVersionForAllRulesInNamespaces",
|
||||
Params: []any{orgID, namespaceUIDs},
|
||||
})
|
||||
|
||||
var result []models.AlertRuleKeyWithVersion
|
||||
|
||||
namespaceUIDsMap := make(map[string]struct{}, len(namespaceUIDs))
|
||||
for _, namespaceUID := range namespaceUIDs {
|
||||
namespaceUIDsMap[namespaceUID] = struct{}{}
|
||||
}
|
||||
|
||||
for _, rule := range f.Rules[orgID] {
|
||||
if rule.NamespaceUID == namespaceUID && rule.OrgID == orgID {
|
||||
if _, ok := namespaceUIDsMap[rule.NamespaceUID]; ok && rule.OrgID == orgID {
|
||||
rule.Version++
|
||||
rule.Updated = time.Now()
|
||||
result = append(result, models.AlertRuleKeyWithVersion{
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
|
||||
"github.com/grafana/grafana/pkg/bus"
|
||||
"github.com/grafana/grafana/pkg/infra/db"
|
||||
"github.com/grafana/grafana/pkg/infra/tracing"
|
||||
"github.com/grafana/grafana/pkg/services/accesscontrol"
|
||||
acmock "github.com/grafana/grafana/pkg/services/accesscontrol/mock"
|
||||
"github.com/grafana/grafana/pkg/services/dashboards"
|
||||
@@ -26,7 +27,7 @@ import (
|
||||
|
||||
func SetupFolderService(tb testing.TB, cfg *setting.Cfg, db db.DB, dashboardStore dashboards.Store, folderStore *folderimpl.DashboardFolderStoreImpl, bus *bus.InProcBus, features featuremgmt.FeatureToggles, ac accesscontrol.AccessControl) folder.Service {
|
||||
tb.Helper()
|
||||
return folderimpl.ProvideService(ac, bus, dashboardStore, folderStore, db, features, supportbundlestest.NewFakeBundleService(), nil)
|
||||
return folderimpl.ProvideService(ac, bus, dashboardStore, folderStore, db, features, supportbundlestest.NewFakeBundleService(), nil, tracing.InitializeTracerForTest())
|
||||
}
|
||||
|
||||
func SetupDashboardService(tb testing.TB, sqlStore db.ReplDB, fs *folderimpl.DashboardFolderStoreImpl, cfg *setting.Cfg) (*dashboardservice.DashboardServiceImpl, dashboards.Store) {
|
||||
|
||||
Reference in New Issue
Block a user