fix: prevent throttling deadlock in cloudwatchlogs output

On a ThrottlingException for CreateLogStream/CreateLogGroup, log-stream
creation could wedge for the entire output plugin with nothing logged.

LogThrottleRetryer.ShouldRetry performed an unconditional blocking send to
a capacity-1 throttle-event channel drained by a single consumer goroutine.
The shared TargetManager was bound (via sync.Once) to the first
destination's client and retryer; when that destination stopped (e.g. log
rotation with auto_removal), its retryer's consumer goroutine exited. The
next throttle then filled the buffer and blocked the send forever, and
because TargetManager.InitTarget holds a mutex across the create call, all
other targets waiting on that mutex stalled as well.

Two changes, either of which breaks the deadlock; both applied:

- logthrottle.go: make the throttle-event send non-blocking (select with a
  default) so a full channel or stopped consumer can never block
  ShouldRetry. The retry decision is unchanged.
- cloudwatchlogs.go: give the shared TargetManager its own dedicated
  retryer and client owned by the plugin, stopped only in Close(), instead
  of reusing the first destination's client. This decouples the
  TargetManager lifecycle from any single destination.

Tests:
- TestShouldRetryDoesNotBlockAfterStop: ShouldRetry returns after the
  retryer is stopped (fails via timeout on the pre-fix code).
- TestSharedRetryerLifecycle: stopping the first destination does not
  prevent creating additional destinations.
- TestInitTargetNoDeadlockUnderThrottling: end-to-end regression driving
  the real SDK retry loop against an always-throttling endpoint through
  InitTarget with a stopped retryer consumer; deadlocks on the pre-fix
  code, passes with the fix.

Author: lorespiz

internal/retryer/logthrottle.go

@@ -53,7 +53,11 @@ func (r *LogThrottleRetryer) ShouldRetry(req *request.Request) bool {
 		if req.Operation != nil {
 			te.Operation = req.Operation.Name
 		}
-		r.throttleChan <- te
+		// Non-blocking: never block ShouldRetry if the consumer has stopped.
+		select {
+		case r.throttleChan <- te:
+		default:
+		}
 	}
 
 	// Fallback to SDK's built in retry rules

internal/retryer/logthrottle_test.go

@@ -123,6 +123,39 @@ func TestLogThrottleRetryerLogging(t *testing.T) {
 	}
 }
 
+// TestShouldRetryDoesNotBlockAfterStop verifies ShouldRetry does not block once the
+// retryer is stopped (its consumer goroutine no longer drains the throttle channel).
+func TestShouldRetryDoesNotBlockAfterStop(t *testing.T) {
+	l := &testLogger{}
+	r := NewLogThrottleRetryer(l)
+
+	// Stop the retryer, which closes the done channel and exits the consumer goroutine
+	r.Stop()
+	time.Sleep(50 * time.Millisecond) // Give the goroutine time to exit
+
+	req := &request.Request{
+		Error:     awserr.New("RequestLimitExceeded", "Test AWS Error", nil),
+		Operation: &request.Operation{Name: "Test"},
+	}
+
+	// Call ShouldRetry in a goroutine and use a timeout to detect blocking
+	done := make(chan bool, 1)
+	go func() {
+		// Call ShouldRetry multiple times to exceed channel capacity (1)
+		for i := 0; i < 10; i++ {
+			r.ShouldRetry(req)
+		}
+		done <- true
+	}()
+
+	select {
+	case <-done:
+		// Success: ShouldRetry did not block
+	case <-time.After(2 * time.Second):
+		t.Fatal("ShouldRetry blocked after retryer was stopped - potential deadlock")
+	}
+}
+
 func setup() {
 	throttleReportTimeout = 400 * time.Millisecond
 	throttleReportCheckPeriod = 50 * time.Millisecond

plugins/outputs/cloudwatchlogs/cloudwatchlogs.go

@@ -77,6 +77,11 @@ type CloudWatchLogs struct {
 	middleware      awsmiddleware.Middleware
 	configurer      *awsmiddleware.Configurer
 	configurerOnce  sync.Once
+
+	// Dedicated retryer/client for the TargetManager, owned by the plugin so its
+	// lifecycle is independent of any destination stop.
+	sharedRetryer *retryer.LogThrottleRetryer
+	sharedClient  *cloudwatchlogs.CloudWatchLogs
 }
 
 var _ logs.LogBackend = (*CloudWatchLogs)(nil)
@@ -101,6 +106,11 @@ func (c *CloudWatchLogs) Close() error {
 		c.workerPool.Stop()
 	}
 
+	// Stop the shared retryer last, after all pushers have drained.
+	if c.sharedRetryer != nil {
+		c.sharedRetryer.Stop()
+	}
+
 	return nil
 }
 
