From 3a460a322697334bc56a59c3d6bdcfe1cd9222de Mon Sep 17 00:00:00 2001 From: Lorenzo Spizzuoco Date: Thu, 9 Jul 2026 13:20:08 +0000 Subject: [PATCH] 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. --- internal/retryer/logthrottle.go | 6 +- internal/retryer/logthrottle_test.go | 33 +++++++ .../outputs/cloudwatchlogs/cloudwatchlogs.go | 15 ++- .../cloudwatchlogs/cloudwatchlogs_test.go | 44 +++++++++ .../internal/pusher/target_deadlock_test.go | 98 +++++++++++++++++++ 5 files changed, 194 insertions(+), 2 deletions(-) create mode 100644 plugins/outputs/cloudwatchlogs/internal/pusher/target_deadlock_test.go diff --git a/internal/retryer/logthrottle.go b/internal/retryer/logthrottle.go index ae1bc2843cc..3a18851cef9 100644 --- a/internal/retryer/logthrottle.go +++ b/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 diff --git a/internal/retryer/logthrottle_test.go b/internal/retryer/logthrottle_test.go index 6cffd793647..06e3aae8c85 100644 --- a/internal/retryer/logthrottle_test.go +++ b/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 diff --git a/plugins/outputs/cloudwatchlogs/cloudwatchlogs.go b/plugins/outputs/cloudwatchlogs/cloudwatchlogs.go index 770ef5e3f97..f11543ee612 100644 --- a/plugins/outputs/cloudwatchlogs/cloudwatchlogs.go +++ b/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{ diff --git a/plugins/outputs/cloudwatchlogs/cloudwatchlogs_test.go b/plugins/outputs/cloudwatchlogs/cloudwatchlogs_test.go index 66f1643fd09..9e924c55ff3 100644 --- a/plugins/outputs/cloudwatchlogs/cloudwatchlogs_test.go +++ b/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() +} diff --git a/plugins/outputs/cloudwatchlogs/internal/pusher/target_deadlock_test.go b/plugins/outputs/cloudwatchlogs/internal/pusher/target_deadlock_test.go new file mode 100644 index 00000000000..16c76786754 --- /dev/null +++ b/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") + } +}