Fallback to iterative DescribeLogGroups if batch not supported (#1915)

Author: duhminick

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

@@ -24,6 +24,8 @@ const (
 	baseRetryDelay      = 1 * time.Second
 	maxRetryDelayTarget = 10 * time.Second
 	numBackoffRetries   = 5
+
+	errMessageLogGroupIdentifierNotSupported = "Input filter on Log group identifiers is not supported."
 )
 
 type Target struct {
@@ -185,21 +187,31 @@ func (m *targetManager) processDescribeLogGroup() {
 		case target := <-m.dlg:
 			batch[target.Group] = target
 			if len(batch) == logGroupIdentifierLimit {
-				m.updateTargetBatch(batch)
+				m.updateTargets(batch)
 				// Reset batch
 				batch = make(map[string]Target, logGroupIdentifierLimit)
 			}
 		case <-t.C:
 			if len(batch) > 0 {
-				m.updateTargetBatch(batch)
+				m.updateTargets(batch)
 				// Reset batch
 				batch = make(map[string]Target, logGroupIdentifierLimit)
 			}
 		}
 	}
 }
 
-func (m *targetManager) updateTargetBatch(targets map[string]Target) {
+func (m *targetManager) updateTargets(targets map[string]Target) {
+	err := m.updateTargetsBatch(targets)
+	if err != nil {
+		m.logger.Debug("falling back to describing log groups by prefix")
+		m.updateTargetsIteratively(targets)
+	}
+}
+
+// updateTargetsBatch will call DLG for the entire batch (single call). Will return an error if
+// DLG by identifiers is not supported.
+func (m *targetManager) updateTargetsBatch(targets map[string]Target) error {
 	identifiers := make([]*string, 0, len(targets))
 	for logGroup := range targets {
 		identifiers = append(identifiers, aws.String(logGroup))
@@ -211,7 +223,11 @@ func (m *targetManager) updateTargetBatch(targets map[string]Target) {
 	for attempt := 0; attempt < numBackoffRetries; attempt++ {
 		output, err := m.service.DescribeLogGroups(describeLogGroupsInput)
 		if err != nil {
-			m.logger.Errorf("failed to describe log group retention for targets %v: %v", targets, err)
+			if aerr, ok := err.(awserr.Error); ok && aerr.Code() == cloudwatchlogs.ErrCodeInvalidParameterException && aerr.Message() == errMessageLogGroupIdentifierNotSupported {
+				return err
+			}
+
+			m.logger.Errorf("failed to batch describe log group retention for targets %v: %v", targets, err)
 			time.Sleep(m.calculateBackoff(attempt))
 			continue
 		}
@@ -225,6 +241,49 @@ func (m *targetManager) updateTargetBatch(targets map[string]Target) {
 		}
 		break
 	}
+	return nil
+}
+
+// updateTargetsIteratively will iterate through the targets and call DLG for each target.
+func (m *targetManager) updateTargetsIteratively(targets map[string]Target) {
+	for _, target := range targets {
+		for attempt := 0; attempt < numBackoffRetries; attempt++ {
+			currentRetention, err := m.getRetention(target)
+			if err != nil {
+				m.logger.Errorf("failed to describe log group retention for target %v: %v", target, err)
+				time.Sleep(m.calculateBackoff(attempt))
+				continue
+			}
+
+			if currentRetention != target.Retention && target.Retention > 0 {
+				m.logger.Debugf("queueing log group %v to update retention policy", target.Group)
+				m.prp <- target
+			}
+			break // no change in retention
+		}
+	}
+}
+
+func (m *targetManager) getRetention(target Target) (int, error) {
+	input := &cloudwatchlogs.DescribeLogGroupsInput{
+		LogGroupNamePrefix: aws.String(target.Group),
+	}
+
+	output, err := m.service.DescribeLogGroups(input)
+	if err != nil {
+		return 0, fmt.Errorf("describe log groups failed: %w", err)
+	}
+
+	for _, group := range output.LogGroups {
+		if *group.LogGroupName == target.Group {
+			if group.RetentionInDays == nil {
+				return 0, nil
+			}
+			return int(*group.RetentionInDays), nil
+		}
+	}
+
+	return 0, fmt.Errorf("log group %v not found", target.Group)
 }
 
 func (m *targetManager) processPutRetentionPolicy() {

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

@@ -460,14 +460,57 @@ func TestDescribeLogGroupsBatching(t *testing.T) {
 		batch := make(map[string]Target)
 		batch["group-1"] = Target{Group: "group-1", Stream: "stream", Retention: 7}
 		batch["group-2"] = Target{Group: "group-2", Stream: "stream", Retention: 7}
-		tm.updateTargetBatch(batch)
+		tm.updateTargets(batch)
 
 		// Wait for ticker to fire (slightly longer than 5 seconds)
 		time.Sleep(5100 * time.Millisecond)
 
 		mockService.AssertNotCalled(t, "PutRetentionPolicy")
 	})
 
+	t.Run("FallbackToPrefix", func(t *testing.T) {
+		mockService := new(mockLogsService)
+
+		// First call with identifiers fails with unsupported error
+		mockService.On("DescribeLogGroups", mock.MatchedBy(func(input *cloudwatchlogs.DescribeLogGroupsInput) bool {
+			return len(input.LogGroupIdentifiers) > 0
+		})).Return(&cloudwatchlogs.DescribeLogGroupsOutput{},
+			awserr.New(cloudwatchlogs.ErrCodeInvalidParameterException, errMessageLogGroupIdentifierNotSupported, nil)).Once()
+
+		// Fallback calls with prefix for each group
+		mockService.On("DescribeLogGroups", mock.MatchedBy(func(input *cloudwatchlogs.DescribeLogGroupsInput) bool {
+			return input.LogGroupNamePrefix != nil && *input.LogGroupNamePrefix == "group-1"
+		})).Return(&cloudwatchlogs.DescribeLogGroupsOutput{
+			LogGroups: []*cloudwatchlogs.LogGroup{
+				{LogGroupName: aws.String("group-1"), RetentionInDays: aws.Int64(1)},
+			},
+		}, nil).Once()
+
+		mockService.On("DescribeLogGroups", mock.MatchedBy(func(input *cloudwatchlogs.DescribeLogGroupsInput) bool {
+			return input.LogGroupNamePrefix != nil && *input.LogGroupNamePrefix == "group-2"
+		})).Return(&cloudwatchlogs.DescribeLogGroupsOutput{
+			LogGroups: []*cloudwatchlogs.LogGroup{
+				{LogGroupName: aws.String("group-2"), RetentionInDays: aws.Int64(7)},
+			},
+		}, nil).Once()
+
+		mockService.On("PutRetentionPolicy", mock.MatchedBy(func(input *cloudwatchlogs.PutRetentionPolicyInput) bool {
+			return *input.LogGroupName == "group-1" && *input.RetentionInDays == 7
+		})).Return(&cloudwatchlogs.PutRetentionPolicyOutput{}, nil).Once()
+
+		manager := NewTargetManager(logger, mockService)
+		tm := manager.(*targetManager)
+
+		batch := make(map[string]Target)
+		batch["group-1"] = Target{Group: "group-1", Stream: "stream", Retention: 7}
+		batch["group-2"] = Target{Group: "group-2", Stream: "stream", Retention: 7}
+
+		tm.updateTargets(batch)
+		time.Sleep(100 * time.Millisecond)
+
+		mockService.AssertExpectations(t)
+	})
+
 	t.Run("RetentionPolicyUpdate", func(t *testing.T) {
 		mockService := new(mockLogsService)
 
@@ -497,7 +540,7 @@ func TestDescribeLogGroupsBatching(t *testing.T) {
 		batch["group-1"] = Target{Group: "group-1", Stream: "stream", Retention: 7}
 		batch["group-2"] = Target{Group: "group-2", Stream: "stream", Retention: 7}
 
-		tm.updateTargetBatch(batch)
+		tm.updateTargets(batch)
 		time.Sleep(100 * time.Millisecond)
 
 		mockService.AssertExpectations(t)
@@ -523,13 +566,38 @@ func TestDescribeLogGroupsBatching(t *testing.T) {
 		batch := make(map[string]Target)
 		batch["group-1"] = Target{Group: "group-1", Stream: "stream", Retention: 7}
 
-		tm.updateTargetBatch(batch)
+		tm.updateTargets(batch)
 		// Sleep enough for retry
 		time.Sleep(2 * time.Second)
 
 		mockService.AssertExpectations(t)
 	})
 
+	t.Run("FallbackToPrefix/OtherError", func(t *testing.T) {
+		mockService := new(mockLogsService)
+
+		// First call with identifiers fails with different error (should not fallback)
+		mockService.On("DescribeLogGroups", mock.MatchedBy(func(input *cloudwatchlogs.DescribeLogGroupsInput) bool {
+			return len(input.LogGroupIdentifiers) > 0
+		})).Return(&cloudwatchlogs.DescribeLogGroupsOutput{},
+			awserr.New(cloudwatchlogs.ErrCodeInvalidParameterException, "Different error message", nil)).Times(numBackoffRetries)
+
+		manager := NewTargetManager(logger, mockService)
+		tm := manager.(*targetManager)
+
+		batch := make(map[string]Target)
+		batch["group-1"] = Target{Group: "group-1", Stream: "stream", Retention: 7}
+
+		tm.updateTargets(batch)
+		time.Sleep(2 * time.Second)
+
+		mockService.AssertExpectations(t)
+		// Should not call prefix-based DescribeLogGroups
+		mockService.AssertNotCalled(t, "DescribeLogGroups", mock.MatchedBy(func(input *cloudwatchlogs.DescribeLogGroupsInput) bool {
+			return input.LogGroupNamePrefix != nil
+		}))
+	})
+
 }
 
 func TestCalculateBackoff(t *testing.T) {