From 19394a1de3567bbdcd82bca04cff3a2ed5c725a0 Mon Sep 17 00:00:00 2001 From: TenSt Date: Thu, 30 Jul 2026 13:27:33 +0200 Subject: [PATCH 1/2] RHINENG-26348: add skip notifications for platform events --- base/mqueue/platform_event.go | 4 ++ base/mqueue/platform_event_test.go | 25 +++++++++ evaluator/evaluate.go | 8 +-- evaluator/notifications.go | 47 +++++++++++------ evaluator/notifications_test.go | 84 +++++++++++++++++++++++++++++- 5 files changed, 147 insertions(+), 21 deletions(-) diff --git a/base/mqueue/platform_event.go b/base/mqueue/platform_event.go index c7ad07c7b..f5c58aeca 100644 --- a/base/mqueue/platform_event.go +++ b/base/mqueue/platform_event.go @@ -22,6 +22,10 @@ type PlatformEvent struct { URL *string `json:"url"` SystemIDs []uuid.UUID `json:"system_ids,omitempty"` RequestIDs []string `json:"request_ids,omitempty"` + // SkipNotifications suppresses instant advisory notification publish for this event. + // Evaluator still marks matching advisory_account_data.notified so later evals do not flood. + // Used by recovery recalc; omit/false for normal upload/recalc traffic. + SkipNotifications bool `json:"skip_notifications,omitempty"` } type EvalData struct { diff --git a/base/mqueue/platform_event_test.go b/base/mqueue/platform_event_test.go index dc0f8b10e..8da835224 100644 --- a/base/mqueue/platform_event_test.go +++ b/base/mqueue/platform_event_test.go @@ -9,6 +9,31 @@ import ( "github.com/stretchr/testify/assert" ) +func TestPlatformEventSkipNotificationsJSON(t *testing.T) { + orgID := "org_1" + event := PlatformEvent{ + AccountID: 1, + OrgID: &orgID, + SkipNotifications: true, + } + data, err := sonic.Marshal(event) + assert.NoError(t, err) + + var parsed PlatformEvent + assert.NoError(t, sonic.Unmarshal(data, &parsed)) + assert.True(t, parsed.SkipNotifications) + + // omitempty: false must not appear in JSON; fresh unmarshal defaults to false + event.SkipNotifications = false + data, err = sonic.Marshal(event) + assert.NoError(t, err) + assert.NotContains(t, string(data), "skip_notifications") + + var parsedFalse PlatformEvent + assert.NoError(t, sonic.Unmarshal(data, &parsedFalse)) + assert.False(t, parsedFalse.SkipNotifications) +} + func TestWriteEventsOfInventoryAccounts(t *testing.T) { var ( acc = 1 diff --git a/evaluator/evaluate.go b/evaluator/evaluate.go index c857fdb4f..da6daa90e 100644 --- a/evaluator/evaluate.go +++ b/evaluator/evaluate.go @@ -520,9 +520,11 @@ func evaluateAndStore(system *models.SystemPlatformV2, } } - // Send instant notification with new advisories - if enableInstantNotifications { - err = publishNewAdvisoriesNotification(tx, system, event.GetOrgID(), systemAdvisoriesNew) + // Instant notifications, or mark-notified only when the event opts out of publishing + // (e.g. recovery recalc with skip_notifications). + if event.SkipNotifications || enableInstantNotifications { + err = publishNewAdvisoriesNotification(tx, system, event.GetOrgID(), systemAdvisoriesNew, + event.SkipNotifications) if err != nil { evaluationCnt.WithLabelValues("error-advisory-notification").Inc() utils.LogError("orgID", event.GetOrgID(), "inventoryID", system.GetInventoryID(), "err", err, diff --git a/evaluator/notifications.go b/evaluator/notifications.go index 962cdddc3..4cb87a71d 100644 --- a/evaluator/notifications.go +++ b/evaluator/notifications.go @@ -66,12 +66,24 @@ func getSystemTags(tx *gorm.DB, system *models.SystemPlatformV2) ([]ntf.SystemTa return tags, nil } -func publishNewAdvisoriesNotification(tx *gorm.DB, system *models.SystemPlatformV2, orgID string, - newAdvisories SystemAdvisoryMap) error { - if notificationsPublisher == nil { +func markAdvisoriesNotified(tx *gorm.DB, accountID int, advisoryIDs []int64) error { + if len(advisoryIDs) == 0 { return nil } + err := tx.Table("advisory_account_data"). + Where("rh_account_id = ? AND advisory_id IN (?)", accountID, advisoryIDs). + Update("notified", time.Now()).Error + if err != nil { + return errors.Wrap(err, "updating notified column failed") + } + return nil +} +// publishNewAdvisoriesNotification publishes instant new-advisory notifications unless +// skipPublish is true. In both cases, matching advisory_account_data rows are marked notified +// when there is something to notify about (so skipPublish still prevents later flood). +func publishNewAdvisoriesNotification(tx *gorm.DB, system *models.SystemPlatformV2, orgID string, + newAdvisories SystemAdvisoryMap, skipPublish bool) error { defer utils.ObserveSecondsSince(time.Now(), evaluationPartDuration.WithLabelValues("advisory-notification-publish")) advisories, err := getUnnotifiedAdvisories(tx, system.Inventory.RhAccountID, newAdvisories) @@ -82,6 +94,21 @@ func publishNewAdvisoriesNotification(tx *gorm.DB, system *models.SystemPlatform return nil } + advisoryIDs := make([]int64, 0, len(advisories)) + for _, a := range advisories { + advisoryIDs = append(advisoryIDs, a.AdvisoryID) + } + + if skipPublish { + utils.LogInfo("inventoryID", system.GetInventoryID(), "advisoryIDs", advisoryIDs, "orgID", orgID, + "skipping advisory notification publish") + return markAdvisoriesNotified(tx, system.Inventory.RhAccountID, advisoryIDs) + } + + if notificationsPublisher == nil { + return nil + } + events := make([]ntf.Event, 0, len(advisories)) for _, advisory := range advisories { // At least empty metadata required to avoid NPE further on at the time of writing. @@ -108,20 +135,8 @@ func publishNewAdvisoriesNotification(tx *gorm.DB, system *models.SystemPlatform return errors.Wrap(err, "writing message to notifications publisher failed") } - advisoryIDs := make([]int64, 0, len(advisories)) - for _, a := range advisories { - advisoryIDs = append(advisoryIDs, a.AdvisoryID) - } - utils.LogInfo("inventoryID", system.GetInventoryID(), "advisoryIDs", advisoryIDs, "orgID", orgID, "notification sent successfully") - err = tx.Table("advisory_account_data"). - Where("rh_account_id = ? AND advisory_id IN (?)", system.Inventory.RhAccountID, advisoryIDs). - Update("notified", time.Now()).Error - if err != nil { - return errors.Wrap(err, "updating notified column failed") - } - - return nil + return markAdvisoriesNotified(tx, system.Inventory.RhAccountID, advisoryIDs) } diff --git a/evaluator/notifications_test.go b/evaluator/notifications_test.go index 5f0bdc8d7..fce75c4a3 100644 --- a/evaluator/notifications_test.go +++ b/evaluator/notifications_test.go @@ -86,6 +86,50 @@ func TestAdvisoriesNotificationPublish(t *testing.T) { database.DeleteAdvisoryAccountData(t, rhAccountID, oldSystemAdvisoryIDs) } +// TestAdvisoriesNotificationSkipPublishViaEvaluate runs a full evaluateHandler path with +// skip_notifications set: advisories are stored and marked notified, but no Kafka notify is sent. +func TestAdvisoriesNotificationSkipPublishViaEvaluate(t *testing.T) { + utils.SkipWithoutDB(t) + utils.SkipWithoutPlatform(t) + core.SetupTestEnvironment() + + configure() + loadCache() + mockWriter := mqueue.MockKafkaWriter{} + notificationsPublisher = &mockWriter + + expectedAddedAdvisories := []string{"RH-1", "RH-2", "RH-100"} + expectedAdvisoryIDs := []int64{1, 2} + oldSystemAdvisoryIDs := []int64{1, 3, 4} + + database.DeleteSystemAdvisories(t, testDBID, expectedAdvisoryIDs) + database.DeleteAdvisoryAccountData(t, rhAccountID, expectedAdvisoryIDs) + database.CreateSystemAdvisories(t, rhAccountID, testDBID, oldSystemAdvisoryIDs) + database.CreateAdvisoryAccountData(t, rhAccountID, oldSystemAdvisoryIDs, 1) + database.CheckCachesValid(t) + database.CheckAdvisoriesAccountDataNotified(t, rhAccountID, oldSystemAdvisoryIDs, false) + + orgID := "1234567" + data, err := sonic.Marshal(mqueue.PlatformEvent{ + SystemIDs: []uuid.UUID{testInventoryID}, + RequestIDs: []string{"request-skip-notif"}, + AccountID: rhAccountID, + OrgID: &orgID, + SkipNotifications: true, + }) + assert.NoError(t, err) + err = evaluateHandler(mqueue.KafkaMessage{Value: data}) + assert.NoError(t, err) + + advisoryIDs := database.CheckAdvisoriesInDB(t, expectedAddedAdvisories) + database.CheckAdvisoriesAccountDataNotified(t, rhAccountID, expectedAdvisoryIDs, true) + assert.Empty(t, mockWriter.Messages, "no notification should be sent when skip_notifications is set") + + database.DeleteSystemAdvisories(t, testDBID, advisoryIDs) + database.DeleteAdvisoryAccountData(t, rhAccountID, advisoryIDs) + database.DeleteAdvisoryAccountData(t, rhAccountID, oldSystemAdvisoryIDs) +} + func TestAdvisoriesNotificationMessage(t *testing.T) { events := make([]ntf.Event, 1) events[0] = ntf.Event{ @@ -188,7 +232,7 @@ func TestAdvisoriesNotificationAlreadyNotified(t *testing.T) { "RH-2": {AdvisoryID: 2}, } - err := publishNewAdvisoriesNotification(database.DB, system, orgID, newAdvs) + err := publishNewAdvisoriesNotification(database.DB, system, orgID, newAdvs, false) assert.NoError(t, err) assert.Empty(t, mockWriter.Messages, "no notification should be sent when all advisories are already notified") } @@ -215,10 +259,46 @@ func TestAdvisoriesNotificationEmptyAdvisoryMap(t *testing.T) { } // An empty map means there is nothing to query — no messages should be produced regardless. - publishNewAdvisoriesNotification(database.DB, system, orgID, SystemAdvisoryMap{}) //nolint:errcheck + publishNewAdvisoriesNotification(database.DB, system, orgID, SystemAdvisoryMap{}, false) //nolint:errcheck assert.Empty(t, mockWriter.Messages, "no notification should be sent when the advisory map is empty") } +// TestAdvisoriesNotificationSkipPublish verifies recovery-style skip_notifications: no Kafka +// message is sent, but advisory_account_data.notified is still set. +func TestAdvisoriesNotificationSkipPublish(t *testing.T) { + utils.SkipWithoutDB(t) + core.SetupTestEnvironment() + configure() + + mockWriter := mqueue.MockKafkaWriter{} + notificationsPublisher = &mockWriter + + advisoryIDs := []int64{1, 2} + database.DeleteAdvisoryAccountData(t, rhAccountID, advisoryIDs) + database.CreateAdvisoryAccountData(t, rhAccountID, advisoryIDs, 1) + database.CheckAdvisoriesAccountDataNotified(t, rhAccountID, advisoryIDs, false) + defer database.DeleteAdvisoryAccountData(t, rhAccountID, advisoryIDs) + + system := &models.SystemPlatformV2{ + Inventory: models.SystemInventory{ + ID: 1, + RhAccountID: rhAccountID, + InventoryID: uuid.MustParse("00000000-0000-0000-0000-000000000001"), + DisplayName: "display name", + }, + Patch: models.SystemPatch{}, + } + newAdvs := SystemAdvisoryMap{ + "RH-1": {AdvisoryID: 1}, + "RH-2": {AdvisoryID: 2}, + } + + err := publishNewAdvisoriesNotification(database.DB, system, orgID, newAdvs, true) + assert.NoError(t, err) + assert.Empty(t, mockWriter.Messages, "no notification should be sent when skipPublish is true") + database.CheckAdvisoriesAccountDataNotified(t, rhAccountID, advisoryIDs, true) +} + // TestGetUnnotifiedAdvisoriesReturnsEmpty documents the return-type contract of // getUnnotifiedAdvisories: when all candidate advisories are already notified the function // must return a non-nil empty slice (not nil). This prevents a future nil-vs-empty regression From 52897badf7a6bec524612179a1ff14d35797df89 Mon Sep 17 00:00:00 2001 From: TenSt Date: Thu, 30 Jul 2026 18:52:18 +0200 Subject: [PATCH 2/2] RHINENG-26348: add job to refill system_advisories_0 partition --- base/mqueue/platform_event.go | 47 +++++++---- base/mqueue/platform_event_test.go | 28 +++++++ deploy/clowdapp.yaml | 53 ++++++++++++- main.go | 3 + tasks/config.go | 2 + .../system_advisories_0_recovery/recovery.go | 77 +++++++++++++++++++ .../recovery_test.go | 60 +++++++++++++++ 7 files changed, 252 insertions(+), 18 deletions(-) create mode 100644 tasks/system_advisories_0_recovery/recovery.go create mode 100644 tasks/system_advisories_0_recovery/recovery_test.go diff --git a/base/mqueue/platform_event.go b/base/mqueue/platform_event.go index f5c58aeca..b138c1358 100644 --- a/base/mqueue/platform_event.go +++ b/base/mqueue/platform_event.go @@ -76,16 +76,18 @@ func writePlatformEvents(ctx context.Context, w Writer, events ...PlatformEvent) return w.WriteMessages(ctx, msgs...) } -func batchSize(grouped map[int][]uuid.UUID) int { - // compute how many batches we will create +func batchCount(grouped map[int][]uuid.UUID, size int) int { + if size <= 0 { + size = BatchSize + } var batches = 0 for _, ev := range grouped { - batches += len(ev)/BatchSize + 1 + batches += (len(ev) + size - 1) / size } return batches } -func (evals EvalDataSlice) getAccountEvalData() (int, accountInventories, accountRequests, orgIDs) { +func (evals EvalDataSlice) getAccountEvalData(size int) (int, accountInventories, accountRequests, orgIDs) { // group systems by account invs := accountInventories{} reqs := accountRequests{} @@ -97,30 +99,41 @@ func (evals EvalDataSlice) getAccountEvalData() (int, accountInventories, accoun orgs[e.RhAccountID] = e.OrgID } } - return batchSize(invs), invs, reqs, orgs + return batchCount(invs, size), invs, reqs, orgs } func (evals EvalDataSlice) WriteEvents(ctx context.Context, w Writer) error { - batches, accInvs, reqs, orgs := evals.getAccountEvalData() - // create events, per BatchSize of systems from one account + return evals.writeEvents(ctx, w, BatchSize, false) +} + +// WriteEventsSkipNotifications publishes recalc events in batches of batchSize systems +// per account with SkipNotifications set. Used by one-off recovery jobs. +func (evals EvalDataSlice) WriteEventsSkipNotifications(ctx context.Context, w Writer, batchSize int) error { + return evals.writeEvents(ctx, w, batchSize, true) +} + +func (evals EvalDataSlice) writeEvents(ctx context.Context, w Writer, size int, skipNotifications bool) error { + if size <= 0 { + size = BatchSize + } + batches, accInvs, reqs, orgs := evals.getAccountEvalData(size) now := types.Rfc3339Timestamp(time.Now()) events := make(PlatformEvents, 0, batches) for acc, invs := range accInvs { - for start := 0; start < len(invs); start += BatchSize { - end := start + BatchSize + for start := 0; start < len(invs); start += size { + end := start + size if end > len(invs) { end = len(invs) } events = append(events, PlatformEvent{ - Timestamp: &now, - AccountID: acc, - SystemIDs: invs[start:end], - RequestIDs: reqs[acc][start:end], - OrgID: orgs[acc], + Timestamp: &now, + AccountID: acc, + SystemIDs: invs[start:end], + RequestIDs: reqs[acc][start:end], + OrgID: orgs[acc], + SkipNotifications: skipNotifications, }) } } - // write events to queue - err := writePlatformEvents(ctx, w, events...) - return err + return writePlatformEvents(ctx, w, events...) } diff --git a/base/mqueue/platform_event_test.go b/base/mqueue/platform_event_test.go index 8da835224..7f9cf3923 100644 --- a/base/mqueue/platform_event_test.go +++ b/base/mqueue/platform_event_test.go @@ -60,4 +60,32 @@ func TestWriteEventsOfInventoryAccounts(t *testing.T) { assert.True(t, len(event.SystemIDs) == 2) assert.Equal(t, inv2, event.SystemIDs[0]) assert.Equal(t, inv3, event.SystemIDs[1]) + assert.False(t, event.SkipNotifications) +} + +func TestWriteEventsSkipNotificationsChunking(t *testing.T) { + acc := 7 + orgID := "org_recovery" + invs := make(EvalDataSlice, 0, 501) + for i := 0; i < 501; i++ { + invs = append(invs, EvalData{ + InventoryID: uuid.New(), + RhAccountID: acc, + OrgID: &orgID, + }) + } + + writer := &MockKafkaWriter{} + assert.NoError(t, invs.WriteEventsSkipNotifications(context.Background(), writer, 500)) + assert.Equal(t, 2, len(writer.Messages)) + + var first, second PlatformEvent + assert.NoError(t, sonic.Unmarshal(writer.Messages[0].Value, &first)) + assert.NoError(t, sonic.Unmarshal(writer.Messages[1].Value, &second)) + assert.True(t, first.SkipNotifications) + assert.True(t, second.SkipNotifications) + assert.Equal(t, 500, len(first.SystemIDs)) + assert.Equal(t, 1, len(second.SystemIDs)) + assert.Equal(t, acc, first.AccountID) + assert.Equal(t, orgID, first.GetOrgID()) } diff --git a/deploy/clowdapp.yaml b/deploy/clowdapp.yaml index 2834554b7..419121e88 100644 --- a/deploy/clowdapp.yaml +++ b/deploy/clowdapp.yaml @@ -594,6 +594,39 @@ objects: 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. + completions: 1 + parallelism: 1 + activeDeadlineSeconds: ${{JOBS_TIMEOUT}} + 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 + - system_advisories_0_recovery + 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: KAFKA_GROUP, value: patchman} + - {name: KAFKA_WRITER_MAX_ATTEMPTS, value: '${KAFKA_WRITER_MAX_ATTEMPTS}'} + - {name: EVAL_TOPIC, value: patchman.evaluator.recalc} + - {name: SSL_CERT_DIR, value: '${SSL_CERT_DIR}'} + - {name: POD_CONFIG, value: '${JOBS_CONFIG}'} # set system_advisories_0_recovery=true for cutover + database: name: patchman version: 16 @@ -637,6 +670,22 @@ objects: jobs: - db-migration +# One-off system_advisories_0 recovery recalc. Keep disabled; enable for cutover deploy with +# JOBS_CONFIG=system_advisories_0_recovery=true, then disable again. +- apiVersion: cloud.redhat.com/v1alpha1 + kind: ClowdJobInvocation + metadata: + annotations: + clowder.redhat.com/expected-image-tag: ${IMAGE_TAG} + labels: + app: patchman + name: system-advisories-0-recovery-${IMAGE_TAG}-${CJI_UID} + spec: + appName: patchman + disabled: ${{SA0_RECOVERY_DISABLED}} + jobs: + - system-advisories-0-recovery + - apiVersion: metrics.console.redhat.com/v1alpha1 kind: FloorPlan metadata: @@ -850,7 +899,7 @@ parameters: - {name: JOBS_TIMEOUT, value: '1800'} # 30 min timeout for jobs - {name: PROMETHEUS_PUSHGATEWAY, required: true, value: "pushgateway"} - {name: DB_READ_REPLICA_ENABLED_JOBS, value: 'TRUE'} -- {name: JOBS_CONFIG, value: ''} +- {name: JOBS_CONFIG, value: ''} # e.g. system_advisories_0_recovery=true for SA0 recovery job cutover # DB migration - {name: DB_MIGRATION_DISABLED, value: 'false'} # Disable db-migration job execution @@ -858,6 +907,8 @@ parameters: description: Unique db-migration CJI name suffix generate: expression from: '[a-z0-9]{6}' +# system_advisories_0 recovery recalc CJI (one-off; keep disabled unless cutover) +- {name: SA0_RECOVERY_DISABLED, value: 'true'} # set false + JOBS_CONFIG=system_advisories_0_recovery=true for cutover # VMaaS sync - {name: VMAAS_SYNC_SCHEDULE, value: '*/5 * * * *'} # Cronjob schedule definition - {name: VMAAS_SYNC_SUSPEND, value: 'false'} # Disable cronjob execution diff --git a/main.go b/main.go index 6bc9d5965..c744e6b9d 100644 --- a/main.go +++ b/main.go @@ -12,6 +12,7 @@ import ( "app/tasks/caches" "app/tasks/cleaning" "app/tasks/repack" + "app/tasks/system_advisories_0_recovery" "app/tasks/system_culling" "app/tasks/vmaas_sync" "app/turnpike" @@ -78,5 +79,7 @@ func runJob(name string) { repack.RunRepack() case "clean_advisory_account_data": cleaning.RunCleanAdvisoryAccountData() + case "system_advisories_0_recovery": + system_advisories_0_recovery.Run() } } diff --git a/tasks/config.go b/tasks/config.go index d059dc58c..8735c36f0 100644 --- a/tasks/config.go +++ b/tasks/config.go @@ -42,4 +42,6 @@ var ( MaxChangedPackages = utils.PodConfig.GetInt("max_changed_packages", 30000) // prune deleted_system table records older than threshold DeletedSystemsThreshold = time.Hour * time.Duration(utils.PodConfig.GetInt("system_delete_hrs", 4)) + // One-off: publish recalc for non-stale system_advisories hash remainder 0 (default off) + EnableSystemAdvisories0Recovery = utils.PodConfig.GetBool("system_advisories_0_recovery", false) ) diff --git a/tasks/system_advisories_0_recovery/recovery.go b/tasks/system_advisories_0_recovery/recovery.go new file mode 100644 index 000000000..ef28a071f --- /dev/null +++ b/tasks/system_advisories_0_recovery/recovery.go @@ -0,0 +1,77 @@ +package system_advisories_0_recovery + +import ( + "app/base" + "app/base/core" + "app/base/mqueue" + "app/base/utils" + "app/tasks" + "time" +) + +const ( + systemAdvisoriesPartitions = 32 + systemAdvisoriesRemainder = 0 + // recoveryBatchSize matches the cutover plan: 500 systems per Kafka message. + recoveryBatchSize = 500 +) + +var evalWriter mqueue.Writer + +func Configure() { + core.ConfigureApp() + evalTopic := utils.FailIfEmpty(utils.CoreCfg.EvalTopic, "EVAL_TOPIC") + evalWriter = mqueue.NewKafkaWriterFromEnv(evalTopic) +} + +func Run() { + tasks.HandleContextCancel(tasks.WaitAndExit) + Configure() + defer utils.LogPanics(true) + + if !tasks.EnableSystemAdvisories0Recovery { + utils.LogInfo("system_advisories_0_recovery disabled (set system_advisories_0_recovery=true in JOBS_CONFIG), skipping") //nolint:lll + return + } + + utils.LogInfo("Starting system_advisories_0 recovery recalc publish") + if err := publishBucket0Recalc(); err != nil { + utils.LogError("err", err, "system_advisories_0 recovery failed") + return + } + utils.LogInfo("system_advisories_0 recovery recalc publish finished") +} + +func publishBucket0Recalc() error { + inventoryAIDs, err := getNonStaleBucket0InventoryIDs() + if err != nil { + return err + } + utils.LogInfo("count", len(inventoryAIDs), "non-stale bucket-0 systems selected for recovery recalc") + + start := time.Now() + err = mqueue.EvalDataSlice(inventoryAIDs).WriteEventsSkipNotifications(base.Context, evalWriter, recoveryBatchSize) + if err != nil { + utils.LogError("err", err, "sending recovery recalc messages failed") + return err + } + utils.LogInfo("count", len(inventoryAIDs), "seconds", time.Since(start).Seconds(), + "systems sent to recovery recalc with skip_notifications") + return nil +} + +func getNonStaleBucket0InventoryIDs() ([]mqueue.EvalData, error) { + var inventoryAIDs []mqueue.EvalData + err := tasks.CancelableDB().Table("system_inventory si"). + Select("si.inventory_id, si.rh_account_id, ra.org_id"). + Joins("JOIN rh_account ra ON ra.id = si.rh_account_id"). + Where("si.stale = false"). + Where("satisfies_hash_partition('system_advisories'::regclass, ?, ?, si.rh_account_id)", + systemAdvisoriesPartitions, systemAdvisoriesRemainder). + Order("si.rh_account_id, si.id"). + Scan(&inventoryAIDs).Error + if err != nil { + return nil, err + } + return inventoryAIDs, nil +} diff --git a/tasks/system_advisories_0_recovery/recovery_test.go b/tasks/system_advisories_0_recovery/recovery_test.go new file mode 100644 index 000000000..72350b962 --- /dev/null +++ b/tasks/system_advisories_0_recovery/recovery_test.go @@ -0,0 +1,60 @@ +package system_advisories_0_recovery + +import ( + "app/base/core" + "app/base/database" + "app/base/mqueue" + "app/base/utils" + "context" + "testing" + + "github.com/bytedance/sonic" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestGetNonStaleBucket0InventoryIDs(t *testing.T) { + utils.SkipWithoutDB(t) + core.SetupTestEnvironment() + + ids, err := getNonStaleBucket0InventoryIDs() + require.NoError(t, err) + + var expected int64 + err = database.DB.Raw(` + SELECT count(*) + FROM system_inventory si + WHERE si.stale = false + AND satisfies_hash_partition('system_advisories'::regclass, ?, ?, si.rh_account_id) + `, systemAdvisoriesPartitions, systemAdvisoriesRemainder).Scan(&expected).Error + require.NoError(t, err) + assert.Equal(t, int(expected), len(ids)) + + for _, row := range ids { + var inBucket bool + err = database.DB.Raw( + `SELECT satisfies_hash_partition('system_advisories'::regclass, ?, ?, ?)`, + systemAdvisoriesPartitions, systemAdvisoriesRemainder, row.RhAccountID, + ).Scan(&inBucket).Error + require.NoError(t, err) + assert.True(t, inBucket) + assert.NotEqual(t, uuid.Nil, row.InventoryID) + } +} + +func TestPublishSetsSkipNotifications(t *testing.T) { + orgID := "org_1" + evals := mqueue.EvalDataSlice{ + {InventoryID: uuid.MustParse("00000000-0000-0000-0000-000000000001"), RhAccountID: 1, OrgID: &orgID}, + {InventoryID: uuid.MustParse("00000000-0000-0000-0000-000000000002"), RhAccountID: 1, OrgID: &orgID}, + } + writer := &mqueue.MockKafkaWriter{} + require.NoError(t, evals.WriteEventsSkipNotifications(context.Background(), writer, recoveryBatchSize)) + require.Len(t, writer.Messages, 1) + + var event mqueue.PlatformEvent + require.NoError(t, sonic.Unmarshal(writer.Messages[0].Value, &event)) + assert.True(t, event.SkipNotifications) + assert.Equal(t, 2, len(event.SystemIDs)) +}