Remove configurable maxRetryTimeout in favor of default hardcoded value

Author: agarakan

plugins/outputs/cloudwatchlogs/cloudwatchlogs.go

@@ -37,8 +37,6 @@ const (
 	LogEntryField     = "value"
 
 	defaultFlushTimeout = 5 * time.Second
-
-	maxRetryTimeout = 14*24*time.Hour + 10*time.Minute
 )
 
 var (
@@ -166,7 +164,7 @@ func (c *CloudWatchLogs) getDest(t pusher.Target, logSrc logs.LogSrc) *cwDest {
 		}
 		c.targetManager = pusher.NewTargetManager(c.Log, client)
 	})
-	p := pusher.NewPusher(c.Log, t, client, c.targetManager, logSrc, c.workerPool, c.ForceFlushInterval.Duration, maxRetryTimeout, &c.pusherWaitGroup, c.Concurrency, c.retryHeap)
+	p := pusher.NewPusher(c.Log, t, client, c.targetManager, logSrc, c.workerPool, c.ForceFlushInterval.Duration, &c.pusherWaitGroup, c.Concurrency, c.retryHeap)
 	cwd := &cwDest{
 		pusher:   p,
 		retryer:  logThrottleRetryer,

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

@@ -15,9 +15,12 @@ import (
 	"github.com/aws/amazon-cloudwatch-agent/sdk/service/cloudwatchlogs"
 )
 
-// CloudWatch Logs PutLogEvents API limits
-// Taken from https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_PutLogEvents.html
 const (
+	// maxRetryTimeout is the default retry timeout for CloudWatch Logs operations
+	maxRetryTimeout = 14*24*time.Hour + 10*time.Minute
+
+	// CloudWatch Logs PutLogEvents API limits
+	// Taken from https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_PutLogEvents.html
 	// The maximum batch size in bytes. This size is calculated as the sum of all event messages in UTF-8,
 	// plus 26 bytes for each log event.
 	reqSizeLimit = 1024 * 1024
@@ -253,11 +256,11 @@ func (t byTimestamp) Less(i, j int) bool {
 }
 
 // initializeStartTime sets the start time and expiration time if not already set.
-func (b *logEventBatch) initializeStartTime(retryDuration time.Duration) {
+func (b *logEventBatch) initializeStartTime() {
 	if b.startTime.IsZero() {
 		b.startTime = time.Now()
 	}
-	b.expireAfter = b.startTime.Add(retryDuration)
+	b.expireAfter = b.startTime.Add(maxRetryTimeout)
 }
 
 // updateRetryMetadata updates the retry metadata after a failed send attempt.

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

@@ -414,7 +414,7 @@ func TestBatchRetryMetadata(t *testing.T) {
 	assert.False(t, batch.isExpired())
 
 	// Test initializeStartTime
-	batch.initializeStartTime(time.Hour)
+	batch.initializeStartTime()
 	assert.False(t, batch.startTime.IsZero())
 
 	// Test updateRetryMetadata

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

@@ -6,7 +6,6 @@ package pusher
 import (
 	"sync"
 	"sync/atomic"
-	"time"
 )
 
 type WorkerPool interface {
@@ -113,13 +112,3 @@ func (s *senderPool) Stop() {
 	// workerpool is stopped by the plugin
 	s.sender.Stop()
 }
-
-// SetRetryDuration sets the retry duration on the wrapped Sender.
-func (s *senderPool) SetRetryDuration(duration time.Duration) {
-	s.sender.SetRetryDuration(duration)
-}
-
-// RetryDuration returns the retry duration of the wrapped Sender.
-func (s *senderPool) RetryDuration() time.Duration {
-	return s.sender.RetryDuration()
-}

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

@@ -107,13 +107,11 @@ func TestSenderPool(t *testing.T) {
 	logger := testutil.NewNopLogger()
 	mockService := new(mockLogsService)
 	mockService.On("PutLogEvents", mock.Anything).Return(&cloudwatchlogs.PutLogEventsOutput{}, nil)
-	s := newSender(logger, mockService, nil, time.Second, nil)
+	s := newSender(logger, mockService, nil, nil)
 	p := NewWorkerPool(12)
 	sp := newSenderPool(p, s)
 
-	assert.Equal(t, time.Second, sp.RetryDuration())
-	sp.SetRetryDuration(time.Minute)
-	assert.Equal(t, time.Minute, sp.RetryDuration())
+	// Retry duration methods removed - just test basic functionality
 
 	var completed atomic.Int32
 	var evts []*logEvent
@@ -144,7 +142,7 @@ func TestSenderPoolRetryHeap(t *testing.T) {
 	retryHeap := NewRetryHeap(10)
 	defer retryHeap.Stop()
 
-	s := newSender(logger, mockService, nil, time.Second, retryHeap)
+	s := newSender(logger, mockService, nil, retryHeap)
 	p := NewWorkerPool(12)
 	defer p.Stop()
 

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

@@ -32,12 +32,11 @@ func NewPusher(
 	entityProvider logs.LogEntityProvider,
 	workerPool WorkerPool,
 	flushTimeout time.Duration,
-	retryDuration time.Duration,
 	wg *sync.WaitGroup,
 	concurrency int,
 	retryHeap RetryHeap,
 ) *Pusher {
-	s := createSender(logger, service, targetManager, workerPool, retryDuration, retryHeap)
+	s := createSender(logger, service, targetManager, workerPool, retryHeap)
 
 	q := newQueue(logger, target, flushTimeout, entityProvider, s, wg)
 	targetManager.PutRetentionPolicy(target)
@@ -62,10 +61,9 @@ func createSender(
 	service cloudWatchLogsService,
 	targetManager TargetManager,
 	workerPool WorkerPool,
-	retryDuration time.Duration,
 	retryHeap RetryHeap,
 ) Sender {
-	s := newSender(logger, service, targetManager, retryDuration, retryHeap)
+	s := newSender(logger, service, targetManager, retryHeap)
 	if workerPool == nil {
 		return s
 	}

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

@@ -111,7 +111,6 @@ func setupPusher(t *testing.T, workerPool WorkerPool, wg *sync.WaitGroup) *Pushe
 		nil,
 		workerPool,
 		time.Second,
-		time.Minute,
 		wg,
 		1,   // concurrency
 		nil, // retryHeap
@@ -149,7 +148,6 @@ func TestPusherRetryHeap(t *testing.T) {
 		nil,
 		workerPool,
 		time.Second,
-		time.Minute,
 		&wg,
 		2, // concurrency > 1
 		retryHeap,

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

@@ -77,15 +77,6 @@ func (m *mockSender) Send(batch *logEventBatch) {
 	m.Called(batch)
 }
 
-func (m *mockSender) SetRetryDuration(d time.Duration) {
-	m.Called(d)
-}
-
-func (m *mockSender) RetryDuration() time.Duration {
-	args := m.Called()
-	return args.Get(0).(time.Duration)
-}
-
 func (m *mockSender) Stop() {
 	m.Called()
 }
@@ -123,7 +114,7 @@ func TestAddSingleEvent_WithAccountId(t *testing.T) {
 	}
 
 	ep := newMockEntityProvider(expectedEntity)
-	q, sender := testPreparation(t, -1, &s, 1*time.Hour, 2*time.Hour, ep, &wg)
+	q, sender := testPreparation(t, -1, &s, 1*time.Hour, ep, &wg)
 	q.AddEvent(newStubLogEvent("MSG", time.Now()))
 	require.False(t, called.Load(), "PutLogEvents has been called too fast, it should wait until FlushTimeout.")
 
@@ -160,7 +151,7 @@ func TestAddSingleEvent_WithoutAccountId(t *testing.T) {
 	}
 
 	ep := newMockEntityProvider(nil)
-	q, sender := testPreparation(t, -1, &s, 1*time.Hour, 2*time.Hour, ep, &wg)
+	q, sender := testPreparation(t, -1, &s, 1*time.Hour, ep, &wg)
 	q.AddEvent(newStubLogEvent("MSG", time.Now()))
 	require.False(t, called.Load(), "PutLogEvents has been called too fast, it should wait until FlushTimeout.")
 
@@ -190,7 +181,7 @@ func TestStopQueueWouldDoFinalSend(t *testing.T) {
 		return &cloudwatchlogs.PutLogEventsOutput{}, nil
 	}
 
-	q, sender := testPreparation(t, -1, &s, 1*time.Hour, 2*time.Hour, nil, &wg)
+	q, sender := testPreparation(t, -1, &s, 1*time.Hour, nil, &wg)
 	q.AddEvent(newStubLogEvent("MSG", time.Now()))
 
 	time.Sleep(10 * time.Millisecond)
@@ -214,7 +205,7 @@ func TestStopPusherWouldStopRetries(t *testing.T) {
 	}
 
 	logSink := testutil.NewLogSink()
-	q, sender := testPreparationWithLogger(t, logSink, -1, &s, 1*time.Hour, 2*time.Hour, nil, &wg)
+	q, sender := testPreparationWithLogger(t, logSink, -1, &s, 1*time.Hour, nil, &wg)
 	q.AddEvent(newStubLogEvent("MSG", time.Now()))
 	time.Sleep(10 * time.Millisecond)
 
@@ -256,7 +247,7 @@ func TestLongMessageHandling(t *testing.T) {
 		return &cloudwatchlogs.PutLogEventsOutput{}, nil
 	}
 
-	q, sender := testPreparation(t, -1, &s, 1*time.Hour, 2*time.Hour, nil, &wg)
+	q, sender := testPreparation(t, -1, &s, 1*time.Hour, nil, &wg)
 	q.AddEvent(newStubLogEvent(longMsg, time.Now()))
 
 	triggerSend(t, q)
@@ -285,7 +276,7 @@ func TestRequestIsLessThan1MB(t *testing.T) {
 		return &cloudwatchlogs.PutLogEventsOutput{}, nil
 	}
 
-	q, sender := testPreparation(t, -1, &s, 1*time.Hour, 2*time.Hour, nil, &wg)
+	q, sender := testPreparation(t, -1, &s, 1*time.Hour, nil, &wg)
 	for i := 0; i < 8; i++ {
 		q.AddEvent(newStubLogEvent(longMsg, time.Now()))
 	}
@@ -311,7 +302,7 @@ func TestRequestIsLessThan10kEvents(t *testing.T) {
 		return &cloudwatchlogs.PutLogEventsOutput{}, nil
 	}
 
-	q, sender := testPreparation(t, -1, &s, 1*time.Hour, 2*time.Hour, nil, &wg)
+	q, sender := testPreparation(t, -1, &s, 1*time.Hour, nil, &wg)
 	for i := 0; i < 30000; i++ {
 		q.AddEvent(newStubLogEvent(msg, time.Now()))
 	}
@@ -337,7 +328,7 @@ func TestTimestampPopulation(t *testing.T) {
 		return &cloudwatchlogs.PutLogEventsOutput{}, nil
 	}
 
-	q, sender := testPreparation(t, -1, &s, 1*time.Hour, 2*time.Hour, nil, &wg)
+	q, sender := testPreparation(t, -1, &s, 1*time.Hour, nil, &wg)
 	for i := 0; i < 3; i++ {
 		q.AddEvent(newStubLogEvent("msg", time.Time{}))
 	}
@@ -361,7 +352,7 @@ func TestIgnoreOutOfTimeRangeEvent(t *testing.T) {
 	}
 
 	logSink := testutil.NewLogSink()
-	q, sender := testPreparationWithLogger(t, logSink, -1, &s, 10*time.Millisecond, 2*time.Hour, nil, &wg)
+	q, sender := testPreparationWithLogger(t, logSink, -1, &s, 10*time.Millisecond, nil, &wg)
 	q.AddEvent(newStubLogEvent("MSG", time.Now().Add(-15*24*time.Hour)))
 	q.AddEventNonBlocking(newStubLogEvent("MSG", time.Now().Add(2*time.Hour+1*time.Minute)))
 
@@ -414,7 +405,7 @@ func TestAddMultipleEvents(t *testing.T) {
 		))
 	}
 	evts[10], evts[90] = evts[90], evts[10] // make events out of order
-	q, sender := testPreparation(t, -1, &s, 1*time.Hour, 2*time.Hour, nil, &wg)
+	q, sender := testPreparation(t, -1, &s, 1*time.Hour, nil, &wg)
 	for _, e := range evts {
 		q.AddEvent(e)
 	}
@@ -466,7 +457,7 @@ func TestSendReqWhenEventsSpanMoreThan24Hrs(t *testing.T) {
 		return nil, nil
 	}
 
-	q, sender := testPreparation(t, -1, &s, 1*time.Hour, 2*time.Hour, nil, &wg)
+	q, sender := testPreparation(t, -1, &s, 1*time.Hour, nil, &wg)
 	q.AddEvent(newStubLogEvent("MSG 25hrs ago", time.Now().Add(-25*time.Hour)))
 	q.AddEvent(newStubLogEvent("MSG 24hrs ago", time.Now().Add(-24*time.Hour)))
 	q.AddEvent(newStubLogEvent("MSG 23hrs ago", time.Now().Add(-23*time.Hour)))
@@ -496,7 +487,7 @@ func TestUnhandledErrorWouldNotResend(t *testing.T) {
 	}
 
 	logSink := testutil.NewLogSink()
-	q, sender := testPreparationWithLogger(t, logSink, -1, &s, 10*time.Millisecond, 2*time.Hour, nil, &wg)
+	q, sender := testPreparationWithLogger(t, logSink, -1, &s, 10*time.Millisecond, nil, &wg)
 	q.AddEvent(newStubLogEvent("msg", time.Now()))
 	time.Sleep(2 * time.Second)
 
@@ -542,7 +533,7 @@ func TestCreateLogGroupAndLogStreamWhenNotFound(t *testing.T) {
 	}
 
 	logSink := testutil.NewLogSink()
-	q, sender := testPreparationWithLogger(t, logSink, -1, &s, 1*time.Hour, 2*time.Hour, nil, &wg)
+	q, sender := testPreparationWithLogger(t, logSink, -1, &s, 1*time.Hour, nil, &wg)
 	var eventWG sync.WaitGroup
 	eventWG.Add(1)
 	q.AddEvent(&stubLogEvent{message: "msg", timestamp: time.Now(), done: eventWG.Done})
@@ -580,7 +571,7 @@ func TestLogRejectedLogEntryInfo(t *testing.T) {
 	}
 
 	logSink := testutil.NewLogSink()
-	q, sender := testPreparationWithLogger(t, logSink, -1, &s, 1*time.Hour, 2*time.Hour, nil, &wg)
+	q, sender := testPreparationWithLogger(t, logSink, -1, &s, 1*time.Hour, nil, &wg)
 	var eventWG sync.WaitGroup
 	eventWG.Add(1)
 	q.AddEvent(&stubLogEvent{message: "msg", timestamp: time.Now(), done: eventWG.Done})
@@ -630,7 +621,7 @@ func TestAddEventNonBlocking(t *testing.T) {
 			start.Add(time.Duration(i)*time.Millisecond),
 		))
 	}
-	q, sender := testPreparation(t, -1, &s, 1*time.Hour, 2*time.Hour, nil, &wg)
+	q, sender := testPreparation(t, -1, &s, 1*time.Hour, nil, &wg)
 	time.Sleep(200 * time.Millisecond) // Wait until pusher started, merge channel is blocked
 
 	for _, e := range evts {
@@ -646,32 +637,6 @@ func TestAddEventNonBlocking(t *testing.T) {
 	wg.Wait()
 }
 
-func TestResendWouldStopAfterExhaustedRetries(t *testing.T) {
-	t.Parallel()
-	var wg sync.WaitGroup
-	var s stubLogsService
-	var cnt atomic.Int32
-
-	s.ple = func(*cloudwatchlogs.PutLogEventsInput) (*cloudwatchlogs.PutLogEventsOutput, error) {
-		cnt.Add(1)
-		return nil, &cloudwatchlogs.ServiceUnavailableException{}
-	}
-
-	logSink := testutil.NewLogSink()
-	q, sender := testPreparationWithLogger(t, logSink, -1, &s, 10*time.Millisecond, time.Second, nil, &wg)
-	q.AddEvent(newStubLogEvent("msg", time.Now()))
-	time.Sleep(2 * time.Second)
-
-	logLines := logSink.Lines()
-	lastLine := logLines[len(logLines)-1]
-	expected := fmt.Sprintf("All %v retries to G/S failed for PutLogEvents, request dropped.", cnt.Load()-1)
-	require.True(t, strings.HasSuffix(lastLine, expected), fmt.Sprintf("Expecting error log to end with request dropped, but received '%s' in the log", logSink.String()))
-
-	q.Stop()
-	sender.Stop()
-	wg.Wait()
-}
-
 // Cannot call q.send() directly as it would cause a race condition. Reset last sent time and trigger flush.
 func triggerSend(t *testing.T, q *queue) {
 	t.Helper()
@@ -684,7 +649,6 @@ func testPreparation(
 	retention int,
 	service cloudWatchLogsService,
 	flushTimeout time.Duration,
-	retryDuration time.Duration,
 	entityProvider logs.LogEntityProvider,
 	wg *sync.WaitGroup,
 ) (*queue, Sender) {
@@ -694,7 +658,6 @@ func testPreparation(
 		retention,
 		service,
 		flushTimeout,
-		retryDuration,
 		entityProvider,
 		wg,
 	)
@@ -706,13 +669,12 @@ func testPreparationWithLogger(
 	retention int,
 	service cloudWatchLogsService,
 	flushTimeout time.Duration,
-	retryDuration time.Duration,
 	entityProvider logs.LogEntityProvider,
 	wg *sync.WaitGroup,
 ) (*queue, Sender) {
 	t.Helper()
 	tm := NewTargetManager(logger, service)
-	s := newSender(logger, service, tm, retryDuration, nil)
+	s := newSender(logger, service, tm, nil)
 	q := newQueue(
 		logger,
 		Target{"G", "S", util.StandardLogGroupClass, retention},

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

@@ -127,8 +127,8 @@ type RetryHeapProcessor struct {
 func NewRetryHeapProcessor(retryHeap RetryHeap, workerPool WorkerPool, service cloudWatchLogsService, targetManager TargetManager, logger telegraf.Logger) *RetryHeapProcessor {
 	// Create processor's own sender and senderPool
 	// Pass retryHeap so failed batches go back to RetryHeap instead of blocking on sync retry
-	// Use the same maxRetryTimeout that main pusher uses for consistency
-	sender := newSender(logger, service, targetManager, 14*24*time.Hour+10*time.Minute, retryHeap)
+	// Use the same default retry timeout that main pusher uses for consistency
+	sender := newSender(logger, service, targetManager, retryHeap)
 	senderPool := newSenderPool(workerPool, sender)
 
 	return &RetryHeapProcessor{