fix: prevent throttling deadlock in cloudwatchlogs output

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")
+	}
+}