From 8271f62dce88e3f26e26ff4e27fd1221c674a1b1 Mon Sep 17 00:00:00 2001
From: Rick <dricross@amazon.com>
Date: Wed, 4 Sep 2024 13:19:06 -0400
Subject: [PATCH] Increase retry delay on PLE for certain exceptions (#1312)

---
 plugins/outputs/cloudwatchlogs/pusher.go     |  31 ++-
 plugins/outputs/cloudwatchlogs/retry.go      | 125 +++++++++++
 plugins/outputs/cloudwatchlogs/retry_test.go | 213 +++++++++++++++++++
 3 files changed, 353 insertions(+), 16 deletions(-)
 create mode 100644 plugins/outputs/cloudwatchlogs/retry.go
 create mode 100644 plugins/outputs/cloudwatchlogs/retry_test.go

diff --git a/plugins/outputs/cloudwatchlogs/pusher.go b/plugins/outputs/cloudwatchlogs/pusher.go
index 10cc50a364..1f45dddd80 100644
--- a/plugins/outputs/cloudwatchlogs/pusher.go
+++ b/plugins/outputs/cloudwatchlogs/pusher.go
@@ -235,7 +235,8 @@ func (p *pusher) send() {
 
 	startTime := time.Now()
 
-	retryCount := 0
+	retryCountShort := 0
+	retryCountLong := 0
 	for {
 		input.SequenceToken = p.sequenceToken
 		output, err := p.Service.PutLogEvents(input)
@@ -304,38 +305,36 @@ func (p *pusher) send() {
 			p.Log.Errorf("Aws error received when sending logs to %v/%v: %v", p.Group, p.Stream, awsErr)
 		}
 
-		wait := retryWait(retryCount)
+		// retry wait strategy depends on the type of error returned
+		var wait time.Duration
+		if chooseRetryWaitStrategy(err) == retryLong {
+			wait = retryWaitLong(retryCountLong)
+			retryCountLong++
+		} else {
+			wait = retryWaitShort(retryCountShort)
+			retryCountShort++
+		}
+
 		if time.Since(startTime)+wait > p.RetryDuration {
-			p.Log.Errorf("All %v retries to %v/%v failed for PutLogEvents, request dropped.", retryCount, p.Group, p.Stream)
+			p.Log.Errorf("All %v retries to %v/%v failed for PutLogEvents, request dropped.", retryCountShort+retryCountLong-1, p.Group, p.Stream)
 			p.reset()
 			return
 		}
 
-		p.Log.Warnf("Retried %v time, going to sleep %v before retrying.", retryCount, wait)
+		p.Log.Warnf("Retried %v time, going to sleep %v before retrying.", retryCountShort+retryCountLong-1, wait)
 
 		select {
 		case <-p.stop:
-			p.Log.Errorf("Stop requested after %v retries to %v/%v failed for PutLogEvents, request dropped.", retryCount, p.Group, p.Stream)
+			p.Log.Errorf("Stop requested after %v retries to %v/%v failed for PutLogEvents, request dropped.", retryCountShort+retryCountLong-1, p.Group, p.Stream)
 			p.reset()
 			return
 		case <-time.After(wait):
 		}
 
-		retryCount++
 	}
 
 }
 
-func retryWait(n int) time.Duration {
-	const base = 200 * time.Millisecond
-	// Max wait time is 1 minute (jittered)
-	d := 1 * time.Minute
-	if n < 5 {
-		d = base * time.Duration(1<<int64(n))
-	}
-	return time.Duration(seededRand.Int63n(int64(d/2)) + int64(d/2))
-}
-
 func (p *pusher) createLogGroupAndStream() error {
 	_, err := p.Service.CreateLogStream(&cloudwatchlogs.CreateLogStreamInput{
 		LogGroupName:  &p.Group,
diff --git a/plugins/outputs/cloudwatchlogs/retry.go b/plugins/outputs/cloudwatchlogs/retry.go
new file mode 100644
index 0000000000..409a97499b
--- /dev/null
+++ b/plugins/outputs/cloudwatchlogs/retry.go
@@ -0,0 +1,125 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package cloudwatchlogs
+
+import (
+	"net"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/aws/request"
+
+	"github.com/aws/amazon-cloudwatch-agent/sdk/service/cloudwatchlogs"
+)
+
+const (
+	// baseRetryDelayShort is the base retry delay for short retry strategy
+	baseRetryDelayShort = 200 * time.Millisecond
+
+	// baseRetryDelayLong is the base retry delay for long retry strategy
+	baseRetryDelayLong = 2 * time.Second
+
+	// numBackoffRetriesShort is the maximum number of consecutive retries using the short retry strategy before using
+	// the maxRetryDelay
+	numBackoffRetriesShort = 5
+
+	// numBackoffRetriesLong is the maximum number of consecutive retries using the long retry strategy before using
+	// the maxRetryDelay
+	numBackoffRetriesLong = 2
+
+	// maxRetryDelay is the maximum retry delay for the either retry strategy
+	maxRetryDelay = 1 * time.Minute
+)
+
+type retryWaitStrategy int
+
+const (
+	retryShort retryWaitStrategy = iota
+	retryLong
+)
+
+// retryWaitShort returns a duration to wait before retrying a request using the short retry strategy
+func retryWaitShort(retryCount int) time.Duration {
+	return retryWait(baseRetryDelayShort, numBackoffRetriesShort, retryCount)
+}
+
+// retryWaitLong returns a duration to wait before retrying a request using the long retry strategy.
+// this strategy is used for errors that should not be retried too quickly
+func retryWaitLong(retryCount int) time.Duration {
+	return retryWait(baseRetryDelayLong, numBackoffRetriesLong, retryCount)
+}
+
+func retryWait(baseRetryDelay time.Duration, maxBackoffRetries int, retryCount int) time.Duration {
+	d := maxRetryDelay
+	if retryCount < maxBackoffRetries {
+		d = baseRetryDelay * time.Duration(1<<int64(retryCount))
+	}
+	return time.Duration(seededRand.Int63n(int64(d/2)) + int64(d/2))
+}
+
+// chooseRetryWaitStrategy decides if a "long" or "short" retry strategy should be used when the PutLogEvents API call
+// returns an error. A short retry strategy should be used for most errors, while a long retry strategy is used for
+// errors where retrying too quickly could cause excessive strain on the backend servers.
+//
+// Specifically, use the long retry strategy for the following PutLogEvents errors:
+//   - 500 (InternalFailure)
+//   - 503 (ServiceUnavailable)
+//   - Connection Refused
+//   - Connection Reset by Peer
+//   - Connection Timeout
+//   - Throttling
+func chooseRetryWaitStrategy(err error) retryWaitStrategy {
+	if isErrConnectionTimeout(err) || isErrConnectionReset(err) || isErrConnectionRefused(err) || request.IsErrorThrottle(err) {
+		return retryLong
+	}
+
+	// Check AWS Error codes if available
+	if awsErr, ok := err.(awserr.Error); ok {
+		switch awsErr.Code() {
+		case
+			cloudwatchlogs.ErrCodeServiceUnavailableException,
+			cloudwatchlogs.ErrCodeThrottlingException,
+			"RequestTimeout",
+			request.ErrCodeResponseTimeout:
+			return retryLong
+		}
+
+		// Check HTTP status codes if available
+		if requestFailure, ok := err.(awserr.RequestFailure); ok {
+			switch requestFailure.StatusCode() {
+			case
+				500, // internal failure
+				503: // service unavailable
+				return retryLong
+			}
+		}
+	}
+
+	// Otherwise, default to short retry strategy
+	return retryShort
+}
+
+func isErrConnectionTimeout(err error) bool {
+	netErr, ok := err.(net.Error)
+	return ok && netErr.Timeout()
+}
+
+func isErrConnectionReset(err error) bool {
+	if strings.Contains(err.Error(), "read: connection reset") {
+		return false
+	}
+
+	if strings.Contains(err.Error(), "use of closed network connection") ||
+		strings.Contains(err.Error(), "connection reset") ||
+		strings.Contains(err.Error(), "broken pipe") {
+		return true
+	}
+
+	return false
+}
+
+func isErrConnectionRefused(err error) bool {
+	return strings.Contains(err.Error(), "connection refused")
+}
diff --git a/plugins/outputs/cloudwatchlogs/retry_test.go b/plugins/outputs/cloudwatchlogs/retry_test.go
new file mode 100644
index 0000000000..0c0bb65001
--- /dev/null
+++ b/plugins/outputs/cloudwatchlogs/retry_test.go
@@ -0,0 +1,213 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package cloudwatchlogs
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/aws/amazon-cloudwatch-agent/sdk/service/cloudwatchlogs"
+)
+
+func TestChooseRetryWaitStrategy(t *testing.T) {
+	t.Parallel()
+	tests := map[string]struct {
+		err              error
+		expectedStrategy retryWaitStrategy
+	}{
+		"ResourceNotFoundException": {
+			err:              &cloudwatchlogs.ResourceNotFoundException{},
+			expectedStrategy: retryShort,
+		},
+		"InvalidSequenceTokenException": {
+			err:              &cloudwatchlogs.InvalidSequenceTokenException{},
+			expectedStrategy: retryShort,
+		},
+		"ServiceUnavailableException": {
+			err:              &cloudwatchlogs.ServiceUnavailableException{},
+			expectedStrategy: retryLong,
+		},
+		"ThrottlingException": {
+			err:              &cloudwatchlogs.ThrottlingException{},
+			expectedStrategy: retryLong,
+		},
+		"500 - InternalFailure": {
+			err:              awserr.NewRequestFailure(awserr.New("500", "InternalFailure", nil), 500, ""),
+			expectedStrategy: retryLong,
+		},
+		"503 - ServiceUnavailable": {
+			err:              awserr.NewRequestFailure(awserr.New("503", "ServiceUnavailable", nil), 503, ""),
+			expectedStrategy: retryLong,
+		},
+		"Connection Refused": {
+			err:              awserr.New("SomeError", "connection refused", nil),
+			expectedStrategy: retryLong,
+		},
+		"Connection Refused - syscall": {
+			err:              syscall.ECONNREFUSED,
+			expectedStrategy: retryLong,
+		},
+		"Connection Reset By Peer": {
+			err:              awserr.New("SomeError", "connection reset by peer", nil),
+			expectedStrategy: retryLong,
+		},
+		"Connection Reset By Peer - syscall": {
+			err:              syscall.ECONNRESET,
+			expectedStrategy: retryLong,
+		},
+		"Connection Timeout": {
+			err:              syscall.ETIMEDOUT,
+			expectedStrategy: retryLong,
+		},
+		"Request Timeout": {
+			err:              awserr.New("RequestTimeout", "request timed out", nil),
+			expectedStrategy: retryLong,
+		},
+		"Response Timeout": {
+			err:              awserr.New("ResponseTimeout", "response timed out", nil),
+			expectedStrategy: retryLong,
+		},
+		"Deadline Exceeded": {
+			err:              os.ErrDeadlineExceeded,
+			expectedStrategy: retryLong,
+		},
+		"Other Errors": {
+			err:              errors.New("Unknown Error"),
+			expectedStrategy: retryShort,
+		},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			chosen := chooseRetryWaitStrategy(tt.err)
+			require.Equal(t, tt.expectedStrategy, chosen)
+		})
+	}
+}
+
+func TestRetryWaitShort(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		retryCount  int
+		minDuration time.Duration
+		maxDuration time.Duration
+	}{
+		{
+			retryCount:  0,
+			minDuration: 100 * time.Millisecond,
+			maxDuration: 200 * time.Millisecond,
+		},
+		{
+			retryCount:  1,
+			minDuration: 200 * time.Millisecond,
+			maxDuration: 400 * time.Millisecond,
+		},
+		{
+			retryCount:  2,
+			minDuration: 400 * time.Millisecond,
+			maxDuration: 800 * time.Millisecond,
+		},
+		{
+			retryCount:  3,
+			minDuration: 800 * time.Millisecond,
+			maxDuration: 1600 * time.Millisecond,
+		},
+		{
+			retryCount:  4,
+			minDuration: 1600 * time.Millisecond,
+			maxDuration: 3200 * time.Millisecond,
+		},
+		{
+			retryCount:  5,
+			minDuration: 30 * time.Second,
+			maxDuration: 1 * time.Minute,
+		},
+		{
+			retryCount:  6,
+			minDuration: 30 * time.Second,
+			maxDuration: 1 * time.Minute,
+		},
+		{
+			retryCount:  7,
+			minDuration: 30 * time.Second,
+			maxDuration: 1 * time.Minute,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(fmt.Sprintf("%d", tt.retryCount), func(t *testing.T) {
+			for _ = range 1000 {
+				duration := retryWaitShort(tt.retryCount)
+				assert.GreaterOrEqual(t, duration, tt.minDuration, "retryWaitShort(%v) should be greater than or equal to %v", tt.retryCount, tt.minDuration)
+				assert.LessOrEqual(t, duration, tt.maxDuration, "retryWaitShort(%v) should be less than or equal to %v", tt.retryCount, tt.maxDuration)
+			}
+		})
+	}
+}
+
+func TestRetryWaitLong(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		retryCount  int
+		minDuration time.Duration
+		maxDuration time.Duration
+	}{
+		{
+			retryCount:  0,
+			minDuration: 1 * time.Second,
+			maxDuration: 2 * time.Second,
+		},
+		{
+			retryCount:  1,
+			minDuration: 2 * time.Second,
+			maxDuration: 4 * time.Second,
+		},
+		{
+			retryCount:  2,
+			minDuration: 30 * time.Second,
+			maxDuration: 1 * time.Minute,
+		},
+		{
+			retryCount:  3,
+			minDuration: 30 * time.Second,
+			maxDuration: 1 * time.Minute,
+		},
+		{
+			retryCount:  4,
+			minDuration: 30 * time.Second,
+			maxDuration: 1 * time.Minute,
+		},
+		{
+			retryCount:  5,
+			minDuration: 30 * time.Second,
+			maxDuration: 1 * time.Minute,
+		},
+		{
+			retryCount:  6,
+			minDuration: 30 * time.Second,
+			maxDuration: 1 * time.Minute,
+		},
+		{
+			retryCount:  7,
+			minDuration: 30 * time.Second,
+			maxDuration: 1 * time.Minute,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(fmt.Sprintf("%d", tt.retryCount), func(t *testing.T) {
+			for _ = range 1000 {
+				duration := retryWaitLong(tt.retryCount)
+				assert.GreaterOrEqual(t, duration, tt.minDuration, "retryWaitLong(%v) should be greater than or equal to %v", tt.retryCount, tt.minDuration)
+				assert.LessOrEqual(t, duration, tt.maxDuration, "retryWaitLong(%v) should be less than or equal to %v", tt.retryCount, tt.maxDuration)
+			}
+		})
+	}
+}