Introduce RetryHeap and RetryHeapProcessor #1972

agarakan · 2025-12-30T16:52:02Z

Description of the issue

Introduce RetryHeap and RetryHeapProcessor implementations for cloudwatchlogs log publishing

Description of changes

RetryHeap is implemented as a min-heap sorted on retryTimestamp for a given batch PutLogEvent request that has already been attempted once and failed.

Implements a min heap on retryTimestamp
depends on go standard library container/heap
Implements Push(), PopReady(), Size() and Stop()
Implementation includes a semaphor to track heap size and allow blocking mechanism when retryHeap hits capacity.
Implementation includes a mutex so pushing to heap does not encounter race condition

RetryHeapProcessor will periodically scan for batches that are ready to be retried based on retryTimestamp and move them to the SenderPool tasks queue for processing by the sender pool.

Scans retry heap periodically (every 100 ms) for ready to retry batches
Maintains reference to the SenderPool to put ready batches back on the senderpool tasks queue

Assumptions: The RetryHeap and RetryHeapProcessor will only be instantiated and used when concurrency is enabled.

No functionality changes are introduced in this PR as this sets up RetryHeap and related components, but does not instantiate them in the code yet.

License

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

Tests

New unit tests are passing

make lint
make fmt
make fmt-sh
make test

Requirements

Before commiting your code, please do the following steps.

Run make fmt and make fmt-sh
Run make lint

Integration Tests

To run integration tests against this PR, add the ready for testing label.

sky333999 · 2025-12-31T20:30:52Z

plugins/outputs/cloudwatchlogs/internal/pusher/retryheap.go

+)
+
+// retryHeapImpl implements heap.Interface for logEventBatch sorted by nextRetryTime
+type retryHeapImpl []*logEventBatch


nit: Add var _ heap.Interface = (*retryHeapImpl)(nil)

Ack will update as follow up

sky333999 · 2025-12-31T20:31:08Z

plugins/outputs/cloudwatchlogs/internal/pusher/retryheap.go

+}
+
+// RetryHeap manages failed batches during their retry wait periods
+type RetryHeap interface {


nit: Add var _ RetryHeap = (*retryHeap)(nil)

Ack will update as follow up

sky333999 · 2025-12-31T21:30:25Z

plugins/outputs/cloudwatchlogs/internal/pusher/retryheap.go

+	for _, batch := range readyBatches {
+		// Check if batch has expired
+		if batch.isExpired(p.maxRetryDuration) {
+			p.logger.Debugf("Dropping expired batch for %s/%s", batch.Group, batch.Stream)


nit: Make it match https://github.com/aws/amazon-cloudwatch-agent/pull/1970/changes#diff-d38a5f3264bf20ef00aa1acd29a08512a7d92c63384dcacfd3737b70cdf7311cR119 for consistency with our logging.

Ack will update as follow up

jefchien · 2026-01-07T21:08:13Z

plugins/outputs/cloudwatchlogs/internal/pusher/retryheap.go

+)
+
+// retryHeapImpl implements heap.Interface for logEventBatch sorted by nextRetryTime
+type retryHeapImpl []*logEventBatch


nit: Why do we have a retryHeapImpl and a retryHeap? Is there ever a case where we'd use one without the other? Is it better to operate on the []*logEventBatch directly instead of accessing it as a field of retryHeap?

jefchien · 2026-01-07T21:13:40Z

plugins/outputs/cloudwatchlogs/internal/pusher/retryheap.go

+func (h *retryHeapImpl) Pop() interface{} {
+	old := *h
+	n := len(old)
+	item := old[n-1]
+	*h = old[0 : n-1]
+	return item
+}


Seems like this is missing a step from the PriorityQueue example

https://pkg.go.dev/container/heap#Interface

func (pq *PriorityQueue) Pop() any { ... old[n-1] = nil // don't stop the GC from reclaiming the item eventually ... }

Is there a reason we decided not to include it?

jefchien · 2026-01-08T16:35:50Z

plugins/outputs/cloudwatchlogs/internal/pusher/retryheap_test.go

+		batch3 := newLogEventBatch(target, nil)
+		batch3.nextRetryTime = time.Now().Add(time.Hour) // Future time, won't be popped
+		heap.Push(batch3)                                // This should block
+		pushCompleted = true


This seems like a data race since you're accessing the same variable on two separate threads/goroutines. Consider using an atomic or a channel.

jefchien · 2026-01-08T16:39:29Z

plugins/outputs/cloudwatchlogs/internal/pusher/retryheap.go

+}
+
+// RetryHeap manages failed batches during their retry wait periods
+type RetryHeap interface {


So is the idea that each sender will have a reference to the RetryHeap to push failed batches onto and then there's a single RetryHeapProcessor that also has a reference to the RetryHeap?

jefchien · 2026-01-08T16:48:31Z

plugins/outputs/cloudwatchlogs/internal/pusher/retryheap.go

+// NewRetryHeap creates a new retry heap with the specified maximum size
+func NewRetryHeap(maxSize int) RetryHeap {
+	rh := &retryHeap{
+		heap:      make(retryHeapImpl, 0),


nit: Would it make sense to initialize it with a capacity of maxSize? That way it doesn't need to reallocate on each push?

jefchien · 2026-01-08T16:56:07Z

plugins/outputs/cloudwatchlogs/internal/pusher/retryheap.go

+}
+
+// Stop stops the retry heap
+func (rh *retryHeap) Stop() {


What is going to stop the retry heap? Does it stop when the RetryHeapProcessor stops?

jefchien · 2026-01-08T16:58:49Z

plugins/outputs/cloudwatchlogs/internal/pusher/retryheap_test.go

+	}
+
+	// Wait for batches to become ready, then pop to release semaphore
+	time.Sleep(4 * time.Second)


I know we use sleeps in other unit tests, but is there a way to make this less flaky? Is there a reason to wait since this isn't using the processor, so doesn't rely on the ticker? The initial batches could be expired when added like in the other tests.

jefchien · 2026-01-08T16:59:20Z

plugins/outputs/cloudwatchlogs/internal/pusher/retryheap_test.go

+
+	// Wait for batches to become ready, then pop to release semaphore
+	time.Sleep(4 * time.Second)
+	heap.PopReady()


Can we validate that the correct batches were taken off the heap?

Introduce retryHeap and retryHeapProcessor

66186a7

agarakan requested a review from a team as a code owner December 30, 2025 16:52

agarakan added the ready for testing Indicates this PR is ready for integration tests to run label Dec 30, 2025

Exchange pushch for semaphor to enformce heap size and blocking

83224b4

agarakan changed the base branch from enable-multithreaded-logging-by-default to poison-pill December 30, 2025 18:08

sky333999 reviewed Dec 31, 2025

View reviewed changes

jefchien reviewed Jan 8, 2026

View reviewed changes

Introduce RetryHeap and RetryHeapProcessor #1972

Are you sure you want to change the base?

Introduce RetryHeap and RetryHeapProcessor #1972

Uh oh!

Conversation

agarakan commented Dec 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description of the issue

Description of changes

License

Tests

Requirements

Integration Tests

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

agarakan commented Dec 30, 2025 •

edited

Loading