From 3eb5aad5c97491bdf0dac7ddcbd822d16cbdf954 Mon Sep 17 00:00:00 2001 From: Michael Mraka Date: Tue, 27 Oct 2026 11:53:43 +0100 Subject: [PATCH 1/3] RHINENG-32308: parse advisory_ids on content-sources template messages --- base/mqueue/template_event.go | 1 + listener/template_test.go | 49 +++++++++++++++++++++++++++++++++++ 2 files changed, 50 insertions(+) diff --git a/base/mqueue/template_event.go b/base/mqueue/template_event.go index e7cf5b3ae..b3ec4e626 100644 --- a/base/mqueue/template_event.go +++ b/base/mqueue/template_event.go @@ -18,6 +18,7 @@ type TemplateResponse struct { Date time.Time `json:"date"` // Latest date to include snapshots for RepositoryUUIDS []string `json:"repository_uuids"` // Repositories added to the template EnvironmentID string `json:"client_environment_id"` // Environment ID used by subscription-manager & candlepin + AdvisoryIDs []string `json:"advisory_ids"` // List of advisory names (RHSA-…, RHBA-…) } type TemplateEvent struct { diff --git a/listener/template_test.go b/listener/template_test.go index 2daaf8ce5..c9927796e 100644 --- a/listener/template_test.go +++ b/listener/template_test.go @@ -179,3 +179,52 @@ func TestTemplateEmptyDescription(t *testing.T) { // cleanup deleteTemplatesInDB(t, after) } + +func TestProcessTemplateEventAdvisoryIDs(t *testing.T) { + raw := []byte(`{ + "id": "1", + "type": "com.redhat.console.repositories.template-updated", + "source": "urn:redhat:source:console:app:repositories", + "subject": "urn:redhat:subject:console:rhel:template-updated", + "time": "2024-01-02T03:04:05Z", + "redhatorgid": "org_1", + "data": [{ + "uuid": "77777777-0000-0000-0000-000000000000", + "name": "Template0", + "org_id": "org_1", + "arch": "x86_64", + "version": "8", + "date": "2024-01-02T03:04:05Z", + "repository_uuids": [], + "client_environment_id": "77777777000000000000000000000000", + "advisory_ids": ["RHBA-2024:1234", "RHSA-2024:5678"] + }] + }`) + + eType, event, err := processTemplateEvent(raw) + assert.Nil(t, err) + assert.Equal(t, "template-updated", eType) + assert.Equal(t, []string{"RHBA-2024:1234", "RHSA-2024:5678"}, event.Data[0].AdvisoryIDs) + + omitted := []byte(`{ + "id": "1", + "type": "com.redhat.console.repositories.template-created", + "source": "urn:redhat:source:console:app:repositories", + "subject": "urn:redhat:subject:console:rhel:template-created", + "time": "2024-01-02T03:04:05Z", + "redhatorgid": "org_1", + "data": [{ + "uuid": "77777777-0000-0000-0000-000000000000", + "name": "Template0", + "org_id": "org_1", + "arch": "x86_64", + "version": "8", + "date": "2024-01-02T03:04:05Z", + "repository_uuids": [], + "client_environment_id": "77777777000000000000000000000000" + }] + }`) + _, event, err = processTemplateEvent(omitted) + assert.Nil(t, err) + assert.Nil(t, event.Data[0].AdvisoryIDs) +} From 680a1000946e1dc28faf31f7e933558a38276d22 Mon Sep 17 00:00:00 2001 From: Michael Mraka Date: Tue, 27 Oct 2026 12:12:40 +0100 Subject: [PATCH 2/3] RHINENG-32308: extract template advisory snapshot apply from the content-sources fetch --- listener/template_advisories.go | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/listener/template_advisories.go b/listener/template_advisories.go index de694eb45..b1c1c8788 100644 --- a/listener/template_advisories.go +++ b/listener/template_advisories.go @@ -15,7 +15,6 @@ import ( "gorm.io/gorm/clause" ) -// nolint: funlen func syncTemplateAdvisories(ctx context.Context, accountID int, templateID int64, templateUUID, orgID string) error { current, err := callCSTemplateAdvisories(ctx, templateUUID) if err != nil { @@ -25,13 +24,19 @@ func syncTemplateAdvisories(ctx context.Context, accountID int, templateID int64 } return errors.Wrap(err, "fetching current template advisories from content-sources") } + return applyTemplateAdvisories(ctx, accountID, templateID, templateUUID, orgID, current.AdvisoryIDs) +} +// applyTemplateAdvisories diffs advisory names against template_advisory and writes the change. +// advisoryIDs is the full snapshot. Nil and empty both mean no advisories. +func applyTemplateAdvisories(ctx context.Context, accountID int, templateID int64, templateUUID, orgID string, + advisoryIDs []string) error { stored, err := lookUpTemplateAdvisories(database.DB(ctx), accountID, templateID) if err != nil { return errors.Wrap(err, "looking up stored template advisories") } - toAdd, toRemove := diffTemplateAdvisories(current.AdvisoryIDs, stored) + toAdd, toRemove := diffTemplateAdvisories(advisoryIDs, stored) if len(toAdd) == 0 && len(toRemove) == 0 { utils.LogInfo("template_uuid", templateUUID, "template advisories unchanged") return nil From 8009f19099fd80f9e32af7ce4bcc5c4e7c7e06da Mon Sep 17 00:00:00 2001 From: Michael Mraka Date: Tue, 27 Oct 2026 12:17:13 +0100 Subject: [PATCH 3/3] RHINENG-32308: sync template advisories from the message when templates_with_advisories is set --- conf/listener.env | 2 + docs/md/architecture.md | 33 +++++-- listener/listener.go | 3 + listener/template_advisories_test.go | 129 +++++++++++++++++++++++++++ listener/templates.go | 12 ++- 5 files changed, 169 insertions(+), 10 deletions(-) diff --git a/conf/listener.env b/conf/listener.env index 74b743666..77aca7b2b 100644 --- a/conf/listener.env +++ b/conf/listener.env @@ -3,4 +3,6 @@ DB_PASSWD=listener CREATED_SYSTEMS_TOPIC=patchman.evaluator.user-evaluation # template_advisory_eval=false — recalc template systems on advisory change/delete (enable with evaluator pods) +# templates_with_advisories=false — read advisory_ids from template create/update messages. +# Leave false until content-sources publishes advisory_ids; otherwise stored template_advisory rows are cleared. POD_CONFIG= diff --git a/docs/md/architecture.md b/docs/md/architecture.md index 6ed5af4b1..26072ad44 100644 --- a/docs/md/architecture.md +++ b/docs/md/architecture.md @@ -20,7 +20,9 @@ archive is uploaded, it upserts **`system_inventory`** (including `vmaas_json` w Kafka message (`patchman.evaluator.upload` topic) to evaluate the system with the `evaluator-upload` component. This component also handles system deleting events (`platform.inventory.events` Kafka topic). It consumes template events from Content Sources (`platform.content-sources.template`), upserts **`template`** metadata, -and on `template-updated` syncs **`template_advisory`** rows from the Content Sources API. When template advisories +and syncs **`template_advisory`**. With `templates_with_advisories` unset or false, only `template-updated` syncs, by +calling Content Sources `GET /templates/{uuid}/advisories/ids`. With `templates_with_advisories=true`, `template-created` +and `template-updated` sync from `advisory_ids` on the message and do not call that API. When template advisories change and `template_advisory_eval` is enabled, it sends a bulk re-evaluation message to `patchman.evaluator.user-evaluation`. See [component environment variables](../../conf/listener.env) @@ -79,14 +81,31 @@ and the [major migration runbook](major-migration-runbook.md). Using container C ### Template content update flow -When Content Sources publishes a `template-updated` event: +Content Sources publishes `template-created`, `template-updated`, and `template-deleted` on +`platform.content-sources.template`. **listener** always upserts **`template`** metadata on create and update. +`template-deleted` unassigns systems, deletes **`template_advisory`**, then deletes **`template`**. -1. **listener** upserts **`template`** metadata and calls Content Sources `GET /templates/{uuid}/advisories/ids`. -2. **listener** diffs the response against **`template_advisory`** and inserts or deletes rows. -3. If advisories changed and `template_advisory_eval=true`, **listener** looks up systems assigned to the +Advisory sync on create and update depends on `templates_with_advisories` (listener `POD_CONFIG`, default `false`): + +- **false** — only `template-updated` syncs. **listener** calls Content Sources + `GET /templates/{uuid}/advisories/ids` when `CONTENT_SOURCES_ADDRESS` and `CONTENT_SOURCES_USER` are set. + `template-created` does not sync advisories. +- **true** — `template-created` and `template-updated` use `advisory_ids` from each object in the message `data` + array (advisory names such as `RHBA-2024:1234`). The advisories API is not called. A missing key or an empty + array is an empty snapshot and removes stored **`template_advisory`** rows for that template. + +Leave the flag false until Content Sources sends `advisory_ids` on every create and update. Enabling it earlier +treats the omitted field as an empty snapshot. Do not enable it on the listener alone in `deploy/clowdapp.yaml` +as part of this change. + +After either path: + +1. **listener** diffs the snapshot against **`template_advisory`** and inserts or deletes rows. Names not yet in + **`advisory_metadata`** are skipped. +2. If advisories changed and `template_advisory_eval=true`, **listener** looks up systems assigned to the template and sends `PlatformEvent{system_ids}` to `patchman.evaluator.user-evaluation`. -4. **evaluator-user-evaluation** re-evaluates each system in the message. -5. For template-assigned systems, **evaluator** skips yum updates, calls VMaaS, and marks advisories in +3. **evaluator-user-evaluation** re-evaluates each system in the message. +4. For template-assigned systems, **evaluator** skips yum updates, calls VMaaS, and marks advisories in **`template_advisory`** as installable; other applicable advisories remain applicable. ### `template_advisory_eval` rollout diff --git a/listener/listener.go b/listener/listener.go index 62ca812ac..947d1b3c2 100644 --- a/listener/listener.go +++ b/listener/listener.go @@ -32,6 +32,7 @@ var ( excludedHostTypes map[string]bool enableBypass bool enableTemplateAdvisoryEval bool + templatesWithAdvisories bool uploadEvalTimeout time.Duration deletionThreshold time.Duration useTraceLevel bool @@ -92,6 +93,8 @@ func configureListener() { enableBypass = utils.PodConfig.GetBool("bypass", false) // Template advisory evaluation: recalc on advisory content change and template delete enableTemplateAdvisoryEval = utils.PodConfig.GetBool("template_advisory_eval", false) + // Use advisory_ids from the template kafka message instead of the content-sources API. + templatesWithAdvisories = utils.PodConfig.GetBool("templates_with_advisories", false) // How long to collect upload messages before grouping them and sending to evaluator uploadEvalTimeout = time.Duration(utils.PodConfig.GetInt("upload_eval_timeout_ms", 500)) * time.Millisecond // Ignore a system if there was a delete message in the last X hours diff --git a/listener/template_advisories_test.go b/listener/template_advisories_test.go index c522a91c7..6d344a059 100644 --- a/listener/template_advisories_test.go +++ b/listener/template_advisories_test.go @@ -6,9 +6,11 @@ import ( "app/base/core" "app/base/database" "app/base/models" + "app/base/mqueue" "app/base/utils" "testing" + "github.com/bytedance/sonic" "github.com/redhatinsights/platform-go-middlewares/v2/identity" "github.com/stretchr/testify/assert" ) @@ -167,3 +169,130 @@ func TestSyncTemplateAdvisories_TemplateNotFound(t *testing.T) { // sync was skipped and no advisories were changed database.CheckTemplateAdvisories(t, templateID, []int64{1, 2}) } + +func setTemplateAdvisoryFlags(t *testing.T, withAdvisories bool) { + t.Helper() + originalWith := templatesWithAdvisories + originalEval := enableTemplateAdvisoryEval + templatesWithAdvisories = withAdvisories + enableTemplateAdvisoryEval = false + t.Cleanup(func() { + templatesWithAdvisories = originalWith + enableTemplateAdvisoryEval = originalEval + }) +} + +func templateAdvisoryCount(t *testing.T, templateID int64) int64 { + t.Helper() + var count int64 + err := database.TestDB().Model(&models.TemplateAdvisory{}). + Where("template_id = ?", templateID). + Count(&count).Error + assert.Nil(t, err) + return count +} + +func loadTemplateByUUID(t *testing.T, templateUUID string) models.Template { + t.Helper() + var stored models.Template + err := database.TestDB(). + Where("rh_account_id = ? AND uuid = ?::uuid", accountID, templateUUID). + First(&stored).Error + assert.Nil(t, err) + return stored +} + +func TestTemplateMessageAdvisoryIDs(t *testing.T) { + utils.SkipWithoutDB(t) + core.SetupTestEnvironment() + setTemplateAdvisoryFlags(t, true) + + // Prove this path does not call Content Sources. + contentSourcesClient = nil + contentSourcesBaseURL = "" + + templateUUID := "88888888-0000-0000-0000-000000000099" + t.Cleanup(func() { + database.DeleteTemplate(t, accountID, templateUUID) + }) + + createEvent := createTempateMsg(t, "template-created", orgID, 1) + createEvent.Data[0].UUID = templateUUID + createEvent.Data[0].EnvironmentID = "88888888000000000000000000000099" + createEvent.Data[0].AdvisoryIDs = []string{"RH-1", "RH-2"} + msg, err := sonic.Marshal(createEvent) + assert.Nil(t, err) + + err = TemplatesMessageHandler(base.Context, mqueue.KafkaMessage{Value: msg}) + assert.Nil(t, err) + + stored := loadTemplateByUUID(t, templateUUID) + database.CheckTemplateAdvisories(t, stored.ID, []int64{1, 2}) + assert.Equal(t, int64(2), templateAdvisoryCount(t, stored.ID)) + + updateEvent := createEvent + updateEvent.Type = "com.redhat.console.repositories.template-updated" + updateEvent.Data[0].AdvisoryIDs = []string{"RH-1", "RH-3"} + msg, err = sonic.Marshal(updateEvent) + assert.Nil(t, err) + + err = TemplatesMessageHandler(base.Context, mqueue.KafkaMessage{Value: msg}) + assert.Nil(t, err) + + database.CheckTemplateAdvisories(t, stored.ID, []int64{1, 3}) + assert.Equal(t, int64(2), templateAdvisoryCount(t, stored.ID)) + var removed int64 + err = database.TestDB().Model(&models.TemplateAdvisory{}). + Where("template_id = ? AND advisory_id = ?", stored.ID, 2). + Count(&removed).Error + assert.Nil(t, err) + assert.Equal(t, int64(0), removed) + + updateEvent.Data[0].AdvisoryIDs = []string{} + msg, err = sonic.Marshal(updateEvent) + assert.Nil(t, err) + err = TemplatesMessageHandler(base.Context, mqueue.KafkaMessage{Value: msg}) + assert.Nil(t, err) + assert.Equal(t, int64(0), templateAdvisoryCount(t, stored.ID)) +} + +func TestTemplateUpdateStillUsesContentSourcesAPI(t *testing.T) { + utils.SkipWithoutDB(t) + core.SetupTestEnvironment() + setupContentSourcesClient(t) + setTemplateAdvisoryFlags(t, false) + + templateUUID := "88888888-0000-0000-0000-000000000098" + t.Cleanup(func() { + database.DeleteTemplate(t, accountID, templateUUID) + }) + + createEvent := createTempateMsg(t, "template-created", orgID, 1) + createEvent.Data[0].UUID = templateUUID + createEvent.Data[0].EnvironmentID = "88888888000000000000000000000098" + // Even if a producer sends the key, flag off must ignore it. + createEvent.Data[0].AdvisoryIDs = []string{"RH-2"} + msg, err := sonic.Marshal(createEvent) + assert.Nil(t, err) + + err = TemplatesMessageHandler(base.Context, mqueue.KafkaMessage{Value: msg}) + assert.Nil(t, err) + + stored := loadTemplateByUUID(t, templateUUID) + assert.Equal(t, int64(0), templateAdvisoryCount(t, stored.ID)) + + database.CreateTemplateAdvisories(t, accountID, stored.ID, []int64{2}) + + updateEvent := createEvent + updateEvent.Type = "com.redhat.console.repositories.template-updated" + updateEvent.Data[0].AdvisoryIDs = []string{"RH-2"} + msg, err = sonic.Marshal(updateEvent) + assert.Nil(t, err) + + err = TemplatesMessageHandler(base.Context, mqueue.KafkaMessage{Value: msg}) + assert.Nil(t, err) + + // Platform mock returns RH-1 and RH-3, so the message's RH-2 is not the source. + database.CheckTemplateAdvisories(t, stored.ID, []int64{1, 3}) + assert.Equal(t, int64(2), templateAdvisoryCount(t, stored.ID)) +} diff --git a/listener/templates.go b/listener/templates.go index 8f7502f65..1154bf1d1 100644 --- a/listener/templates.go +++ b/listener/templates.go @@ -151,10 +151,16 @@ func TemplateUpdate(ctx context.Context, template mqueue.TemplateResponse, event return errors.Wrap(err, "creating/updating template from message") } - if contentSourcesBaseURL != "" && utils.CoreCfg.ContentSourcesUser != "" && eventType == TemplateEventUpdate { - ctx := identity.WithIdentity(base.Context, + switch { + case templatesWithAdvisories: + err = applyTemplateAdvisories(ctx, accountID, row.ID, template.UUID, template.OrgID, template.AdvisoryIDs) + if err != nil { + return errors.Wrap(err, "syncing template advisories from message") + } + case contentSourcesBaseURL != "" && utils.CoreCfg.ContentSourcesUser != "" && eventType == TemplateEventUpdate: + csCtx := identity.WithIdentity(base.Context, utils.XRHIDForOrg(template.OrgID, utils.CoreCfg.ContentSourcesUser)) - err = syncTemplateAdvisories(ctx, accountID, row.ID, template.UUID, template.OrgID) + err = syncTemplateAdvisories(csCtx, accountID, row.ID, template.UUID, template.OrgID) if err != nil { return errors.Wrap(err, "syncing template advisories") }