feat: add aggregator consumer - #2284
Merged
katarinazaprazna merged 7 commits intoJul 30, 2026
Merged
Conversation
Added `checkAdvisoryDrift` to compare advisory data between new and legacy tables. Logs discrepancies for missing or mismatched records.
The aggregator sends account-level notifications without system-specific context. Make Context a pointer so it can be omitted, add MakeAccountNotification constructor for this use case, and move NewAdvisoryEvent to a shared constant.
Reviewer's GuideImplements an advisory update aggregator that buffers Kafka events, groups them per account, refreshes account-level advisory caches, performs drift checks against legacy data, and emits account-scoped notifications using an updated, more flexible notification API. File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
katarinazaprazna
changed the title
katarinazaprazna
marked this pull request as ready for review
July 29, 2026 21:30
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 3 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="aggregator/aggregator.go" line_range="23-24" />
<code_context>
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
}
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Guard against non-positive batch sizes from configuration to avoid degenerate buffering behavior
If `advisory_batch_size` is 0 or negative, `cap(batchSize+1)` and `len(advisoryBuffer) >= batchSize` will make `shouldFlush` true immediately/on every append, effectively disabling batching. Consider clamping `batchSize` to at least 1 (or another small default) after reading the config to avoid this misconfiguration causing unexpected behavior.
</issue_to_address>
### Comment 2
<location path="aggregator/events_test.go" line_range="64-73" />
<code_context>
+func TestBufferedEventsProcessedOnBatchThreshold(t *testing.T) {
</code_context>
<issue_to_address>
**suggestion (testing):** Avoid leaking global buffer state and timers across tests by resetting them and stopping the timer
This test depends on package-level globals (`advisoryBuffer`, `batchSize`, `flushTimeout`, `flushTimer`), which are shared across all tests and can cause cross-test interference (e.g., a long-lived `flushTimer` firing in other tests or `batchSize` remaining at 3). Please add reset/cleanup logic (for example via `t.Cleanup`) to stop the timer and restore these globals to their defaults, or centralize this in a small helper that reinitializes the buffer and timer for each test.
</issue_to_address>
### Comment 3
<location path="aggregator/events.go" line_range="14" />
<code_context>
)
-// TODO: stub - will process advisory update events in batches and update account_advisory table
+var (
+ batchSize int
+ flushTimeout time.Duration
</code_context>
<issue_to_address>
**issue (complexity):** Consider encapsulating the buffering and timer logic into an advisoryBatcher type to avoid package-level mutable state and simplify the handler behaviorally unchanged.
Encapsulating the buffering/timer logic into a type will reduce global cross-cutting state, make initialization explicit, and keep advisoryUpdateHandler simpler without changing behavior.
You can wrap the current globals into a small `advisoryBatcher` and wire it from whatever config/bootstraps the handler:
```go
type advisoryBatcher struct {
batchSize int
flushTimeout time.Duration
mu sync.Mutex
buffer []mqueue.AdvisoryUpdateEvent
timer *time.Timer
}
func newAdvisoryBatcher(batchSize int, flushTimeout time.Duration) *advisoryBatcher {
b := &advisoryBatcher{
batchSize: batchSize,
flushTimeout: flushTimeout,
buffer: make([]mqueue.AdvisoryUpdateEvent, 0, batchSize+1),
}
b.timer = time.AfterFunc(87600*time.Hour, func() {
utils.LogInfo("flushing advisory buffer after timeout")
b.flush()
})
return b
}
```
Then move buffer/timer operations into methods, keeping lock usage local and avoiding globals:
```go
func (b *advisoryBatcher) handle(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")
return nil
}
b.mu.Lock()
b.buffer = append(b.buffer, event)
b.timer.Reset(b.flushTimeout)
shouldFlush := len(b.buffer) >= b.batchSize
b.mu.Unlock()
if shouldFlush {
utils.LogInfo("flushing full advisory buffer")
b.flush()
}
return nil
}
func (b *advisoryBatcher) flush() {
b.mu.Lock()
if len(b.buffer) == 0 {
b.mu.Unlock()
return
}
batch := make([]mqueue.AdvisoryUpdateEvent, len(b.buffer))
copy(batch, b.buffer)
b.buffer = b.buffer[:0]
b.mu.Unlock()
grouped := groupAdvisoryUpdates(batch)
processAdvisoryBatch(grouped)
}
```
And the handler wiring becomes an explicit dependency instead of relying on `initBuffer` + globals:
```go
var advisoryBatcherInstance = newAdvisoryBatcher(batchSize, flushTimeout)
func advisoryUpdateHandler(m mqueue.KafkaMessage) error {
return advisoryBatcherInstance.handle(m)
}
```
This keeps all current behavior (batching, timeout-based flush, grouping, DB work) but:
- Removes package-level mutable state.
- Makes initialization explicit and tied to handler setup.
- Localizes locking and timer interaction inside one type, reducing concurrency complexity.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
katarinazaprazna
force-pushed
the
add-aggregator-consumer
branch
from
July 29, 2026 21:49
790b18c to
8687006
Compare
Batch incoming AdvisoryUpdateEvent messages by rh_account_id, deduplicate advisory IDs, and update the account_advisory table. After each batch, run the drift check against the legacy table and publish new-advisory notifications (gated behind enable_notifications flag).
katarinazaprazna
force-pushed
the
add-aggregator-consumer
branch
from
July 29, 2026 22:04
8687006 to
7abc4d9
Compare
MichaelMraka
left a comment
MichaelMraka
left a comment
Collaborator
There was a problem hiding this comment.
PR looks fine, please check failing unit test.
Aggregator tests delete account_advisory rows for rh_account_id=1 during cleanup. When manager reads from DB, it is empty, causing TestAdvisoriesExportJSON to fail. Add TestMain to aggregator to repopulate account_advisory after all tests complete.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #2284 +/- ##
==========================================
+ Coverage 59.10% 59.27% +0.16%
==========================================
Files 147 148 +1
Lines 9332 9488 +156
==========================================
+ Hits 5516 5624 +108
- Misses 3244 3276 +32
- Partials 572 588 +16
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
MichaelMraka
approved these changes
Jul 30, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
patchman.advisory.updateby account, refresh account_advisory caches, send notificationsFollow-up
Secure Coding Practices Checklist GitHub Link
Secure Coding Checklist
Summary by Sourcery
Introduce an advisory update aggregator that batches events per account to refresh advisory caches, perform drift checks against legacy data, and emit account-level notifications.
New Features:
Enhancements:
Tests:
Summary by Sourcery
Introduce an advisory update aggregator that buffers and batches advisory update events per account to refresh caches, perform drift checks, and emit account-level notifications.
New Features:
Enhancements:
Tests: