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