diff --git a/aggregator/drift_check.go b/aggregator/drift_check.go index 0ed8e4d5f..54ff8aa9c 100644 --- a/aggregator/drift_check.go +++ b/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 } diff --git a/aggregator/drift_check_test.go b/aggregator/drift_check_test.go index 407a1c58b..6eec2d99c 100644 --- a/aggregator/drift_check_test.go +++ b/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 { diff --git a/aggregator/events.go b/aggregator/events.go index a4ff24d75..30930b827 100644 --- a/aggregator/events.go +++ b/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") diff --git a/deploy/clowdapp.yaml b/deploy/clowdapp.yaml index 11ea1e94c..706c3516d 100644 --- a/deploy/clowdapp.yaml +++ b/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}'} + - name: system-advisories-0-recovery # One-shot Job (no schedule): runs on deploy / CJI like db-migration. # No-op unless JOBS_CONFIG includes system_advisories_0_recovery=true. @@ -934,6 +962,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 diff --git a/main.go b/main.go index c744e6b9d..87786c96f 100644 --- a/main.go +++ b/main.go @@ -77,6 +77,8 @@ func runJob(name string) { caches.RunPackageRefresh() case "repack": repack.RunRepack() + case "account_advisory_backfill": + caches.RunAccountAdvisoryBackfill() case "clean_advisory_account_data": cleaning.RunCleanAdvisoryAccountData() case "system_advisories_0_recovery": diff --git a/tasks/caches/backfill_account_advisory.go b/tasks/caches/backfill_account_advisory.go new file mode 100644 index 000000000..ff762783e --- /dev/null +++ b/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) + } +} diff --git a/tasks/caches/caches.go b/tasks/caches/caches.go index 51175aa35..39384439b 100644 --- a/tasks/caches/caches.go +++ b/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()