diff --git a/database_admin/migrations/150_delete_functions.down.sql b/database_admin/migrations/150_delete_functions.down.sql new file mode 100644 index 000000000..dbaa85734 --- /dev/null +++ b/database_admin/migrations/150_delete_functions.down.sql @@ -0,0 +1,150 @@ +DROP FUNCTION IF EXISTS delete_system(inventory_id_in uuid); +CREATE OR REPLACE FUNCTION delete_system(inventory_id_in uuid) + RETURNS TABLE + ( + deleted_inventory_id uuid + ) +AS +$delete_system$ +DECLARE + v_system_id INT; + v_account_id INT; +BEGIN + -- opt out to refresh cache and then delete + SELECT id, rh_account_id + FROM system_inventory + WHERE inventory_id = inventory_id_in + LIMIT 1 + FOR UPDATE OF system_inventory + INTO v_system_id, v_account_id; + + IF v_system_id IS NULL OR v_account_id IS NULL THEN + RAISE NOTICE 'Not found'; + RETURN; + END IF; + + UPDATE system_inventory + SET stale = true + WHERE rh_account_id = v_account_id + AND id = v_system_id; + + DELETE + FROM system_advisories + WHERE rh_account_id = v_account_id + AND system_id = v_system_id; + + DELETE + FROM system_repo + WHERE rh_account_id = v_account_id + AND system_id = v_system_id; + + DELETE + FROM system_package2 + WHERE rh_account_id = v_account_id + AND system_id = v_system_id; + + DELETE + FROM system_patch + WHERE rh_account_id = v_account_id + AND system_id = v_system_id; + + RETURN QUERY DELETE FROM system_inventory + WHERE rh_account_id = v_account_id AND + id = v_system_id + RETURNING inventory_id; +END; +$delete_system$ LANGUAGE 'plpgsql'; + +CREATE OR REPLACE FUNCTION delete_systems(inventory_ids UUID[]) + RETURNS INTEGER +AS +$$ +DECLARE + tmp_cnt INTEGER; +BEGIN + + WITH systems as ( + SELECT rh_account_id, id + FROM system_inventory + WHERE inventory_id = ANY (inventory_ids) + ORDER BY rh_account_id, id FOR UPDATE OF system_inventory), + marked as ( + UPDATE system_inventory sp + SET stale = true + WHERE (rh_account_id, id) in (select rh_account_id, id from systems) + ), + advisories as ( + DELETE + FROM system_advisories + WHERE (rh_account_id, system_id) in (select rh_account_id, id from systems) + ), + repos as ( + DELETE + FROM system_repo + WHERE (rh_account_id, system_id) in (select rh_account_id, id from systems) + ), + packages2 as ( + DELETE + FROM system_package2 + WHERE (rh_account_id, system_id) in (select rh_account_id, id from systems) + ), + patch_systems as ( + DELETE + FROM system_patch + WHERE (rh_account_id, system_id) in (select rh_account_id, id from systems) + ), + deleted as ( + DELETE + FROM system_inventory + WHERE (rh_account_id, id) in (select rh_account_id, id from systems) + RETURNING id + ) + SELECT count(*) + FROM deleted + INTO tmp_cnt; + + RETURN tmp_cnt; +END +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION delete_culled_systems(delete_limit INTEGER) + RETURNS INTEGER +AS +$fun$ +DECLARE + ids UUID[]; +BEGIN + ids := ARRAY( + SELECT inventory_id + FROM system_inventory + WHERE culled_timestamp < now() + ORDER BY id + LIMIT delete_limit + ); + return delete_systems(ids); +END; +$fun$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION mark_stale_systems(mark_limit integer) + RETURNS INTEGER +AS +$fun$ +DECLARE + marked integer; +BEGIN + WITH ids AS ( + SELECT rh_account_id, id, stale_warning_timestamp < now() as expired + FROM system_inventory + WHERE stale != (stale_warning_timestamp < now()) + ORDER BY rh_account_id, id FOR UPDATE OF system_inventory + LIMIT mark_limit + ) + UPDATE system_inventory si + SET stale = ids.expired + FROM ids + WHERE si.rh_account_id = ids.rh_account_id + AND si.id = ids.id; + GET DIAGNOSTICS marked = ROW_COUNT; + RETURN marked; +END; +$fun$ LANGUAGE plpgsql; diff --git a/database_admin/migrations/150_delete_functions.up.sql b/database_admin/migrations/150_delete_functions.up.sql new file mode 100644 index 000000000..b70492bef --- /dev/null +++ b/database_admin/migrations/150_delete_functions.up.sql @@ -0,0 +1,60 @@ +DROP FUNCTION IF EXISTS delete_system(inventory_id_in uuid); +CREATE OR REPLACE FUNCTION delete_system(inventory_id_in uuid) + RETURNS uuid +AS +$delete_system$ +DECLARE + v_system_id INT; + v_account_id INT; + v_inventory_id uuid; +BEGIN + -- opt out to refresh cache and then delete + SELECT id, rh_account_id + FROM system_inventory + WHERE inventory_id = inventory_id_in + LIMIT 1 + FOR UPDATE OF system_inventory + INTO v_system_id, v_account_id; + + IF v_system_id IS NULL OR v_account_id IS NULL THEN + RAISE NOTICE 'Not found'; + RETURN NULL; + END IF; + + UPDATE system_inventory + SET stale = true + WHERE rh_account_id = v_account_id + AND id = v_system_id; + + DELETE + FROM system_advisories + WHERE rh_account_id = v_account_id + AND system_id = v_system_id; + + DELETE + FROM system_repo + WHERE rh_account_id = v_account_id + AND system_id = v_system_id; + + DELETE + FROM system_package2 + WHERE rh_account_id = v_account_id + AND system_id = v_system_id; + + DELETE + FROM system_patch + WHERE rh_account_id = v_account_id + AND system_id = v_system_id; + + DELETE FROM system_inventory + WHERE rh_account_id = v_account_id AND + id = v_system_id + RETURNING inventory_id INTO v_inventory_id; + + RETURN v_inventory_id; +END; +$delete_system$ LANGUAGE 'plpgsql'; + +DROP FUNCTION IF EXISTS delete_systems(UUID[]); +DROP FUNCTION IF EXISTS delete_culled_systems(INTEGER); +DROP FUNCTION IF EXISTS mark_stale_systems(INTEGER); diff --git a/database_admin/schema/create_schema.sql b/database_admin/schema/create_schema.sql index 98e07798f..b7c627b9c 100644 --- a/database_admin/schema/create_schema.sql +++ b/database_admin/schema/create_schema.sql @@ -7,7 +7,7 @@ CREATE TABLE IF NOT EXISTS schema_migrations INSERT INTO schema_migrations -VALUES (149, false); +VALUES (150, false); -- --------------------------------------------------------------------------- -- Functions @@ -301,15 +301,13 @@ $refresh_system_cached_counts$ CREATE OR REPLACE FUNCTION delete_system(inventory_id_in uuid) - RETURNS TABLE - ( - deleted_inventory_id uuid - ) + RETURNS uuid AS $delete_system$ DECLARE v_system_id INT; v_account_id INT; + v_inventory_id uuid; BEGIN -- opt out to refresh cache and then delete SELECT id, rh_account_id @@ -321,7 +319,7 @@ BEGIN IF v_system_id IS NULL OR v_account_id IS NULL THEN RAISE NOTICE 'Not found'; - RETURN; + RETURN NULL; END IF; UPDATE system_inventory @@ -349,106 +347,14 @@ BEGIN WHERE rh_account_id = v_account_id AND system_id = v_system_id; - RETURN QUERY DELETE FROM system_inventory + DELETE FROM system_inventory WHERE rh_account_id = v_account_id AND id = v_system_id - RETURNING inventory_id; -END; -$delete_system$ LANGUAGE 'plpgsql'; - -CREATE OR REPLACE FUNCTION delete_systems(inventory_ids UUID[]) - RETURNS INTEGER -AS -$$ -DECLARE - tmp_cnt INTEGER; -BEGIN - - WITH systems as ( - SELECT rh_account_id, id - FROM system_inventory - WHERE inventory_id = ANY (inventory_ids) - ORDER BY rh_account_id, id FOR UPDATE OF system_inventory), - marked as ( - UPDATE system_inventory sp - SET stale = true - WHERE (rh_account_id, id) in (select rh_account_id, id from systems) - ), - advisories as ( - DELETE - FROM system_advisories - WHERE (rh_account_id, system_id) in (select rh_account_id, id from systems) - ), - repos as ( - DELETE - FROM system_repo - WHERE (rh_account_id, system_id) in (select rh_account_id, id from systems) - ), - packages2 as ( - DELETE - FROM system_package2 - WHERE (rh_account_id, system_id) in (select rh_account_id, id from systems) - ), - patch_systems as ( - DELETE - FROM system_patch - WHERE (rh_account_id, system_id) in (select rh_account_id, id from systems) - ), - deleted as ( - DELETE - FROM system_inventory - WHERE (rh_account_id, id) in (select rh_account_id, id from systems) - RETURNING id - ) - SELECT count(*) - FROM deleted - INTO tmp_cnt; + RETURNING inventory_id INTO v_inventory_id; - RETURN tmp_cnt; -END -$$ LANGUAGE plpgsql; - -CREATE OR REPLACE FUNCTION delete_culled_systems(delete_limit INTEGER) - RETURNS INTEGER -AS -$fun$ -DECLARE - ids UUID[]; -BEGIN - ids := ARRAY( - SELECT inventory_id - FROM system_inventory - WHERE culled_timestamp < now() - ORDER BY id - LIMIT delete_limit - ); - return delete_systems(ids); + RETURN v_inventory_id; END; -$fun$ LANGUAGE plpgsql; - -CREATE OR REPLACE FUNCTION mark_stale_systems(mark_limit integer) - RETURNS INTEGER -AS -$fun$ -DECLARE - marked integer; -BEGIN - WITH ids AS ( - SELECT rh_account_id, id, stale_warning_timestamp < now() as expired - FROM system_inventory - WHERE stale != (stale_warning_timestamp < now()) - ORDER BY rh_account_id, id FOR UPDATE OF system_inventory - LIMIT mark_limit - ) - UPDATE system_inventory si - SET stale = ids.expired - FROM ids - WHERE si.rh_account_id = ids.rh_account_id - AND si.id = ids.id; - GET DIAGNOSTICS marked = ROW_COUNT; - RETURN marked; -END; -$fun$ LANGUAGE plpgsql; +$delete_system$ LANGUAGE 'plpgsql'; CREATE OR REPLACE FUNCTION hash_partition_id(id int, parts int) RETURNS int AS diff --git a/tasks/system_culling/system_culling.go b/tasks/system_culling/system_culling.go index d24e02024..4c227b57b 100644 --- a/tasks/system_culling/system_culling.go +++ b/tasks/system_culling/system_culling.go @@ -46,27 +46,76 @@ func runSystemCulling() { } } -// https://github.com/go-gorm/gorm/issues/3722 -func deleteCulledSystems(tx *gorm.DB, limitDeleted int) (nDeleted int, err error) { - var nDeletedArr []int - err = tx.Raw("select delete_culled_systems(?)", limitDeleted). - Find(&nDeletedArr).Error - if len(nDeletedArr) > 0 { - nDeleted = nDeletedArr[0] +// systems are deleted in independent transactions to avoid locking multiple rows for long time +func deleteCulledSystems(tx *gorm.DB, limitDeleted int) (nDeleted int64, err error) { + var inventoryIDs []string + err = tx.Model(&models.SystemInventory{}). + Where("culled_timestamp < ?", time.Now()). + Order("id"). + Limit(limitDeleted). + Pluck("inventory_id", &inventoryIDs).Error + if err != nil { + return 0, err + } + + for _, id := range inventoryIDs { + var rowsAffected int64 + delErr := tasks.CancelableDB().Transaction(func(tx2 *gorm.DB) error { + res := tx2.Exec("select delete_system(?::uuid)", id) + if res.Error != nil { + return res.Error + } + rowsAffected = res.RowsAffected + return nil + }) + if delErr != nil { + utils.LogWarn("inventoryID", id, "err", delErr.Error(), "Delete culled system") + continue + } + nDeleted += rowsAffected } - return nDeleted, err + return nDeleted, nil } -func markSystemsStale(tx *gorm.DB, markedLimit int) (nMarked int, err error) { - var nMarkedArr []int - err = tx.Raw("select mark_stale_systems(?)", markedLimit). - Find(&nMarkedArr).Error - if len(nMarkedArr) > 0 { - nMarked = nMarkedArr[0] +// each update runs in its own transaction to avoid holding locks across many rows +func markSystemsStale(tx *gorm.DB, markedLimit int) (nMarked int64, err error) { + var candidates []struct { + RhAccountID int `gorm:"column:rh_account_id"` + ID int64 `gorm:"column:id"` + Expired bool `gorm:"column:expired"` + } + now := time.Now() + err = tx.Model(&models.SystemInventory{}). + Select("rh_account_id, id, (stale_warning_timestamp < ?) as expired", now). + Where("stale != (stale_warning_timestamp < ?)", now). + Order("rh_account_id").Order("id"). + Limit(markedLimit). + Find(&candidates).Error + if err != nil { + return 0, err + } + + for _, c := range candidates { + var rowsAffected int64 + markErr := tasks.CancelableDB().Transaction(func(tx2 *gorm.DB) error { + res := tx2.Model(&models.SystemInventory{}). + Where("rh_account_id = ? AND id = ?", c.RhAccountID, c.ID). + Update("stale", c.Expired) + if res.Error != nil { + return res.Error + } + rowsAffected = res.RowsAffected + return nil + }) + if markErr != nil { + utils.LogWarn("rhAccountID", c.RhAccountID, "systemID", c.ID, "err", markErr.Error(), "Mark stale system") + continue + } + nMarked += rowsAffected } - return nMarked, err + return nMarked, nil } func pruneDeletedSystems(tx *gorm.DB, limitDeleted int) (int64, error) { diff --git a/tasks/system_culling/system_culling_test.go b/tasks/system_culling/system_culling_test.go index 1d1e8fc6a..3a7071f52 100644 --- a/tasks/system_culling/system_culling_test.go +++ b/tasks/system_culling/system_culling_test.go @@ -68,11 +68,11 @@ func TestSingleSystemStale(t *testing.T) { nMarked, err := markSystemsStale(database.DB, 0) assert.Nil(t, err) - assert.Equal(t, 0, nMarked) + assert.Equal(t, int64(0), nMarked) nMarked, err = markSystemsStale(database.DB, 1) assert.Nil(t, err) - assert.Equal(t, 1, nMarked) + assert.Equal(t, int64(1), nMarked) oldAffected = accountData[0].SystemsInstallable assert.NoError(t, database.DB.Find(&accountData, "rh_account_id = ? AND advisory_id = ?", @@ -113,7 +113,7 @@ func TestMarkSystemsStale(t *testing.T) { } nMarked, err := markSystemsStale(database.DB, 500) assert.Nil(t, err) - assert.Equal(t, 18, nMarked) + assert.Equal(t, int64(18), nMarked) inventories = loadAllSystemInventories(t, database.DB) for i := range inventories { @@ -147,7 +147,7 @@ func TestMarkSystemsNotStale(t *testing.T) { } nMarked, err := markSystemsStale(database.DB, 500) assert.NoError(t, err) - assert.Equal(t, 18, nMarked) + assert.Equal(t, int64(18), nMarked) inventories = loadAllSystemInventories(t, database.DB) for i := range inventories { @@ -192,12 +192,12 @@ func TestCullSystems(t *testing.T) { // first batch nDeleted, err := deleteCulledSystems(database.DB, 3) assert.Nil(t, err) - assert.Equal(t, 3, nDeleted) + assert.Equal(t, int64(3), nDeleted) // second batch nDeleted, err = deleteCulledSystems(database.DB, 3) assert.Nil(t, err) - assert.Equal(t, 1, nDeleted) + assert.Equal(t, int64(1), nDeleted) assert.NoError(t, database.DB.Model(&models.SystemInventory{}).Count(&cntAfter).Error) assert.Equal(t, cnt-int64(nToDelete), cntAfter) diff --git a/turnpike/controllers/admin.go b/turnpike/controllers/admin.go index 74ddc8c61..c393a5f00 100644 --- a/turnpike/controllers/admin.go +++ b/turnpike/controllers/admin.go @@ -244,7 +244,7 @@ func SystemDeleteHandler(c *gin.Context) { return } - query := tx.Exec("select deleted_inventory_id from delete_system(?::uuid)", systemInventoryID[0]) + query := tx.Exec("select delete_system(?::uuid)", systemInventoryID[0]) if err := query.Error; err != nil { utils.LogAndRespError(c, err, "Could not delete system") return