diff --git a/aggregator/aggregator.go b/aggregator/aggregator.go
index 4bcbf320d..b151c6233 100644
--- a/aggregator/aggregator.go
+++ b/aggregator/aggregator.go
@@ -6,6 +6,7 @@ import (
"app/base/mqueue"
"app/base/utils"
"sync"
+ "time"
"github.com/gin-gonic/gin"
)
@@ -18,11 +19,14 @@ var (
func readPodConfig() {
consumerCount = utils.PodConfig.GetInt("consumer_count", 1)
- enableNotifications = utils.PodConfig.GetBool("instant_notifications", false)
+ enableNotifications = utils.PodConfig.GetBool("enable_notifications", false)
+ batchSize = utils.PodConfig.GetInt("advisory_batch_size", 4000)
+ flushTimeout = time.Duration(utils.PodConfig.GetInt("advisory_flush_timeout_ms", 500)) * time.Millisecond
}
func configure() {
advisoryUpdateTopic = utils.FailIfEmpty(utils.CoreCfg.AdvisoryUpdateTopic, "ADVISORY_UPDATE_TOPIC")
+ initBuffer()
configureNotifications()
}
diff --git a/aggregator/aggregator_test.go b/aggregator/aggregator_test.go
new file mode 100644
index 000000000..cb0e9e0ca
--- /dev/null
+++ b/aggregator/aggregator_test.go
@@ -0,0 +1,15 @@
+package aggregator
+
+import (
+ "app/base/database"
+ "os"
+ "testing"
+)
+
+func TestMain(m *testing.M) {
+ exitCode := m.Run()
+ if database.DB != nil {
+ database.DB.Exec("SELECT refresh_account_advisory_caches_multi(NULL, 1)")
+ }
+ os.Exit(exitCode)
+}
diff --git a/aggregator/drift_check.go b/aggregator/drift_check.go
new file mode 100644
index 000000000..0ed8e4d5f
--- /dev/null
+++ b/aggregator/drift_check.go
@@ -0,0 +1,67 @@
+package aggregator
+
+import (
+ "app/base/database"
+ "app/base/utils"
+)
+
+type advisoryCounts struct {
+ AdvisoryID int64
+ SystemsApplicable int
+ SystemsInstallable int
+}
+
+func checkAdvisoryDrift(rhAccountID int, advisoryIDs []int64) {
+ if len(advisoryIDs) == 0 {
+ return
+ }
+
+ var newCounts []advisoryCounts
+ err := database.DB.Table("account_advisory aa").
+ Select(`aa.advisory_id,
+ SUM(aa.systems_installable) as systems_installable,
+ SUM(aa.systems_applicable) as systems_applicable`).
+ Where("aa.rh_account_id = ? AND aa.advisory_id IN (?)", rhAccountID, advisoryIDs).
+ Group("aa.advisory_id").
+ Find(&newCounts).Error
+ if err != nil {
+ utils.LogError("err", err, "rh_account_id", rhAccountID, "drift check: failed to query account_advisory")
+ return
+ }
+
+ newCountsMap := make(map[int64]advisoryCounts, len(newCounts))
+ for _, c := range newCounts {
+ newCountsMap[c.AdvisoryID] = c
+ }
+
+ var legacyCounts []advisoryCounts
+ err = database.DB.Table("advisory_account_data").
+ Select("advisory_id, systems_applicable, systems_installable").
+ Where("rh_account_id = ? AND advisory_id IN (?)", rhAccountID, advisoryIDs).
+ Find(&legacyCounts).Error
+ if err != nil {
+ utils.LogError("err", err, "rh_account_id", rhAccountID, "drift check: failed to query advisory_account_data")
+ return
+ }
+
+ for _, legacy := range legacyCounts {
+ newVal, ok := newCountsMap[legacy.AdvisoryID]
+ if !ok {
+ utils.LogWarn("rh_account_id", rhAccountID, "advisory_id", legacy.AdvisoryID,
+ "drift check: advisory present in legacy table but missing from account_advisory")
+ continue
+ }
+ if legacy.SystemsApplicable != newVal.SystemsApplicable || legacy.SystemsInstallable != newVal.SystemsInstallable {
+ utils.LogWarn("rh_account_id", rhAccountID, "advisory_id", legacy.AdvisoryID,
+ "legacy_applicable", legacy.SystemsApplicable, "new_applicable", newVal.SystemsApplicable,
+ "legacy_installable", legacy.SystemsInstallable, "new_installable", newVal.SystemsInstallable,
+ "drift check: count mismatch between legacy and new table")
+ }
+ delete(newCountsMap, legacy.AdvisoryID)
+ }
+
+ for _, newVal := range newCountsMap {
+ utils.LogWarn("rh_account_id", rhAccountID, "advisory_id", newVal.AdvisoryID,
+ "drift check: advisory present in account_advisory but missing from legacy table")
+ }
+}
diff --git a/aggregator/drift_check_test.go b/aggregator/drift_check_test.go
new file mode 100644
index 000000000..407a1c58b
--- /dev/null
+++ b/aggregator/drift_check_test.go
@@ -0,0 +1,61 @@
+package aggregator
+
+import (
+ "app/base/core"
+ "app/base/database"
+ "app/base/models"
+ "app/base/utils"
+ "testing"
+
+ log "github.com/sirupsen/logrus"
+ "github.com/stretchr/testify/assert"
+)
+
+func TestCheckAdvisoryDriftCountMismatch(t *testing.T) {
+ utils.SkipWithoutDB(t)
+ core.SetupTestEnvironment()
+
+ assert.Nil(t, database.DB.Exec("SELECT refresh_advisory_caches(NULL, 1)").Error)
+ assert.Nil(t, database.DB.Exec("SELECT backfill_account_advisory(1)").Error)
+ defer database.DeleteAccountAdvisoryByAccount(t, 1)
+
+ assert.Nil(t, database.DB.Model(&models.AccountAdvisory{}).
+ Where("advisory_id = 1 AND rh_account_id = 1").
+ Update("systems_installable", 999).Error)
+
+ hook := utils.NewTestLogHook(log.WarnLevel)
+ log.AddHook(hook)
+
+ checkAdvisoryDrift(1, []int64{1})
+
+ found := false
+ for _, entry := range hook.LogEntries {
+ if entry.Message == "drift check: count mismatch between legacy and new table" {
+ found = true
+ break
+ }
+ }
+ assert.True(t, found, "expected count mismatch warning")
+}
+
+func TestCheckAdvisoryDriftMissingFromNew(t *testing.T) {
+ utils.SkipWithoutDB(t)
+ core.SetupTestEnvironment()
+
+ assert.Nil(t, database.DB.Exec("SELECT refresh_advisory_caches(NULL, 1)").Error)
+ defer database.DeleteAccountAdvisoryByAccount(t, 1)
+
+ hook := utils.NewTestLogHook(log.WarnLevel)
+ log.AddHook(hook)
+
+ checkAdvisoryDrift(1, []int64{1, 2})
+
+ found := false
+ for _, entry := range hook.LogEntries {
+ if entry.Message == "drift check: advisory present in legacy table but missing from account_advisory" {
+ found = true
+ break
+ }
+ }
+ assert.True(t, found, "expected missing from new table warning")
+}
diff --git a/aggregator/events.go b/aggregator/events.go
index a511d3e64..a4ff24d75 100644
--- a/aggregator/events.go
+++ b/aggregator/events.go
@@ -1,20 +1,103 @@
package aggregator
import (
+ "app/base/database"
"app/base/mqueue"
"app/base/utils"
+ "sync"
+ "time"
"github.com/bytedance/sonic"
+ "github.com/lib/pq"
)
-// TODO: stub - will process advisory update events in batches and update account_advisory table
+var (
+ batchSize int
+ flushTimeout time.Duration
+ advisoryBuffer []mqueue.AdvisoryUpdateEvent
+ bufferLock sync.Mutex
+ flushTimer *time.Timer
+)
+
+func initBuffer() {
+ advisoryBuffer = make([]mqueue.AdvisoryUpdateEvent, 0, batchSize+1)
+ flushTimer = time.AfterFunc(87600*time.Hour, func() {
+ utils.LogInfo("flushing advisory buffer after timeout")
+ flushAdvisoryBuffer()
+ })
+}
+
+func flushAdvisoryBuffer() {
+ bufferLock.Lock()
+ if len(advisoryBuffer) == 0 {
+ bufferLock.Unlock()
+ return
+ }
+ // Copy and unlock before processing so new events can accumulate while DB work runs
+ batch := make([]mqueue.AdvisoryUpdateEvent, len(advisoryBuffer))
+ copy(batch, advisoryBuffer)
+ advisoryBuffer = advisoryBuffer[:0]
+ bufferLock.Unlock()
+
+ grouped := groupAdvisoryUpdates(batch)
+ processAdvisoryBatch(grouped)
+}
+
func advisoryUpdateHandler(m mqueue.KafkaMessage) error {
var event mqueue.AdvisoryUpdateEvent
if err := sonic.Unmarshal(m.Value, &event); err != nil {
- utils.LogError("err", err, "Could not deserialize advisory update event")
+ utils.LogError("err", err, "could not deserialize advisory update event")
return nil
}
- // TODO: advisory update code goes here
- _ = event
+
+ bufferLock.Lock()
+ advisoryBuffer = append(advisoryBuffer, event)
+ flushTimer.Reset(flushTimeout)
+ shouldFlush := len(advisoryBuffer) >= batchSize
+ bufferLock.Unlock()
+
+ if shouldFlush {
+ utils.LogInfo("flushing full advisory buffer")
+ flushAdvisoryBuffer()
+ }
return nil
}
+
+func groupAdvisoryUpdates(events []mqueue.AdvisoryUpdateEvent) map[int][]int64 {
+ sets := make(map[int]map[int64]struct{})
+ for _, e := range events {
+ if _, ok := sets[e.RhAccountID]; !ok {
+ sets[e.RhAccountID] = make(map[int64]struct{})
+ }
+ for _, id := range e.AdvisoryIDs {
+ sets[e.RhAccountID][id] = struct{}{}
+ }
+ }
+
+ grouped := make(map[int][]int64, len(sets))
+ for accID, idSet := range sets {
+ ids := make([]int64, 0, len(idSet))
+ for id := range idSet {
+ ids = append(ids, id)
+ }
+ grouped[accID] = ids
+ }
+ return grouped
+}
+
+func processAdvisoryBatch(grouped map[int][]int64) {
+ for rhAccountID, advisoryIDs := range grouped {
+ utils.LogInfo("rh_account_id", rhAccountID, "advisory_count", len(advisoryIDs), "refreshing account advisory caches")
+ err := database.DB.Exec("SELECT refresh_account_advisory_caches_multi(?, ?)", pq.Array(advisoryIDs), rhAccountID).Error //nolint:lll
+ if err != nil {
+ utils.LogError("err", err, "rh_account_id", rhAccountID, "failed to refresh account advisory caches")
+ continue
+ }
+
+ 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/aggregator/events_test.go b/aggregator/events_test.go
new file mode 100644
index 000000000..692c98fa0
--- /dev/null
+++ b/aggregator/events_test.go
@@ -0,0 +1,93 @@
+package aggregator
+
+import (
+ "app/base/core"
+ "app/base/database"
+ "app/base/mqueue"
+ "app/base/utils"
+ "sort"
+ "testing"
+ "time"
+
+ "github.com/bytedance/sonic"
+ "github.com/stretchr/testify/assert"
+)
+
+func toKafkaMessage(t *testing.T, event mqueue.AdvisoryUpdateEvent) mqueue.KafkaMessage {
+ t.Helper()
+ data, err := sonic.Marshal(event)
+ assert.Nil(t, err)
+ return mqueue.KafkaMessage{Value: data}
+}
+
+func TestGroupAdvisoryUpdatesSingleAccount(t *testing.T) {
+ events := []mqueue.AdvisoryUpdateEvent{
+ {RhAccountID: 1, AdvisoryIDs: []int64{1, 2}},
+ {RhAccountID: 1, AdvisoryIDs: []int64{2, 3}},
+ }
+ grouped := groupAdvisoryUpdates(events)
+
+ assert.Len(t, grouped, 1)
+ ids := grouped[1]
+ sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
+ assert.Equal(t, []int64{1, 2, 3}, ids)
+}
+
+func TestGroupAdvisoryUpdatesMultipleAccounts(t *testing.T) {
+ events := []mqueue.AdvisoryUpdateEvent{
+ {RhAccountID: 1, AdvisoryIDs: []int64{1, 2}},
+ {RhAccountID: 2, AdvisoryIDs: []int64{5}},
+ {RhAccountID: 1, AdvisoryIDs: []int64{3}},
+ }
+ grouped := groupAdvisoryUpdates(events)
+
+ assert.Len(t, grouped, 2)
+
+ ids1 := grouped[1]
+ sort.Slice(ids1, func(i, j int) bool { return ids1[i] < ids1[j] })
+ assert.Equal(t, []int64{1, 2, 3}, ids1)
+ assert.Equal(t, []int64{int64(5)}, grouped[2])
+}
+
+func TestGroupAdvisoryUpdatesDeduplicates(t *testing.T) {
+ events := []mqueue.AdvisoryUpdateEvent{
+ {RhAccountID: 1, AdvisoryIDs: []int64{1, 1, 2}},
+ {RhAccountID: 1, AdvisoryIDs: []int64{2, 2, 1}},
+ }
+ grouped := groupAdvisoryUpdates(events)
+
+ ids := grouped[1]
+ sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
+ assert.Equal(t, []int64{1, 2}, ids)
+}
+
+func TestBufferedEventsProcessedOnBatchThreshold(t *testing.T) {
+ utils.SkipWithoutDB(t)
+ core.SetupTestEnvironment()
+
+ assert.Nil(t, database.DB.Exec("SELECT refresh_advisory_caches(NULL, 1)").Error)
+ defer database.DeleteAccountAdvisoryByAccount(t, 1)
+
+ batchSize = 3
+ flushTimeout = time.Hour
+ initBuffer()
+
+ msg := toKafkaMessage(t, mqueue.AdvisoryUpdateEvent{RhAccountID: 1, AdvisoryIDs: []int64{1, 2}})
+
+ // First two events accumulate in the buffer
+ assert.Nil(t, advisoryUpdateHandler(msg))
+ assert.Equal(t, 1, len(advisoryBuffer))
+ assert.Nil(t, advisoryUpdateHandler(msg))
+ assert.Equal(t, 2, len(advisoryBuffer))
+
+ // Third event triggers flush and processAdvisoryBatch runs
+ assert.Nil(t, advisoryUpdateHandler(msg))
+ assert.Equal(t, 0, len(advisoryBuffer))
+
+ // Verify account_advisory was populated
+ var count int64
+ assert.Nil(t, database.DB.Table("account_advisory").
+ Where("rh_account_id = 1 AND advisory_id IN (?)", []int64{1, 2}).
+ Count(&count).Error)
+ assert.True(t, count > 0, "account_advisory rows should be created after flush")
+}
diff --git a/aggregator/notifications.go b/aggregator/notifications.go
index bc48d0fae..759383d07 100644
--- a/aggregator/notifications.go
+++ b/aggregator/notifications.go
@@ -1,8 +1,14 @@
package aggregator
import (
+ "app/base"
+ "app/base/database"
"app/base/mqueue"
+ ntf "app/base/notification"
"app/base/utils"
+ "time"
+
+ "gorm.io/gorm"
)
var notificationsPublisher mqueue.Writer
@@ -12,3 +18,85 @@ func configureNotifications() {
notificationsPublisher = mqueue.NewKafkaWriterFromEnv(topic)
}
}
+
+func getUnnotifiedAdvisories(tx *gorm.DB, rhAccountID int, advisoryIDs []int64) ([]ntf.Advisory, error) {
+ var advisories []ntf.Advisory
+ err := tx.Table("account_advisory aa").
+ Select("DISTINCT am.id as advisory_id, am.name as advisory_name, at.name as advisory_type, am.synopsis").
+ Joins("INNER JOIN advisory_metadata am ON am.id = aa.advisory_id").
+ Joins("INNER JOIN advisory_type at ON at.id = am.advisory_type_id").
+ Where("aa.rh_account_id = ? AND aa.advisory_id IN (?) AND aa.notified IS NULL AND aa.systems_installable > 0",
+ rhAccountID, advisoryIDs).
+ Order("am.name ASC").
+ Scan(&advisories).Error
+ return advisories, err
+}
+
+func publishNewAdvisoryNotification(rhAccountID int, advisoryIDs []int64) error {
+ if notificationsPublisher == nil || !enableNotifications {
+ return nil
+ }
+
+ tx := database.DB.WithContext(base.Context).Begin()
+ defer tx.Rollback() //nolint:errcheck
+
+ advisories, err := getUnnotifiedAdvisories(tx, rhAccountID, advisoryIDs)
+ if err != nil {
+ return err
+ }
+ if len(advisories) == 0 {
+ return nil
+ }
+
+ var orgID string
+ err = tx.Table("rh_account").Select("org_id").Where("id = ?", rhAccountID).Scan(&orgID).Error
+ if err != nil {
+ return err
+ }
+
+ events := make([]ntf.Event, 0, len(advisories))
+ for _, advisory := range advisories {
+ events = append(events, ntf.Event{Payload: advisory, Metadata: ntf.Metadata{}})
+ }
+
+ notif, err := ntf.MakeAccountNotification(orgID, ntf.NewAdvisoryEvent, events)
+ if err != nil {
+ return err
+ }
+
+ msg, err := mqueue.MessageFromJSON(orgID, notif, nil)
+ if err != nil {
+ return err
+ }
+
+ err = notificationsPublisher.WriteMessages(base.Context, msg)
+ if err != nil {
+ return err
+ }
+
+ notifiedIDs := make([]int64, 0, len(advisories))
+ for _, a := range advisories {
+ notifiedIDs = append(notifiedIDs, a.AdvisoryID)
+ }
+
+ err = markAdvisoriesNotified(tx, rhAccountID, notifiedIDs)
+ if err != nil {
+ return err
+ }
+
+ err = tx.Commit().Error
+ if err != nil {
+ return err
+ }
+
+ utils.LogInfo("rh_account_id", rhAccountID, "org_id", orgID, "advisory_count", len(advisories),
+ "new advisory notification sent")
+
+ return nil
+}
+
+func markAdvisoriesNotified(tx *gorm.DB, rhAccountID int, advisoryIDs []int64) error {
+ return tx.Table("account_advisory").
+ Where("rh_account_id = ? AND advisory_id IN (?)", rhAccountID, advisoryIDs).
+ Update("notified", time.Now()).Error
+}
diff --git a/aggregator/notifications_test.go b/aggregator/notifications_test.go
new file mode 100644
index 000000000..435758c02
--- /dev/null
+++ b/aggregator/notifications_test.go
@@ -0,0 +1,79 @@
+package aggregator
+
+import (
+ "app/base/core"
+ "app/base/database"
+ "app/base/mqueue"
+ ntf "app/base/notification"
+ "app/base/utils"
+ "testing"
+
+ "github.com/bytedance/sonic"
+ "github.com/stretchr/testify/assert"
+)
+
+func TestPublishNewAdvisoryNotificationSkipsAlreadyNotified(t *testing.T) {
+ utils.SkipWithoutDB(t)
+ core.SetupTestEnvironment()
+
+ mockWriter := mqueue.MockKafkaWriter{}
+ notificationsPublisher = &mockWriter
+ enableNotifications = true
+ defer func() {
+ enableNotifications = false
+ notificationsPublisher = nil
+ }()
+
+ testWorkspaceID := "00000000-0000-0000-0000-000000000001"
+ advisoryIDs := []int64{1, 2}
+
+ database.CreateAccountAdvisory(t, 1, testWorkspaceID, advisoryIDs, 1)
+ // Mark them as already notified
+ assert.Nil(t, database.DB.Table("account_advisory").
+ Where("rh_account_id = 1 AND advisory_id IN (?)", advisoryIDs).
+ Update("notified", "2026-01-01").Error)
+ defer database.DeleteAccountAdvisoryByAccount(t, 1)
+
+ err := publishNewAdvisoryNotification(1, advisoryIDs)
+ assert.NoError(t, err)
+ assert.Empty(t, mockWriter.Messages)
+}
+
+func TestPublishNewAdvisoryNotificationSuccess(t *testing.T) {
+ utils.SkipWithoutDB(t)
+ core.SetupTestEnvironment()
+
+ mockWriter := mqueue.MockKafkaWriter{}
+ notificationsPublisher = &mockWriter
+ enableNotifications = true
+ defer func() {
+ enableNotifications = false
+ notificationsPublisher = nil
+ }()
+
+ // Backfill to populate account_advisory from system_advisories
+ assert.Nil(t, database.DB.Exec("SELECT backfill_account_advisory(1)").Error)
+ defer database.DeleteAccountAdvisoryByAccount(t, 1)
+
+ // Advisory IDs 1-8 exist for rh_account_id=1 in test data
+ advisoryIDs := []int64{1, 2}
+
+ err := publishNewAdvisoryNotification(1, advisoryIDs)
+ assert.NoError(t, err)
+
+ assert.Equal(t, 1, len(mockWriter.Messages))
+ assert.Equal(t, "org_1", string(mockWriter.Messages[0].Key))
+
+ var notif ntf.Notification
+ assert.Nil(t, sonic.Unmarshal(mockWriter.Messages[0].Value, ¬if))
+ assert.Equal(t, "org_1", notif.OrgID)
+ assert.Nil(t, notif.Context)
+ assert.NotEmpty(t, notif.Events)
+
+ // Verify advisories were marked as notified (count varies by workspace)
+ var count int64
+ assert.Nil(t, database.DB.Table("account_advisory").
+ Where("rh_account_id = 1 AND advisory_id IN (?) AND notified IS NOT NULL", advisoryIDs).
+ Count(&count).Error)
+ assert.True(t, count > 0)
+}
diff --git a/base/notification/notification.go b/base/notification/notification.go
index 1c9ccf34c..b45f5d4ff 100644
--- a/base/notification/notification.go
+++ b/base/notification/notification.go
@@ -11,13 +11,13 @@ import (
)
const (
- Version = "v1.1.0"
- Bundle = "rhel"
- Application = "patch"
+ Version = "v1.1.0"
+ Bundle = "rhel"
+ Application = "patch"
+ NewAdvisoryEvent = "new-advisory"
)
-// TODO: Remove Context after migrating to the new aggregator component
-// Advisories apply to multiple systems, so for aggregated notifications, system-specific context is unnecessary
+// TODO: Remove Context, MakeNotification and *Context field on Notification after fully migrating to the aggregator
// See: https://redhat.atlassian.net/browse/RHINENG-26543
type Context struct {
@@ -64,8 +64,8 @@ type Notification struct {
// ISO-8601 formatted date (per platform convention when the message was sent).
Timestamp string `json:"timestamp"`
// Extra information that are common to all the events that are sent in this message.
- Context Context `json:"context,omitempty"`
- Events []Event `json:"events"`
+ Context *Context `json:"context,omitempty"`
+ Events []Event `json:"events"`
// Recipients settings - Applications can add extra email recipients by adding entries to this array.
// This setting extends whatever the Administrators configured in their Notifications settings (since v1.1.0).
Recipients []Recipient `json:"recipients,omitempty"`
@@ -101,7 +101,7 @@ func MakeNotification(inv *models.SystemInventory, systemTags []SystemTag, orgID
EventType: eventType,
// ISO-8601 formatted time
Timestamp: time.Now().Format(time.RFC3339),
- Context: Context{
+ Context: &Context{
InventoryID: inv.InventoryID,
DisplayName: inv.DisplayName,
HostURL: hostURL,
@@ -111,3 +111,19 @@ func MakeNotification(inv *models.SystemInventory, systemTags []SystemTag, orgID
OrgID: orgID,
}, nil
}
+
+func MakeAccountNotification(orgID string, eventType string, events []Event) (*Notification, error) {
+ if orgID == "" || orgID == "null" {
+ return nil, errors.New("invalid orgID")
+ }
+
+ return &Notification{
+ Version: Version,
+ Bundle: Bundle,
+ Application: Application,
+ EventType: eventType,
+ Timestamp: time.Now().Format(time.RFC3339),
+ Events: events,
+ OrgID: orgID,
+ }, nil
+}
diff --git a/docs/md/architecture.md b/docs/md/architecture.md
index 1be0a15cc..6ed5af4b1 100644
--- a/docs/md/architecture.md
+++ b/docs/md/architecture.md
@@ -2,7 +2,7 @@
The project is written as a set of communicating containers. It allows to scale different parts of the application
according to needs. It also increases application robustness, because even if one component have some issues (errors, down times), the others are not impacted with that work.
-The components are `manager`, `listener`, `evaluator-{upload,recalc}`, `vmaas_sync`, `database` and
+The components are `manager`, `listener`, `evaluator-{upload,recalc}`, `aggregator`, `vmaas_sync`, `database` and
`database_admin`.
### Components
@@ -43,6 +43,15 @@ resolves template-assigned systems before sending. The requests are separated
as when there is a heavy load from inventory at the time it may take very long for systems to be recalculated/updated.
See [component environment variables](../../conf/evaluator_user_evaluation.env)
+- **aggregator** - maintains per-account, per-workspace advisory counts. When the evaluator processes a system upload or
+recalculation and updates **`system_advisories`**, it publishes an `AdvisoryUpdateEvent` to the `patchman.advisory.update`
+Kafka topic listing which advisory IDs changed for a given account. The aggregator consumes these events and recounts
+how many systems have each advisory applicable or installable, writing the results to **`account_advisory`**. This is the
+workspace-aware replacement for **`advisory_account_data`** (previously maintained by the evaluator). Incoming events are
+batched before processing. When `enable_notifications` is set in `POD_CONFIG`, the aggregator also publishes
+new installable advisories to `platform.notifications.ingress` and marks them as notified in **`account_advisory`**.
+See [component environment variables](../../conf/aggregator.env)
+
- **vmaas-sync** - connects to [VMaaS](https://github.com/RedHatInsights/vmaas), and upon receiving notification about
updated data, syncs new advisories into the database, and requests re-evaluation for systems which could be affected by
new advisories. It's done via messaging to the `patchman.evaluator.recalc` Kafka topic and receiving by the
diff --git a/docs/md/graphics/schema.dot b/docs/md/graphics/schema.dot
index 154267440..47768ff3a 100644
--- a/docs/md/graphics/schema.dot
+++ b/docs/md/graphics/schema.dot
@@ -1,5 +1,6 @@
// dot schema.dot -Tpng -o schema.png -Gdpi=100
digraph G {
+ rankdir="LR"
node[shape="Mrecord"
style="filled"]
@@ -20,6 +21,12 @@ digraph G {
evaluator_recalc -> db [label="psql:5432", dir="both"]
evaluator_recalc -> remediations [label="kafka:9092", style="dotted,bold"]
+ evaluator_upload -> aggregator [label="kafka:9092", style="dotted,bold"]
+ aggregator -> db [label="psql:5432", dir="both"]
+ aggregator -> notifications [label="kafka:9092", style="dotted,bold"]
+
+ {rank=same; evaluator_upload; aggregator}
+
vmaas -> vmaas_sync [label="http:8080", style="dashed"]
vmaas -> vmaas_sync [label="ws:8082", style="dashed"]
vmaas_sync -> db [label="psql:5432", dir="both"]
@@ -70,6 +77,13 @@ digraph G {
>
fillcolor="lightblue"]
+ aggregator [label=<
+
+  |
+ | aggregator |
+
>
+ fillcolor="lightblue"]
+
db_admin [label=<
|
@@ -114,5 +128,11 @@ digraph G {
>
fillcolor="gray92"]
+ notifications [label=<
+ >
+ fillcolor="gray92"]
+
client [label="HTTP Client", fillcolor="gray92"]
}
diff --git a/docs/md/graphics/schema.png b/docs/md/graphics/schema.png
index fb1a3cc55..6818971d6 100644
Binary files a/docs/md/graphics/schema.png and b/docs/md/graphics/schema.png differ
diff --git a/evaluator/notifications.go b/evaluator/notifications.go
index 0724034b6..962cdddc3 100644
--- a/evaluator/notifications.go
+++ b/evaluator/notifications.go
@@ -13,8 +13,6 @@ import (
"gorm.io/gorm"
)
-const NewAdvisoryEvent = "new-advisory"
-
var notificationsPublisher mqueue.Writer
func configureNotifications() {
@@ -95,7 +93,7 @@ func publishNewAdvisoriesNotification(tx *gorm.DB, system *models.SystemPlatform
return errors.Wrap(err, "getting system tags failed")
}
- notif, err := ntf.MakeNotification(&system.Inventory, tags, orgID, NewAdvisoryEvent, events)
+ notif, err := ntf.MakeNotification(&system.Inventory, tags, orgID, ntf.NewAdvisoryEvent, events)
if err != nil {
return errors.Wrap(err, "creating notification failed")
}
diff --git a/evaluator/notifications_test.go b/evaluator/notifications_test.go
index 8cc4071d2..5f0bdc8d7 100644
--- a/evaluator/notifications_test.go
+++ b/evaluator/notifications_test.go
@@ -106,7 +106,7 @@ func TestAdvisoriesNotificationMessage(t *testing.T) {
orgID := "1234567"
url := fmt.Sprintf("https://localhost/insights/inventory/%s", testInventoryID.String())
- notification, err := ntf.MakeNotification(inv, tags, orgID, NewAdvisoryEvent, events)
+ notification, err := ntf.MakeNotification(inv, tags, orgID, ntf.NewAdvisoryEvent, events)
assert.Nil(t, err)
assert.Equal(t, orgID, notification.OrgID)
assert.Equal(t, url, notification.Context.HostURL)