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