feat: add account_advisory backfill job

aggregator/drift_check.go

@@ -11,7 +11,7 @@ type advisoryCounts struct {
 	SystemsInstallable int
 }
 
-func checkAdvisoryDrift(rhAccountID int, advisoryIDs []int64) {
+func CheckAdvisoryDrift(rhAccountID int, advisoryIDs []int64) {
 	if len(advisoryIDs) == 0 {
 		return
 	}

aggregator/drift_check_test.go

@@ -26,7 +26,7 @@ func TestCheckAdvisoryDriftCountMismatch(t *testing.T) {
 	hook := utils.NewTestLogHook(log.WarnLevel)
 	log.AddHook(hook)
 
-	checkAdvisoryDrift(1, []int64{1})
+	CheckAdvisoryDrift(1, []int64{1})
 
 	found := false
 	for _, entry := range hook.LogEntries {
@@ -48,7 +48,7 @@ func TestCheckAdvisoryDriftMissingFromNew(t *testing.T) {
 	hook := utils.NewTestLogHook(log.WarnLevel)
 	log.AddHook(hook)
 
-	checkAdvisoryDrift(1, []int64{1, 2})
+	CheckAdvisoryDrift(1, []int64{1, 2})
 
 	found := false
 	for _, entry := range hook.LogEntries {

aggregator/events.go

@@ -94,7 +94,7 @@ func processAdvisoryBatch(grouped map[int][]int64) {
 			continue
 		}
 
-		checkAdvisoryDrift(rhAccountID, advisoryIDs)
+		CheckAdvisoryDrift(rhAccountID, advisoryIDs)
 
 		if err := publishNewAdvisoryNotification(rhAccountID, advisoryIDs); err != nil {
 			utils.LogError("err", err, "rh_account_id", rhAccountID, "failed to publish new advisory notification")

deploy/clowdapp.yaml

@@ -594,6 +594,34 @@ objects:
                                                       key: vmaas-sync-database-password}}}
         - {name: POD_CONFIG, value: '${JOBS_CONFIG}'}
 
+    - name: account-advisory-backfill
+      activeDeadlineSeconds: ${{JOBS_TIMEOUT}}
+      schedule: ${ACCOUNT_ADVISORY_BACKFILL_SCHEDULE}
+      suspend: ${{ACCOUNT_ADVISORY_BACKFILL_SUSPEND}}
+      concurrencyPolicy: Forbid
+      podSpec:
+        image: ${IMAGE}:${IMAGE_TAG}
+        initContainers:
+          - name: check-for-db
+            image: ${IMAGE}:${IMAGE_TAG}
+            command:
+              - ./database_admin/check-upgraded.sh
+            env:
+            - {name: POD_CONFIG, value: '${DATABASE_ADMIN_CONFIG}'}
+        command:
+          - ./scripts/entrypoint.sh
+          - job
+          - account_advisory_backfill
+        env:
+        - {name: LOG_LEVEL, value: '${LOG_LEVEL_JOBS}'}
+        - {name: GIN_MODE, value: '${GIN_MODE}'}
+        - {name: SENTRY_DSN, valueFrom: {secretKeyRef: {name: patchman-sentry, key: sentry-dsn}}}
+        - {name: DB_DEBUG, value: '${DB_DEBUG_JOBS}'}
+        - {name: DB_USER, value: vmaas_sync}
+        - {name: DB_PASSWD, valueFrom: {secretKeyRef: {name: patchman-engine-database-passwords,
+                                                      key: vmaas-sync-database-password}}}
+        - {name: POD_CONFIG, value: '${JOBS_CONFIG}'}
+
     database:
       name: patchman
       version: 16
@@ -883,6 +911,9 @@ parameters:
 # Clean advisory_account_data
 - {name: CLEAN_AAD_SCHEDULE, value: '0 12 * * *'} # Cronjob schedule definition
 - {name: CLEAN_AAD_SUSPEND, value: 'false'} # Disable cronjob execution
+# Backfill account_advisory
+- {name: ACCOUNT_ADVISORY_BACKFILL_SCHEDULE, value: '0 3 * * *'} # Cronjob schedule definition
+- {name: ACCOUNT_ADVISORY_BACKFILL_SUSPEND, value: 'true'} # Suspended until ready to run
 
 # Database admin
 - {name: MIGRATION_TIMEOUT, value: '7200'}  # 2h timeout for db-migration job

main.go

@@ -76,6 +76,8 @@ func runJob(name string) {
 		caches.RunPackageRefresh()
 	case "repack":
 		repack.RunRepack()
+	case "account_advisory_backfill":
+		caches.RunAccountAdvisoryBackfill()
 	case "clean_advisory_account_data":
 		cleaning.RunCleanAdvisoryAccountData()
 	}

tasks/caches/backfill_account_advisory.go

@@ -0,0 +1,65 @@
+package caches
+
+import (
+	"app/aggregator"
+	"app/base/database"
+	"app/base/utils"
+	"app/tasks"
+	"sync"
+
+	"gorm.io/gorm"
+)
+
+func BackfillAccountAdvisory() {
+	var wg sync.WaitGroup
+	backfillAccountAdvisoryPerAccounts(&wg)
+	wg.Wait()
+}
+
+func backfillAccountAdvisoryPerAccounts(wg *sync.WaitGroup) {
+	var rhAccountIDs []int
+	err := tasks.WithReadReplicaTx(func(tx *gorm.DB) error {
+		return tx.Table("rh_account").
+			Order("hash_partition_id(id, 128), id").
+			Pluck("id", &rhAccountIDs).Error
+	})
+	if err != nil {
+		utils.LogError("err", err, "unable to load rh_account IDs for account_advisory backfill")
+		return
+	}
+
+	utils.LogInfo("accounts", len(rhAccountIDs), "starting account_advisory backfill")
+
+	guard := make(chan struct{}, 4)
+
+	for i, rhAccountID := range rhAccountIDs {
+		guard <- struct{}{}
+		wg.Add(1)
+		go func(i, rhAccountID int) {
+			defer func() {
+				<-guard
+				wg.Done()
+			}()
+
+			err := tasks.WithTx(func(tx *gorm.DB) error {
+				utils.LogInfo("i", i, "rh_account_id", rhAccountID, "backfilling account_advisory")
+				return tx.Exec("SELECT backfill_account_advisory(?)", rhAccountID).Error
+			})
+			if err != nil {
+				utils.LogError("err", err, "rh_account_id", rhAccountID, "failed to backfill account_advisory")
+				return
+			}
+			utils.LogInfo("i", i, "rh_account_id", rhAccountID, "backfilled account_advisory")
+
+			var advisoryIDs []int64
+			if err := database.DB.Table("account_advisory").
+				Where("rh_account_id = ?", rhAccountID).
+				Distinct("advisory_id").
+				Pluck("advisory_id", &advisoryIDs).Error; err != nil {
+				utils.LogError("err", err, "rh_account_id", rhAccountID, "failed to load advisory IDs for drift check")
+				return
+			}
+			aggregator.CheckAdvisoryDrift(rhAccountID, advisoryIDs)
+		}(i, rhAccountID)
+	}
+}

tasks/caches/caches.go

@@ -21,6 +21,14 @@ func RunAdvisoryRefresh() {
 	RefreshAdvisoryCaches()
 }
 
+func RunAccountAdvisoryBackfill() {
+	tasks.HandleContextCancel(tasks.WaitAndExit)
+	configure()
+	utils.LogInfo("Starting account_advisory backfill")
+	BackfillAccountAdvisory()
+	utils.LogInfo("Finished account_advisory backfill")
+}
+
 func RunPackageRefresh() {
 	tasks.HandleContextCancel(tasks.WaitAndExit)
 	configure()