RHINENG-26117: Add recompute infrastructure for account_advisory

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)

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

database_admin/migrations/154_account_advisory.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS account_advisory;

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');

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;

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;

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
 (

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;

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

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)
+}