@@ -151,7 +161,10 @@ func (c *CloudWatchLogs) getDest(t pusher.Target, logSrc logs.LogSrc) *cwDest {
 		if c.Concurrency > 1 {
 			c.workerPool = pusher.NewWorkerPool(c.Concurrency)
 		}
-		c.targetManager = pusher.NewTargetManager(c.Log, client)
+		// Dedicated retryer/client so the TargetManager isn't tied to the first dest.
+		c.sharedRetryer = retryer.NewLogThrottleRetryer(c.Log)
+		c.sharedClient = c.createClient(c.sharedRetryer)
+		c.targetManager = pusher.NewTargetManager(c.Log, c.sharedClient)
 	})
 	p := pusher.NewPusher(c.Log, t, client, c.targetManager, logSrc, c.workerPool, c.ForceFlushInterval.Duration, maxRetryTimeout, &c.pusherWaitGroup)
 	cwd := &cwDest{

plugins/outputs/cloudwatchlogs/cloudwatchlogs_test.go

@@ -6,6 +6,7 @@ package cloudwatchlogs
 import (
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/influxdata/telegraf/testutil"
 	"github.com/stretchr/testify/require"
@@ -100,3 +101,46 @@ func TestDuplicateDestination(t *testing.T) {
 	// Then the destination for cloudwatchlogs endpoint would be the same
 	require.Equal(t, d1, d2)
 }
+
+// TestSharedRetryerLifecycle verifies that stopping one destination does not affect
+// the shared TargetManager's ability to create new targets, and that the shared
+// retryer is separate from any destination's retryer.
+func TestSharedRetryerLifecycle(t *testing.T) {
+	c := &CloudWatchLogs{
+		Log:       testutil.Logger{Name: "test"},
+		AccessKey: "access_key",
+		SecretKey: "secret_key",
+		cwDests:   sync.Map{},
+	}
+
+	// Create the first destination - this initializes the shared TargetManager
+	d1 := c.CreateDest("group1", "stream1", -1, "", nil).(*cwDest)
+
+	// Verify that the shared retryer was created and is separate from d1's retryer
+	require.NotNil(t, c.sharedRetryer, "shared retryer should be initialized")
+	require.NotNil(t, c.sharedClient, "shared client should be initialized")
+	require.NotSame(t, c.sharedRetryer, d1.retryer, "shared retryer should be separate from destination retryer")
+
+	// Stop the first destination (simulates log rotation with auto_removal)
+	d1.Stop()
+
+	// Create a second destination - this should not block or fail
+	done := make(chan *cwDest, 1)
+	go func() {
+		d2 := c.CreateDest("group2", "stream2", -1, "", nil).(*cwDest)
+		done <- d2
+	}()
+
+	select {
+	case d2 := <-done:
+		require.NotNil(t, d2, "second destination should be created successfully")
+		require.NotSame(t, d1, d2, "second destination should be different from first")
+		// Clean up
+		d2.Stop()
+	case <-time.After(5 * time.Second):
+		t.Fatal("creating second destination blocked after first destination was stopped - potential deadlock")
+	}
+
+	// Clean up
+	c.Close()
+}

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

@@ -0,0 +1,98 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package pusher
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/session"
+
+	"github.com/aws/amazon-cloudwatch-agent/internal/retryer"
+	"github.com/aws/amazon-cloudwatch-agent/sdk/service/cloudwatchlogs"
+	"github.com/aws/amazon-cloudwatch-agent/tool/testutil"
+)
+
+// newThrottlingClient returns a real CloudWatch Logs client whose endpoint points
+// at a local server that always responds with a ThrottlingException, wired with a
+// LogThrottleRetryer. The returned retryer is also handed back so the test can stop
+// its consumer goroutine to reproduce the dead-consumer condition.
+func newThrottlingClient(t *testing.T) (*cloudwatchlogs.CloudWatchLogs, *retryer.LogThrottleRetryer, func()) {
+	t.Helper()
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		// JSON 1.1 protocol: the SDK classifies the error from the error type,
+		// which it reads from this header / body. ThrottlingException is a
+		// throttling error, so the SDK will invoke ShouldRetry and retry.
+		w.Header().Set("X-Amzn-Errortype", "ThrottlingException")
+		w.Header().Set("Content-Type", "application/x-amz-json-1.1")
+		w.WriteHeader(http.StatusBadRequest)
+		_, _ = w.Write([]byte(`{"__type":"ThrottlingException","message":"Rate exceeded"}`))
+	}))
+
+	r := retryer.NewLogThrottleRetryer(testutil.NewNopLogger())
+	// Bound the retry count: fast, but >1 so a dead consumer fills the capacity-1
+	// throttle channel and (pre-fix) the next send blocks.
+	r.NumMaxRetries = 2
+
+	sess := session.Must(session.NewSession())
+	client := cloudwatchlogs.New(sess, &aws.Config{
+		Region:      aws.String("us-east-1"),
+		Endpoint:    aws.String(srv.URL),
+		DisableSSL:  aws.Bool(true),
+		Credentials: credentials.NewStaticCredentials("ak", "sk", ""),
+		Retryer:     r,
+	})
+
+	return client, r, srv.Close
+}
+
+// TestInitTargetNoDeadlockUnderThrottling drives the real SDK retry loop through
+// InitTarget while CreateLogStream is throttled and the retryer's consumer has been
+// stopped. InitTarget holds a mutex across the create call, so a blocking throttle
+// send would wedge every target. Asserts both targets' InitTarget return in time.
+func TestInitTargetNoDeadlockUnderThrottling(t *testing.T) {
+	t.Parallel()
+	client, r, closeSrv := newThrottlingClient(t)
+	defer closeSrv()
+
+	manager := NewTargetManager(testutil.NewNopLogger(), client)
+
+	// Stop the retryer's consumer BEFORE any calls, reproducing the dead-consumer
+	// condition that arises when the destination owning the retryer stops.
+	r.Stop()
+	time.Sleep(50 * time.Millisecond)
+
+	var wg sync.WaitGroup
+	done := make(chan struct{})
+	for i, target := range []Target{
+		{Group: "group-A", Stream: "stream-A"},
+		{Group: "group-B", Stream: "stream-B"},
+	} {
+		wg.Add(1)
+		go func(_ int, tg Target) {
+			defer wg.Done()
+			// Returns a throttling error after retries are exhausted; the point is
+			// that it RETURNS rather than parking forever inside the held mutex.
+			_ = manager.InitTarget(tg)
+		}(i, target)
+	}
+
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		// Both InitTarget calls returned: no deadlock.
+	case <-time.After(60 * time.Second):
+		t.Fatal("InitTarget deadlocked under throttling with a stopped retryer consumer")
+	}
+}