From 91bd1cdb41daaee593acba748274c78534a1a454 Mon Sep 17 00:00:00 2001 From: Alexander Weaver Date: Fri, 16 Dec 2022 09:07:44 -0600 Subject: [PATCH] Revert "Alerting: Store alertmanager configuration history in a separate table in the database" (#60470) Revert "Alerting: Store alertmanager configuration history in a separate table in the database (#60197)" This reverts commit ec80f38c34b6bde14da91db90554aa006db83972. --- pkg/services/ngalert/store/alertmanager.go | 126 ++++++++++++++---- .../sqlstore/migrations/ualert/tables.go | 3 - .../sqlstore/migrations/ualert/ualert.go | 41 ------ 3 files changed, 100 insertions(+), 70 deletions(-) diff --git a/pkg/services/ngalert/store/alertmanager.go b/pkg/services/ngalert/store/alertmanager.go index 20db7d31dfe..da3d90cc706 100644 --- a/pkg/services/ngalert/store/alertmanager.go +++ b/pkg/services/ngalert/store/alertmanager.go @@ -6,6 +6,9 @@ import ( "fmt" "time" + "xorm.io/builder" + "xorm.io/core" + "github.com/grafana/grafana/pkg/infra/db" "github.com/grafana/grafana/pkg/services/ngalert/models" ) @@ -28,7 +31,7 @@ func (st *DBstore) GetLatestAlertmanagerConfiguration(ctx context.Context, query return st.SQLStore.WithDbSession(ctx, func(sess *db.Session) error { c := &models.AlertConfiguration{} // The ID is already an auto incremental column, using the ID as an order should guarantee the latest. - ok, err := sess.Table("alert_configuration").Where("org_id = ?", query.OrgID).Get(c) + ok, err := sess.Desc("id").Where("org_id = ?", query.OrgID).Limit(1).Get(c) if err != nil { return err } @@ -46,7 +49,8 @@ func (st *DBstore) GetLatestAlertmanagerConfiguration(ctx context.Context, query func (st *DBstore) GetAllLatestAlertmanagerConfiguration(ctx context.Context) ([]*models.AlertConfiguration, error) { var result []*models.AlertConfiguration err := st.SQLStore.WithDbSession(ctx, func(sess *db.Session) error { - if err := sess.Table("alert_configuration").Find(&result); err != nil { + condition := builder.In("id", builder.Select("MAX(id)").From("alert_configuration").GroupBy("org_id")) + if err := sess.Table("alert_configuration").Where(condition).Find(&result); err != nil { return err } return nil @@ -74,23 +78,10 @@ func (st DBstore) SaveAlertmanagerConfigurationWithCallback(ctx context.Context, ConfigurationVersion: cmd.ConfigurationVersion, Default: cmd.Default, OrgID: cmd.OrgID, - CreatedAt: time.Now().Unix(), } - // TODO: If we are more structured around how we seed configurations in the future, this can be a pure update instead of upsert. This should improve perf and code clarity. - upsertSQL := st.SQLStore.GetDialect().UpsertSQL( - "alert_configuration", - []string{"org_id"}, - []string{"alertmanager_configuration", "configuration_version", "created_at", "default", "org_id", "configuration_hash"}, - ) - params := append(make([]interface{}, 0), cmd.AlertmanagerConfiguration, cmd.ConfigurationVersion, config.CreatedAt, config.Default, config.OrgID, config.ConfigurationHash) - if _, err := sess.SQL(upsertSQL, params...).Query(); err != nil { + if _, err := sess.Insert(config); err != nil { return err } - - if _, err := sess.Table("alert_configuration_history").Insert(config); err != nil { - return err - } - if _, err := st.deleteOldConfigurations(ctx, cmd.OrgID, ConfigRecordsLimit); err != nil { st.Logger.Warn("failed to delete old am configs", "org", cmd.OrgID, "error", err) } @@ -102,7 +93,6 @@ func (st DBstore) SaveAlertmanagerConfigurationWithCallback(ctx context.Context, }) } -// UpdateAlertmanagerConfiguration replaces an alertmanager configuration with optimistic locking. It assumes that an existing revision of the configuration exists in the store, and will return an error otherwise. func (st *DBstore) UpdateAlertmanagerConfiguration(ctx context.Context, cmd *models.SaveAlertmanagerConfigurationCmd) error { return st.SQLStore.WithTransactionalDbSession(ctx, func(sess *db.Session) error { config := models.AlertConfiguration{ @@ -113,25 +103,109 @@ func (st *DBstore) UpdateAlertmanagerConfiguration(ctx context.Context, cmd *mod OrgID: cmd.OrgID, CreatedAt: time.Now().Unix(), } - rows, err := sess.Table("alert_configuration"). - Where("org_id = ? AND configuration_hash = ?", config.OrgID, cmd.FetchedConfigurationHash). - Update(config) + res, err := sess.Exec(fmt.Sprintf(getInsertQuery(st.SQLStore.GetDialect().DriverName()), st.SQLStore.GetDialect().Quote("default")), + config.AlertmanagerConfiguration, + config.ConfigurationHash, + config.ConfigurationVersion, + config.OrgID, + config.CreatedAt, + st.SQLStore.GetDialect().BooleanStr(config.Default), + cmd.OrgID, + cmd.OrgID, + cmd.FetchedConfigurationHash, + ) + if err != nil { + return err + } + rows, err := res.RowsAffected() if err != nil { return err } if rows == 0 { return ErrVersionLockedObjectNotFound } - if _, err := sess.Table("alert_configuration_history").Insert(config); err != nil { - return err - } if _, err := st.deleteOldConfigurations(ctx, cmd.OrgID, ConfigRecordsLimit); err != nil { st.Logger.Warn("failed to delete old am configs", "org", cmd.OrgID, "error", err) } - return nil + return err }) } +// getInsertQuery is used to determinate the insert query for the alertmanager config +// based on the provided sql driver. This is necesarry as such an advanced query +// is not supported by our ORM and we need to generate it manually for each SQL dialect. +// We introduced this as part of a bug fix as the old approach wasn't working. +// Rel: https://github.com/grafana/grafana/issues/51356 +func getInsertQuery(driver string) string { + switch driver { + case core.MYSQL: + return ` + INSERT INTO alert_configuration + (alertmanager_configuration, configuration_hash, configuration_version, org_id, created_at, %s) + SELECT T.* FROM (SELECT ? AS alertmanager_configuration,? AS configuration_hash,? AS configuration_version,? AS org_id,? AS created_at,? AS 'default') AS T + WHERE + EXISTS ( + SELECT 1 + FROM alert_configuration + WHERE + org_id = ? + AND + id = (SELECT MAX(id) FROM alert_configuration WHERE org_id = ?) + AND + configuration_hash = ? + )` + case core.POSTGRES: + return ` + INSERT INTO alert_configuration + (alertmanager_configuration, configuration_hash, configuration_version, org_id, created_at, %s) + SELECT T.* FROM (VALUES($1,$2,$3,$4::bigint,$5::integer,$6::boolean)) AS T + WHERE + EXISTS ( + SELECT 1 + FROM alert_configuration + WHERE + org_id = $7 + AND + id = (SELECT MAX(id) FROM alert_configuration WHERE org_id = $8::bigint) + AND + configuration_hash = $9 + )` + case core.SQLITE: + return ` + INSERT INTO alert_configuration + (alertmanager_configuration, configuration_hash, configuration_version, org_id, created_at, %s) + SELECT T.* FROM (VALUES(?,?,?,?,?,?)) AS T + WHERE + EXISTS ( + SELECT 1 + FROM alert_configuration + WHERE + org_id = ? + AND + id = (SELECT MAX(id) FROM alert_configuration WHERE org_id = ?) + AND + configuration_hash = ? + )` + default: + // SQLite version + return ` + INSERT INTO alert_configuration + (alertmanager_configuration, configuration_hash, configuration_version, org_id, created_at, %s) + SELECT T.* FROM (VALUES(?,?,?,?,?,?)) AS T + WHERE + EXISTS ( + SELECT 1 + FROM alert_configuration + WHERE + org_id = ? + AND + id = (SELECT MAX(id) FROM alert_configuration WHERE org_id = ?) + AND + configuration_hash = ? + )` + } +} + func (st *DBstore) deleteOldConfigurations(ctx context.Context, orgID int64, limit int) (int64, error) { if limit < 1 { return 0, fmt.Errorf("failed to delete old configurations: limit is set to '%d' but needs to be > 0", limit) @@ -144,7 +218,7 @@ func (st *DBstore) deleteOldConfigurations(ctx context.Context, orgID int64, lim var affectedRows int64 err := st.SQLStore.WithDbSession(ctx, func(sess *db.Session) error { highest := &models.AlertConfiguration{} - ok, err := sess.Table("alert_configuration_history").Desc("id").Where("org_id = ?", orgID).OrderBy("id").Limit(1, limit-1).Get(highest) + ok, err := sess.Desc("id").Where("org_id = ?", orgID).OrderBy("id").Limit(1, limit-1).Get(highest) if err != nil { return err } @@ -163,7 +237,7 @@ func (st *DBstore) deleteOldConfigurations(ctx context.Context, orgID int64, lim res, err := sess.Exec(` DELETE FROM - alert_configuration_history + alert_configuration WHERE org_id = ? AND diff --git a/pkg/services/sqlstore/migrations/ualert/tables.go b/pkg/services/sqlstore/migrations/ualert/tables.go index 4742484b192..875f9f68894 100644 --- a/pkg/services/sqlstore/migrations/ualert/tables.go +++ b/pkg/services/sqlstore/migrations/ualert/tables.go @@ -35,9 +35,6 @@ func AddTablesMigrations(mg *migrator.Migrator) { AddAlertImageMigrations(mg) AddAlertmanagerConfigHistoryMigrations(mg) - ExtractAlertmanagerConfigurationHistoryMigration(mg) - mg.AddMigration("drop non-unique orgID index", migrator.NewDropIndexMigration(migrator.Table{Name: "alert_configuration"}, &migrator.Index{Cols: []string{"org_id"}})) - mg.AddMigration("add unique index on orgID", migrator.NewAddIndexMigration(migrator.Table{Name: "alert_configuration"}, &migrator.Index{Type: migrator.UniqueIndex, Cols: []string{"org_id"}})) } // AddAlertDefinitionMigrations should not be modified. diff --git a/pkg/services/sqlstore/migrations/ualert/ualert.go b/pkg/services/sqlstore/migrations/ualert/ualert.go index ab7ac04eb25..8bf0f50f980 100644 --- a/pkg/services/sqlstore/migrations/ualert/ualert.go +++ b/pkg/services/sqlstore/migrations/ualert/ualert.go @@ -878,44 +878,3 @@ func (c updateRulesOrderInGroup) Exec(sess *xorm.Session, migrator *migrator.Mig } return nil } - -func ExtractAlertmanagerConfigurationHistoryMigration(mg *migrator.Migrator) { - if !mg.Cfg.UnifiedAlerting.IsEnabled() { - return - } - mg.AddMigration("extract alertmanager configuration history to separate table", &extractAlertmanagerConfigurationHistory{}) -} - -type extractAlertmanagerConfigurationHistory struct { - migrator.MigrationBase -} - -func (c extractAlertmanagerConfigurationHistory) SQL(migrator.Dialect) string { - return codeMigration -} - -func (c extractAlertmanagerConfigurationHistory) Exec(sess *xorm.Session, migrator *migrator.Migrator) error { - var orgs []int64 - if err := sess.Table("alert_configuration").Distinct("org_id").Find(&orgs); err != nil { - return fmt.Errorf("failed to retrieve the organizations with alerting configurations: %w", err) - } - - // Quote the column called "default" because it's a reserved keyword in SQL. - fields := fmt.Sprintf("org_id, alertmanager_configuration, configuration_hash, configuration_version, created_at, %s", migrator.Dialect.Quote("default")) - for _, orgID := range orgs { - _, err := sess.Exec(` - INSERT INTO alert_configuration_history (`+fields+`) - SELECT `+fields+` - FROM alert_configuration - WHERE org_id = ? AND id != (SELECT MAX(id) FROM alert_configuration WHERE org_id = ?)`, - orgID, orgID) - if err != nil { - return fmt.Errorf("failed to move old configurations to history table: %w", err) - } - _, err = sess.Exec("DELETE FROM alert_configuration WHERE org_id = ? AND id != (SELECT MAX(id) FROM alert_configuration WHERE org_id = ?)", orgID, orgID) - if err != nil { - return fmt.Errorf("failed to evict old configurations after moving to history table: %w", err) - } - } - return nil -}