RHINENG-26348: add job to refill 0 partition

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 {
@@ -72,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{}
@@ -93,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...)
 }

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
@@ -35,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())
 }

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,14 +899,16 @@ 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
 - name: CJI_UID
   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

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,

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