diff --git a/base/database/testing.go b/base/database/testing.go index 681835997..df573e985 100644 --- a/base/database/testing.go +++ b/base/database/testing.go @@ -10,6 +10,7 @@ import ( "testing" "time" + "github.com/google/uuid" "github.com/stretchr/testify/assert" ) @@ -297,6 +298,48 @@ func DeleteSystemAdvisories(t *testing.T, systemID int64, advisoryIDs []int64) { assert.Nil(t, DB.Exec("SELECT * FROM update_system_caches(?)", systemID).Error) } +func CreateAccountAdvisory(t *testing.T, rhAccountID int, workspaceID string, advisoryIDs []int64, + systemsInstallable int) { + wsID := uuid.MustParse(workspaceID) + for _, advisoryID := range advisoryIDs { + err := DB.Create(&models.AccountAdvisory{ + AdvisoryID: advisoryID, RhAccountID: rhAccountID, WorkspaceID: wsID, + SystemsInstallable: systemsInstallable, SystemsApplicable: systemsInstallable}).Error + assert.Nil(t, err) + } + CheckAccountAdvisory(t, rhAccountID, workspaceID, advisoryIDs, systemsInstallable) +} + +func CheckAccountAdvisory(t *testing.T, rhAccountID int, workspaceID string, advisoryIDs []int64, + systemsInstallable int) { + var accountAdvisory []models.AccountAdvisory + err := DB.Where("rh_account_id = ? AND workspace_id = ? AND advisory_id IN (?)", + rhAccountID, workspaceID, advisoryIDs). + Find(&accountAdvisory).Error + assert.Nil(t, err) + + sum := 0 + for _, item := range accountAdvisory { + sum += item.SystemsInstallable + } + assert.Equal(t, systemsInstallable*len(advisoryIDs), sum, "sum of systems_installable does not match") +} + +func DeleteAccountAdvisory(t *testing.T, rhAccountID int, workspaceID string, advisoryIDs []int64) { + query := DB.Model(&models.AccountAdvisory{}).Where("rh_account_id = ? AND workspace_id = ? AND advisory_id IN (?)", + rhAccountID, workspaceID, advisoryIDs) + assert.Nil(t, query.Delete(&models.AccountAdvisory{}).Error) + + var cnt int64 + assert.Nil(t, query.Count(&cnt).Error) + assert.Equal(t, int64(0), cnt) +} + +func DeleteAccountAdvisoryByAccount(t *testing.T, rhAccountID int) { + assert.Nil(t, DB.Where("rh_account_id = ?", rhAccountID). + Delete(&models.AccountAdvisory{}).Error) +} + func DeleteAdvisoryAccountData(t *testing.T, rhAccountID int, advisoryIDs []int64) { query := DB.Model(&models.AdvisoryAccountData{}).Where("rh_account_id = ? AND advisory_id IN (?)", rhAccountID, advisoryIDs) diff --git a/base/models/models.go b/base/models/models.go index 65d65b864..98fb831e9 100644 --- a/base/models/models.go +++ b/base/models/models.go @@ -281,6 +281,22 @@ func (AdvisoryAccountData) TableName() string { } type AdvisoryAccountDataSlice []AdvisoryAccountData + +type AccountAdvisory struct { + AdvisoryID int64 `gorm:"primaryKey"` + RhAccountID int `gorm:"primaryKey"` + WorkspaceID uuid.UUID `gorm:"primaryKey"` + SystemsApplicable int + SystemsInstallable int + Notified *time.Time +} + +func (AccountAdvisory) TableName() string { + return "account_advisory" +} + +type AccountAdvisorySlice []AccountAdvisory + type Repo struct { ID int64 `gorm:"primaryKey"` Name string diff --git a/database_admin/migrations/154_account_advisory.down.sql b/database_admin/migrations/154_account_advisory.down.sql new file mode 100644 index 000000000..d2de54398 --- /dev/null +++ b/database_admin/migrations/154_account_advisory.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS account_advisory; diff --git a/database_admin/migrations/154_account_advisory.up.sql b/database_admin/migrations/154_account_advisory.up.sql new file mode 100644 index 000000000..464e22792 --- /dev/null +++ b/database_admin/migrations/154_account_advisory.up.sql @@ -0,0 +1,22 @@ +CREATE TABLE IF NOT EXISTS account_advisory +( + advisory_id BIGINT NOT NULL, + rh_account_id INT NOT NULL, + workspace_id UUID NOT NULL, + systems_applicable INT NOT NULL DEFAULT 0, + systems_installable INT NOT NULL DEFAULT 0, + notified TIMESTAMP WITH TIME ZONE NULL, + CONSTRAINT account_advisory_advisory_id + FOREIGN KEY (advisory_id) + REFERENCES advisory_metadata (id), + PRIMARY KEY (rh_account_id, workspace_id, advisory_id) +) PARTITION BY HASH (rh_account_id); + +SELECT create_table_partitions('account_advisory', 32, + $$WITH (fillfactor = '70', autovacuum_vacuum_scale_factor = '0.05') + TABLESPACE pg_default$$); + +SELECT grant_table_partitions('SELECT, INSERT, UPDATE, DELETE', 'account_advisory', 'manager'); +SELECT grant_table_partitions('SELECT, INSERT, UPDATE, DELETE', 'account_advisory', 'evaluator'); +SELECT grant_table_partitions('SELECT, INSERT, UPDATE, DELETE', 'account_advisory', 'listener'); +SELECT grant_table_partitions('SELECT, INSERT, UPDATE, DELETE', 'account_advisory', 'vmaas_sync'); diff --git a/database_admin/migrations/155_account_advisory_queries.down.sql b/database_admin/migrations/155_account_advisory_queries.down.sql new file mode 100644 index 000000000..7e17530df --- /dev/null +++ b/database_admin/migrations/155_account_advisory_queries.down.sql @@ -0,0 +1,10 @@ +DROP FUNCTION IF EXISTS backfill_account_advisory(INTEGER); + +DROP FUNCTION IF EXISTS refresh_account_advisory_caches(INTEGER, INTEGER); + +DROP FUNCTION IF EXISTS refresh_account_advisory_caches_multi(INTEGER[], INTEGER); + +DROP INDEX IF EXISTS account_advisory_systems_applicable_idx; +DROP INDEX IF EXISTS account_advisory_systems_installable_idx; + +TRUNCATE account_advisory; diff --git a/database_admin/migrations/155_account_advisory_queries.up.sql b/database_admin/migrations/155_account_advisory_queries.up.sql new file mode 100644 index 000000000..81d34bf76 --- /dev/null +++ b/database_admin/migrations/155_account_advisory_queries.up.sql @@ -0,0 +1,65 @@ +CREATE INDEX ON account_advisory (systems_applicable); +CREATE INDEX ON account_advisory (systems_installable); + +CREATE OR REPLACE FUNCTION refresh_account_advisory_caches_multi(advisory_ids_in INTEGER[] DEFAULT NULL, + rh_account_id_in INTEGER DEFAULT NULL) + RETURNS VOID AS +$refresh_account_advisory$ +BEGIN + PERFORM aa.rh_account_id, aa.workspace_id, aa.advisory_id + FROM account_advisory aa + WHERE (aa.advisory_id = ANY (advisory_ids_in) OR advisory_ids_in IS NULL) + AND (aa.rh_account_id = rh_account_id_in OR rh_account_id_in IS NULL) + FOR UPDATE OF aa; + + WITH current_counts AS ( + SELECT sa.advisory_id, sa.rh_account_id, si.workspace_id, + count(sa.*) FILTER (WHERE sa.status_id = 0) AS systems_installable, + count(sa.*) AS systems_applicable + FROM system_advisories sa + JOIN system_inventory si + ON sa.rh_account_id = si.rh_account_id AND sa.system_id = si.id + JOIN system_patch sp + ON si.id = sp.system_id AND sp.rh_account_id = si.rh_account_id + WHERE sp.last_evaluation IS NOT NULL + AND si.stale = FALSE + AND si.workspace_id IS NOT NULL + AND (sa.advisory_id = ANY (advisory_ids_in) OR advisory_ids_in IS NULL) + AND (si.rh_account_id = rh_account_id_in OR rh_account_id_in IS NULL) + GROUP BY sa.advisory_id, sa.rh_account_id, si.workspace_id + ), + upserted AS ( + INSERT INTO account_advisory (advisory_id, rh_account_id, workspace_id, systems_installable, systems_applicable) + SELECT advisory_id, rh_account_id, workspace_id, systems_installable, systems_applicable + FROM current_counts + ON CONFLICT (rh_account_id, workspace_id, advisory_id) DO UPDATE SET + systems_installable = EXCLUDED.systems_installable, + systems_applicable = EXCLUDED.systems_applicable + ) + DELETE FROM account_advisory + WHERE (advisory_id, rh_account_id, workspace_id) NOT IN (SELECT advisory_id, rh_account_id, workspace_id FROM current_counts) + AND (advisory_id = ANY (advisory_ids_in) OR advisory_ids_in IS NULL) + AND (rh_account_id = rh_account_id_in OR rh_account_id_in IS NULL); +END; +$refresh_account_advisory$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION refresh_account_advisory_caches(advisory_id_in INTEGER DEFAULT NULL, + rh_account_id_in INTEGER DEFAULT NULL) + RETURNS VOID AS +$refresh_account_advisory$ +BEGIN + IF advisory_id_in IS NOT NULL THEN + PERFORM refresh_account_advisory_caches_multi(ARRAY [advisory_id_in], rh_account_id_in); + ELSE + PERFORM refresh_account_advisory_caches_multi(NULL, rh_account_id_in); + END IF; +END; +$refresh_account_advisory$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION backfill_account_advisory(rh_account_id_in INTEGER) + RETURNS VOID AS +$backfill$ +BEGIN + PERFORM refresh_account_advisory_caches_multi(NULL, rh_account_id_in); +END; +$backfill$ LANGUAGE plpgsql; diff --git a/database_admin/schema/create_schema.sql b/database_admin/schema/create_schema.sql index 068cbbcc5..e38372765 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 (153, false); +VALUES (155, false); -- --------------------------------------------------------------------------- -- Functions @@ -171,6 +171,69 @@ BEGIN END; $refresh_advisory$ language plpgsql; +CREATE OR REPLACE FUNCTION refresh_account_advisory_caches_multi(advisory_ids_in INTEGER[] DEFAULT NULL, + rh_account_id_in INTEGER DEFAULT NULL) + RETURNS VOID AS +$refresh_account_advisory$ +BEGIN + PERFORM aa.rh_account_id, aa.workspace_id, aa.advisory_id + FROM account_advisory aa + WHERE (aa.advisory_id = ANY (advisory_ids_in) OR advisory_ids_in IS NULL) + AND (aa.rh_account_id = rh_account_id_in OR rh_account_id_in IS NULL) + FOR UPDATE OF aa; + + WITH current_counts AS ( + SELECT sa.advisory_id, sa.rh_account_id, si.workspace_id, + count(sa.*) FILTER (WHERE sa.status_id = 0) AS systems_installable, + count(sa.*) AS systems_applicable + FROM system_advisories sa + JOIN system_inventory si + ON sa.rh_account_id = si.rh_account_id AND sa.system_id = si.id + JOIN system_patch sp + ON si.id = sp.system_id AND sp.rh_account_id = si.rh_account_id + WHERE sp.last_evaluation IS NOT NULL + AND si.stale = FALSE + AND si.workspace_id IS NOT NULL + AND (sa.advisory_id = ANY (advisory_ids_in) OR advisory_ids_in IS NULL) + AND (si.rh_account_id = rh_account_id_in OR rh_account_id_in IS NULL) + GROUP BY sa.advisory_id, sa.rh_account_id, si.workspace_id + ), + upserted AS ( + INSERT INTO account_advisory (advisory_id, rh_account_id, workspace_id, systems_installable, systems_applicable) + SELECT advisory_id, rh_account_id, workspace_id, systems_installable, systems_applicable + FROM current_counts + ON CONFLICT (rh_account_id, workspace_id, advisory_id) DO UPDATE SET + systems_installable = EXCLUDED.systems_installable, + systems_applicable = EXCLUDED.systems_applicable + ) + DELETE FROM account_advisory + WHERE (advisory_id, rh_account_id, workspace_id) NOT IN (SELECT advisory_id, rh_account_id, workspace_id FROM current_counts) + AND (advisory_id = ANY (advisory_ids_in) OR advisory_ids_in IS NULL) + AND (rh_account_id = rh_account_id_in OR rh_account_id_in IS NULL); +END; +$refresh_account_advisory$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION refresh_account_advisory_caches(advisory_id_in INTEGER DEFAULT NULL, + rh_account_id_in INTEGER DEFAULT NULL) + RETURNS VOID AS +$refresh_account_advisory$ +BEGIN + IF advisory_id_in IS NOT NULL THEN + PERFORM refresh_account_advisory_caches_multi(ARRAY [advisory_id_in], rh_account_id_in); + ELSE + PERFORM refresh_account_advisory_caches_multi(NULL, rh_account_id_in); + END IF; +END; +$refresh_account_advisory$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION backfill_account_advisory(rh_account_id_in INTEGER) + RETURNS VOID AS +$backfill$ +BEGIN + PERFORM refresh_account_advisory_caches_multi(NULL, rh_account_id_in); +END; +$backfill$ LANGUAGE plpgsql; + CREATE OR REPLACE FUNCTION refresh_system_caches(system_id_in BIGINT DEFAULT NULL, rh_account_id_in INTEGER DEFAULT NULL) RETURNS INTEGER AS @@ -783,6 +846,33 @@ GRANT SELECT, INSERT, UPDATE, DELETE ON advisory_account_data TO vmaas_sync; CREATE INDEX ON advisory_account_data (systems_applicable); CREATE INDEX ON advisory_account_data (systems_installable); +-- account_advisory +CREATE TABLE IF NOT EXISTS account_advisory +( + advisory_id BIGINT NOT NULL, + rh_account_id INT NOT NULL, + workspace_id UUID NOT NULL, + systems_applicable INT NOT NULL DEFAULT 0, + systems_installable INT NOT NULL DEFAULT 0, + notified TIMESTAMP WITH TIME ZONE NULL, + CONSTRAINT account_advisory_advisory_id + FOREIGN KEY (advisory_id) + REFERENCES advisory_metadata (id), + PRIMARY KEY (rh_account_id, workspace_id, advisory_id) +) PARTITION BY HASH (rh_account_id); + +SELECT create_table_partitions('account_advisory', 32, + $$WITH (fillfactor = '70', autovacuum_vacuum_scale_factor = '0.05') + TABLESPACE pg_default$$); + +SELECT grant_table_partitions('SELECT, INSERT, UPDATE, DELETE', 'account_advisory', 'manager'); +SELECT grant_table_partitions('SELECT, INSERT, UPDATE, DELETE', 'account_advisory', 'evaluator'); +SELECT grant_table_partitions('SELECT, INSERT, UPDATE, DELETE', 'account_advisory', 'listener'); +SELECT grant_table_partitions('SELECT, INSERT, UPDATE, DELETE', 'account_advisory', 'vmaas_sync'); + +CREATE INDEX ON account_advisory (systems_applicable); +CREATE INDEX ON account_advisory (systems_installable); + -- repo CREATE TABLE IF NOT EXISTS repo ( diff --git a/dev/test_data.sql b/dev/test_data.sql index 386e7d17a..7987baaa2 100644 --- a/dev/test_data.sql +++ b/dev/test_data.sql @@ -7,6 +7,7 @@ DELETE FROM deleted_system; DELETE FROM repo; DELETE FROM timestamp_kv; DELETE FROM advisory_account_data; +DELETE FROM account_advisory; DELETE FROM package_account_data; DELETE FROM package; DELETE FROM package_name; diff --git a/docs/md/database.md b/docs/md/database.md index 25e3d9e4c..fa191a700 100644 --- a/docs/md/database.md +++ b/docs/md/database.md @@ -7,6 +7,7 @@ Main database tables description: - **advisory_metadata** - stores info about advisories (`description`, `summary`, `solution` etc.). It's synced and stored on trigger by `vmaas_sync` component. It allows to display detail information about the advisory. - **system_advisories** - stores info about advisories evaluated for particular systems (system - advisory M-N mapping table). `system_id` references **system_inventory.id**. Contains info when system advisory was firstly reported and patched (if so). Records are created and updated by `evaluator` component. It allows to display list of advisories related to a system. - **advisory_account_data** - stores info about all advisories detected within at least one system that belongs to a given account. So it provides overall statistics about system advisories displayed by the application. +- **account_advisory** - workspace-scoped version of `advisory_account_data`. Stores per-advisory aggregate counts (`systems_applicable`, `systems_installable`) and notification state for each workspace within an account. Keyed by `(rh_account_id, workspace_id, advisory_id)`, partitioned by `rh_account_id` (32 partitions). - **package_name** - names of the packages installed on systems - **package** - list of all packages versions, precisely all EVRAs (epoch-version-release-arch) - **system_package2** - list of packages installed on a system diff --git a/tasks/caches/refresh_account_advisory_caches_test.go b/tasks/caches/refresh_account_advisory_caches_test.go new file mode 100644 index 000000000..f015bc205 --- /dev/null +++ b/tasks/caches/refresh_account_advisory_caches_test.go @@ -0,0 +1,84 @@ +package caches + +import ( + "app/base/core" + "app/base/database" + "app/base/models" + "app/base/utils" + "testing" + + "github.com/stretchr/testify/assert" +) + +const testWorkspace = "00000000-0000-0000-0000-000000000001" + +func TestRefreshAccountAdvisoryCaches(t *testing.T) { + utils.SkipWithoutDB(t) + core.SetupTestEnvironment() + configure() + + workspace := testWorkspace + + // populate account_advisory using backfill + assert.Nil(t, database.DB.Exec("SELECT backfill_account_advisory(1)").Error) + + // capture correct counts before corrupting + countAdv1 := database.PluckInt(database.DB.Table("account_advisory"). + Where("advisory_id = 1 AND rh_account_id = 1 AND workspace_id = ?", workspace), + "systems_installable") + countAdv2 := database.PluckInt(database.DB.Table("account_advisory"). + Where("advisory_id = 2 AND rh_account_id = 1 AND workspace_id = ?", workspace), + "systems_installable") + + // set wrong counts + assert.Nil(t, database.DB.Model(&models.AccountAdvisory{}). + Where("advisory_id = 1 AND rh_account_id = 1 AND workspace_id = ?", workspace). + Update("systems_installable", 99).Error) + assert.Nil(t, database.DB.Model(&models.AccountAdvisory{}). + Where("advisory_id = 2 AND rh_account_id = 1 AND workspace_id = ?", workspace). + Update("systems_installable", 77).Error) + + // refresh should correct them + assert.Nil(t, database.DB.Exec("SELECT refresh_account_advisory_caches(NULL, 1)").Error) + + assert.Equal(t, countAdv1, database.PluckInt(database.DB.Table("account_advisory"). + Where("advisory_id = 1 AND rh_account_id = 1 AND workspace_id = ?", workspace), + "systems_installable")) + assert.Equal(t, countAdv2, database.PluckInt(database.DB.Table("account_advisory"). + Where("advisory_id = 2 AND rh_account_id = 1 AND workspace_id = ?", workspace), + "systems_installable")) + + // cleanup + database.DeleteAccountAdvisoryByAccount(t, 1) +} + +func TestRefreshAccountAdvisoryCachesRemovesOrphanedRows(t *testing.T) { + utils.SkipWithoutDB(t) + core.SetupTestEnvironment() + configure() + + workspace := testWorkspace + assert.Nil(t, database.DB.Exec("SELECT backfill_account_advisory(1)").Error) + + // mark all systems in this workspace as stale + assert.Nil(t, database.DB.Exec( + "UPDATE system_inventory SET stale = true WHERE rh_account_id = 1 AND workspace_id = ?", + workspace).Error) + + // refresh should remove rows for this workspace since no non-stale systems remain + assert.Nil(t, database.DB.Exec("SELECT refresh_account_advisory_caches(NULL, 1)").Error) + + var count int64 + assert.Nil(t, database.DB.Table("account_advisory"). + Where("rh_account_id = 1 AND workspace_id = ?", workspace). + Count(&count).Error) + assert.Equal(t, int64(0), count) + + // restore systems to non-stale + assert.Nil(t, database.DB.Exec( + "UPDATE system_inventory SET stale = false WHERE rh_account_id = 1 AND workspace_id = ?", + workspace).Error) + + // cleanup + database.DeleteAccountAdvisoryByAccount(t, 1) +} diff --git a/tasks/vmaas_sync/metrics_db_test.go b/tasks/vmaas_sync/metrics_db_test.go index f1437f974..2566436a0 100644 --- a/tasks/vmaas_sync/metrics_db_test.go +++ b/tasks/vmaas_sync/metrics_db_test.go @@ -17,12 +17,13 @@ func TestTableSizes(t *testing.T) { for _, item := range tableSizes { uniqueTables[item.Key] = true } - assert.Equal(t, 230, len(tableSizes)) - assert.Equal(t, 230, len(uniqueTables)) + assert.Equal(t, 263, len(tableSizes)) + assert.Equal(t, 263, len(uniqueTables)) assert.True(t, uniqueTables["public.system_inventory"]) // check whether table names were loaded assert.True(t, uniqueTables["public.system_patch"]) // check whether table names were loaded assert.True(t, uniqueTables["public.package"]) assert.True(t, uniqueTables["public.repo"]) + assert.True(t, uniqueTables["public.account_advisory"]) } func TestDatabaseSize(t *testing.T) {