Alerting: Optimize clean up rule versions (#102561)
* improve removal of old versions Signed-off-by: Yuri Tseretyan <yuriy.tseretyan@grafana.com> * Apply suggestions from code review Co-authored-by: Alexander Akhmetov <me@alx.cx> --------- Signed-off-by: Yuri Tseretyan <yuriy.tseretyan@grafana.com> Co-authored-by: Alexander Akhmetov <me@alx.cx>
This commit is contained in:
Yuri Tseretyan
co-authored by
Alexander Akhmetov
Alexander Akhmetov
parent
1f707d16ed
commit
14f4620835
@@ -453,14 +453,7 @@ func (st DBstore) UpdateAlertRules(ctx context.Context, user *ngmodels.UserUID,
|
||||
if _, err := sess.Insert(&ruleVersions); err != nil {
|
||||
return fmt.Errorf("failed to create new rule versions: %w", err)
|
||||
}
|
||||
|
||||
for _, rule := range ruleVersions {
|
||||
// delete old versions of alert rule
|
||||
_, err = st.deleteOldAlertRuleVersions(ctx, rule.RuleUID, rule.RuleOrgID, st.Cfg.RuleVersionRecordLimit)
|
||||
if err != nil {
|
||||
st.Logger.Warn("Failed to delete old alert rule versions", "org", rule.RuleOrgID, "rule", rule.RuleUID, "error", err)
|
||||
}
|
||||
}
|
||||
st.deleteOldAlertRuleVersions(ctx, sess, ruleVersions)
|
||||
}
|
||||
if len(keys) > 0 {
|
||||
_ = st.Bus.Publish(ctx, &RuleChangeEvent{
|
||||
@@ -471,50 +464,31 @@ func (st DBstore) UpdateAlertRules(ctx context.Context, user *ngmodels.UserUID,
|
||||
})
|
||||
}
|
||||
|
||||
func (st DBstore) deleteOldAlertRuleVersions(ctx context.Context, ruleUID string, orgID int64, limit int) (int64, error) {
|
||||
if limit < 0 {
|
||||
return 0, fmt.Errorf("failed to delete old alert rule versions: limit is set to '%d' but needs to be > 0", limit)
|
||||
func (st DBstore) deleteOldAlertRuleVersions(ctx context.Context, sess *db.Session, versions []alertRuleVersion) {
|
||||
if st.Cfg.RuleVersionRecordLimit < 1 {
|
||||
return
|
||||
}
|
||||
|
||||
if limit < 1 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
var affectedRows int64
|
||||
err := st.SQLStore.WithDbSession(ctx, func(sess *db.Session) error {
|
||||
highest := &alertRuleVersion{}
|
||||
ok, err := sess.Table("alert_rule_version").Desc("id").Where("rule_org_id = ?", orgID).Where("rule_uid = ?", ruleUID).Limit(1, limit).Get(highest)
|
||||
if err != nil {
|
||||
return err
|
||||
logger := st.Logger.FromContext(ctx)
|
||||
for _, rv := range versions {
|
||||
deleteTo := rv.Version - int64(st.Cfg.RuleVersionRecordLimit)
|
||||
// if the last version is less that retention, do nothing
|
||||
if deleteTo <= 1 {
|
||||
continue
|
||||
}
|
||||
if !ok {
|
||||
// No alert rule versions past the limit exist. Nothing to clean up.
|
||||
affectedRows = 0
|
||||
return nil
|
||||
}
|
||||
|
||||
res, err := sess.Exec(`
|
||||
DELETE FROM
|
||||
alert_rule_version
|
||||
WHERE
|
||||
rule_org_id = ? AND rule_uid = ?
|
||||
AND
|
||||
id <= ?
|
||||
`, orgID, ruleUID, highest.ID)
|
||||
logger := logger.New("org_id", rv.RuleOrgID, "rule_uid", rv.RuleUID, "version", rv.Version, "limit", st.Cfg.RulesPerRuleGroupLimit)
|
||||
res, err := sess.Exec(`DELETE FROM alert_rule_version WHERE rule_guid = ? AND version <= ?`, rv.RuleGUID, deleteTo)
|
||||
if err != nil {
|
||||
return err
|
||||
logger.Error("Failed to delete old alert rule versions", "error", err)
|
||||
return
|
||||
}
|
||||
rows, err := res.RowsAffected()
|
||||
if err != nil {
|
||||
return err
|
||||
rows = -1
|
||||
}
|
||||
affectedRows = rows
|
||||
if affectedRows > 0 {
|
||||
st.Logger.Info("Deleted old alert_rule_version(s)", "org", orgID, "limit", limit, "delete_count", affectedRows)
|
||||
if rows != 0 {
|
||||
logger.Info("Deleted old alert_rule_version(s)", "deleted", rows)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return affectedRows, err
|
||||
}
|
||||
}
|
||||
|
||||
// preventIntermediateUniqueConstraintViolations prevents unique constraint violations caused by an intermediate update.
|
||||
|
||||
@@ -1625,23 +1625,20 @@ func TestIntegration_AlertRuleVersionsCleanup(t *testing.T) {
|
||||
t.Skip("skipping integration test")
|
||||
}
|
||||
usr := models.UserUID("test")
|
||||
cfg := setting.NewCfg()
|
||||
cfg.UnifiedAlerting = setting.UnifiedAlertingSettings{
|
||||
cfg := setting.UnifiedAlertingSettings{
|
||||
BaseInterval: time.Duration(rand.Int63n(100)+1) * time.Second,
|
||||
}
|
||||
sqlStore := db.InitTestDB(t)
|
||||
folderService := setupFolderService(t, sqlStore, cfg, featuremgmt.WithFeatures())
|
||||
folderService := setupFolderService(t, sqlStore, setting.NewCfg(), featuremgmt.WithFeatures())
|
||||
b := &fakeBus{}
|
||||
store := createTestStore(sqlStore, folderService, &logtest.Fake{}, cfg.UnifiedAlerting, b)
|
||||
|
||||
generator := models.RuleGen
|
||||
generator = generator.With(generator.WithIntervalMatching(store.Cfg.BaseInterval), generator.WithUniqueOrgID())
|
||||
generator = generator.With(generator.WithIntervalMatching(cfg.BaseInterval), generator.WithUniqueOrgID())
|
||||
|
||||
t.Run("when calling the cleanup with fewer records than the limit all records should stay", func(t *testing.T) {
|
||||
alertingCfgSnapshot := cfg.UnifiedAlerting
|
||||
defer func() {
|
||||
cfg.UnifiedAlerting = alertingCfgSnapshot
|
||||
}()
|
||||
cfg.UnifiedAlerting = setting.UnifiedAlertingSettings{BaseInterval: alertingCfgSnapshot.BaseInterval, RuleVersionRecordLimit: 10}
|
||||
cfg := setting.UnifiedAlertingSettings{BaseInterval: cfg.BaseInterval, RuleVersionRecordLimit: 10}
|
||||
store := createTestStore(sqlStore, folderService, &logtest.Fake{}, cfg, b)
|
||||
|
||||
rule := createRule(t, store, generator)
|
||||
firstNewRule := models.CopyRule(rule)
|
||||
firstNewRule.Title = util.GenerateShortUID()
|
||||
@@ -1685,43 +1682,27 @@ func TestIntegration_AlertRuleVersionsCleanup(t *testing.T) {
|
||||
})
|
||||
|
||||
t.Run("only oldest records surpassing the limit should be deleted", func(t *testing.T) {
|
||||
alertingCfgSnapshot := cfg.UnifiedAlerting
|
||||
defer func() {
|
||||
cfg.UnifiedAlerting = alertingCfgSnapshot
|
||||
}()
|
||||
cfg.UnifiedAlerting = setting.UnifiedAlertingSettings{BaseInterval: alertingCfgSnapshot.BaseInterval, RuleVersionRecordLimit: 1}
|
||||
cfg := setting.UnifiedAlertingSettings{BaseInterval: cfg.BaseInterval, RuleVersionRecordLimit: 2}
|
||||
store := createTestStore(sqlStore, folderService, &logtest.Fake{}, cfg, b)
|
||||
rule := createRule(t, store, generator)
|
||||
oldRule := models.CopyRule(rule)
|
||||
oldRule.Title = "old-record"
|
||||
err := store.UpdateAlertRules(context.Background(), &usr, []models.UpdateRule{{
|
||||
Existing: rule,
|
||||
New: *oldRule,
|
||||
}}) // first entry in `rule_version_history` table happens here
|
||||
require.NoError(t, err)
|
||||
|
||||
rule.Version = rule.Version + 1
|
||||
middleRule := models.CopyRule(rule)
|
||||
middleRule.Title = "middle-record"
|
||||
err = store.UpdateAlertRules(context.Background(), &usr, []models.UpdateRule{{
|
||||
Existing: rule,
|
||||
New: *middleRule,
|
||||
}}) // second entry in `rule_version_history` table happens here
|
||||
require.NoError(t, err)
|
||||
for i := 0; i < 4; i++ {
|
||||
r, err := store.GetAlertRuleByUID(context.Background(), &models.GetAlertRuleByUIDQuery{UID: rule.UID})
|
||||
require.NoError(t, err)
|
||||
rn := models.CopyRule(r)
|
||||
rn.Title = util.GenerateShortUID()
|
||||
err = store.UpdateAlertRules(context.Background(), &models.AlertingUserUID, []models.UpdateRule{
|
||||
{
|
||||
Existing: r,
|
||||
New: *rn,
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
rule.Version = rule.Version + 1
|
||||
newerRule := models.CopyRule(rule)
|
||||
newerRule.Title = "newer-record"
|
||||
err = store.UpdateAlertRules(context.Background(), &usr, []models.UpdateRule{{
|
||||
Existing: rule,
|
||||
New: *newerRule,
|
||||
}}) // second entry in `rule_version_history` table happens here
|
||||
rule, err := store.GetAlertRuleByUID(context.Background(), &models.GetAlertRuleByUIDQuery{UID: rule.UID})
|
||||
require.NoError(t, err)
|
||||
|
||||
// only the `old-record` should be deleted since limit is set to 1 and there are total 2 records
|
||||
rowsAffected, err := store.deleteOldAlertRuleVersions(context.Background(), rule.UID, rule.OrgID, 1)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, int64(2), rowsAffected)
|
||||
|
||||
err = sqlStore.WithDbSession(context.Background(), func(sess *db.Session) error {
|
||||
var alertRuleVersions []*alertRuleVersion
|
||||
err := sess.Table(alertRuleVersion{}).Desc("id").Where("rule_org_id = ? and rule_uid = ?", rule.OrgID, rule.UID).Find(&alertRuleVersions)
|
||||
@@ -1729,22 +1710,13 @@ func TestIntegration_AlertRuleVersionsCleanup(t *testing.T) {
|
||||
return err
|
||||
}
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, alertRuleVersions, 1)
|
||||
assert.Equal(t, "newer-record", alertRuleVersions[0].Title)
|
||||
assert.Len(t, alertRuleVersions, 2)
|
||||
assert.Equal(t, rule.Title, alertRuleVersions[0].Title)
|
||||
assert.Equal(t, rule.Version, alertRuleVersions[0].Version)
|
||||
return err
|
||||
})
|
||||
require.NoError(t, err)
|
||||
})
|
||||
|
||||
t.Run("limit set to 0 should not fail", func(t *testing.T) {
|
||||
count, err := store.deleteOldAlertRuleVersions(context.Background(), "", 1, 0)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, int64(0), count)
|
||||
})
|
||||
t.Run("limit set to negative should fail", func(t *testing.T) {
|
||||
_, err := store.deleteOldAlertRuleVersions(context.Background(), "", 1, -1)
|
||||
require.Error(t, err)
|
||||
})
|
||||
}
|
||||
|
||||
func TestIntegration_ListAlertRules(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user