reduce PutRetentionPolicy calls by checking existing policy (#1545)

Author: movence

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

@@ -30,6 +30,7 @@ type stubLogsService struct {
 	clg func(input *cloudwatchlogs.CreateLogGroupInput) (*cloudwatchlogs.CreateLogGroupOutput, error)
 	cls func(input *cloudwatchlogs.CreateLogStreamInput) (*cloudwatchlogs.CreateLogStreamOutput, error)
 	prp func(input *cloudwatchlogs.PutRetentionPolicyInput) (*cloudwatchlogs.PutRetentionPolicyOutput, error)
+	dlg func(input *cloudwatchlogs.DescribeLogGroupsInput) (*cloudwatchlogs.DescribeLogGroupsOutput, error)
 }
 
 func (s *stubLogsService) PutLogEvents(in *cloudwatchlogs.PutLogEventsInput) (*cloudwatchlogs.PutLogEventsOutput, error) {
@@ -60,6 +61,13 @@ func (s *stubLogsService) PutRetentionPolicy(in *cloudwatchlogs.PutRetentionPoli
 	return nil, nil
 }
 
+func (s *stubLogsService) DescribeLogGroups(in *cloudwatchlogs.DescribeLogGroupsInput) (*cloudwatchlogs.DescribeLogGroupsOutput, error) {
+	if s.dlg != nil {
+		return s.dlg(in)
+	}
+	return nil, nil
+}
+
 func TestAddSingleEvent_WithAccountId(t *testing.T) {
 	t.Parallel()
 	var wg sync.WaitGroup

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

@@ -19,6 +19,7 @@ type cloudWatchLogsService interface {
 	CreateLogStream(input *cloudwatchlogs.CreateLogStreamInput) (*cloudwatchlogs.CreateLogStreamOutput, error)
 	CreateLogGroup(input *cloudwatchlogs.CreateLogGroupInput) (*cloudwatchlogs.CreateLogGroupOutput, error)
 	PutRetentionPolicy(input *cloudwatchlogs.PutRetentionPolicyInput) (*cloudwatchlogs.PutRetentionPolicyOutput, error)
+	DescribeLogGroups(input *cloudwatchlogs.DescribeLogGroupsInput) (*cloudwatchlogs.DescribeLogGroupsOutput, error)
 }
 
 type Sender interface {

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

@@ -40,6 +40,11 @@ func (m *mockLogsService) PutRetentionPolicy(input *cloudwatchlogs.PutRetentionP
 	return args.Get(0).(*cloudwatchlogs.PutRetentionPolicyOutput), args.Error(1)
 }
 
+func (m *mockLogsService) DescribeLogGroups(input *cloudwatchlogs.DescribeLogGroupsInput) (*cloudwatchlogs.DescribeLogGroupsOutput, error) {
+	args := m.Called(input)
+	return args.Get(0).(*cloudwatchlogs.DescribeLogGroupsOutput), args.Error(1)
+}
+
 type mockTargetManager struct {
 	mock.Mock
 }

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

@@ -4,7 +4,9 @@
 package pusher
 
 import (
+	"fmt"
 	"sync"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
@@ -13,6 +15,15 @@ import (
 	"github.com/aws/amazon-cloudwatch-agent/sdk/service/cloudwatchlogs"
 )
 
+const (
+	retentionChannelSize = 100
+	// max wait time with backoff and jittering:
+	// 0 + 2.4 + 4.8 + 9.6 + 10 ~= 26.8 sec
+	baseRetryDelay      = 1 * time.Second
+	maxRetryDelayTarget = 10 * time.Second
+	numBackoffRetries   = 5
+)
+
 type Target struct {
 	Group, Stream, Class string
 	Retention            int
@@ -29,69 +40,100 @@ type targetManager struct {
 	// cache of initialized targets
 	cache map[Target]struct{}
 	mu    sync.Mutex
+	dlg   chan Target
+	prp   chan Target
 }
 
 func NewTargetManager(logger telegraf.Logger, service cloudWatchLogsService) TargetManager {
-	return &targetManager{
+	tm := &targetManager{
 		logger:  logger,
 		service: service,
 		cache:   make(map[Target]struct{}),
+		dlg:     make(chan Target, retentionChannelSize),
+		prp:     make(chan Target, retentionChannelSize),
 	}
+
+	go tm.processDescribeLogGroup()
+	go tm.processPutRetentionPolicy()
+	return tm
 }
 
 // InitTarget initializes a Target if it hasn't been initialized before.
 func (m *targetManager) InitTarget(target Target) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if _, ok := m.cache[target]; !ok {
-		err := m.createLogGroupAndStream(target)
+		newGroup, err := m.createLogGroupAndStream(target)
 		if err != nil {
 			return err
 		}
-		m.PutRetentionPolicy(target)
+		if target.Retention > 0 {
+			if newGroup {
+				m.logger.Debugf("sending new log group %v to prp channel", target.Group)
+				m.prp <- target
+			} else {
+				m.logger.Debugf("sending existing log group %v to dlg channel", target.Group)
+				m.dlg <- target
+			}
+		}
 		m.cache[target] = struct{}{}
 	}
 	return nil
 }
 
-func (m *targetManager) createLogGroupAndStream(t Target) error {
+func (m *targetManager) PutRetentionPolicy(target Target) {
+	// new pusher will call this so start with dlg
+	if target.Retention > 0 {
+		m.logger.Debugf("sending log group %v to dlg channel by pusher", target.Group)
+		m.dlg <- target
+	}
+}
+
+func (m *targetManager) createLogGroupAndStream(t Target) (bool, error) {
 	err := m.createLogStream(t)
 	if err == nil {
-		return nil
+		return false, nil
 	}
 
 	m.logger.Debugf("creating stream fail due to : %v", err)
+	newGroup := false
 	if awsErr, ok := err.(awserr.Error); ok && awsErr.Code() == cloudwatchlogs.ErrCodeResourceNotFoundException {
 		err = m.createLogGroup(t)
+		newGroup = true
 
 		// attempt to create stream again if group created successfully.
 		if err == nil {
-			m.logger.Debugf("successfully created log group %v. Retrying log stream %v", t.Group, t.Stream)
+			m.logger.Debugf("retrying log stream %v", t.Stream)
 			err = m.createLogStream(t)
 		} else {
 			m.logger.Debugf("creating group fail due to : %v", err)
 		}
 	}
 
 	if awsErr, ok := err.(awserr.Error); ok && awsErr.Code() == cloudwatchlogs.ErrCodeResourceAlreadyExistsException {
-		m.logger.Debugf("Resource was already created. %v\n", err)
-		return nil // if the log group or log stream already exist, this is not worth returning an error for
+		m.logger.Debugf("resource was already created. %v\n", err)
+		return false, nil
 	}
 
-	return err
+	return newGroup, err
 }
 
 func (m *targetManager) createLogGroup(t Target) error {
-	var err error
+	var input *cloudwatchlogs.CreateLogGroupInput
 	if t.Class != "" {
-		_, err = m.service.CreateLogGroup(&cloudwatchlogs.CreateLogGroupInput{
+		input = &cloudwatchlogs.CreateLogGroupInput{
 			LogGroupName:  &t.Group,
 			LogGroupClass: &t.Class,
-		})
+		}
 	} else {
-		_, err = m.service.CreateLogGroup(&cloudwatchlogs.CreateLogGroupInput{
+		input = &cloudwatchlogs.CreateLogGroupInput{
 			LogGroupName: &t.Group,
-		})
+		}
+	}
+	_, err := m.service.CreateLogGroup(input)
+	if err == nil {
+		m.logger.Debugf("successfully created log group %v", t.Group)
+		return nil
 	}
 	return err
 }
@@ -109,26 +151,88 @@ func (m *targetManager) createLogStream(t Target) error {
 	return err
 }
 
-// PutRetentionPolicy tries to set the retention policy for a log group. Does not retry on failure.
-func (m *targetManager) PutRetentionPolicy(t Target) {
-	if t.Retention > 0 {
-		i := aws.Int64(int64(t.Retention))
-		putRetentionInput := &cloudwatchlogs.PutRetentionPolicyInput{
-			LogGroupName:    &t.Group,
-			RetentionInDays: i,
+func (m *targetManager) processDescribeLogGroup() {
+	for target := range m.dlg {
+		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
 		}
-		_, err := m.service.PutRetentionPolicy(putRetentionInput)
-		if err != nil {
-			// since this gets called both before we start pushing logs, and after we first attempt
-			// to push a log to a non-existent log group, we don't want to dirty the log with an error
-			// if the error is that the log group doesn't exist (yet).
-			if awsErr, ok := err.(awserr.Error); ok && awsErr.Code() == cloudwatchlogs.ErrCodeResourceNotFoundException {
-				m.logger.Debugf("Log group %v not created yet: %v", t.Group, err)
-			} else {
-				m.logger.Errorf("Unable to put retention policy for log group %v: %v ", t.Group, err)
+	}
+}
+
+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
 			}
-		} else {
-			m.logger.Debugf("successfully updated log retention policy for log group %v", t.Group)
+			return int(*group.RetentionInDays), nil
+		}
+	}
+
+	return 0, fmt.Errorf("log group %v not found", target.Group)
+}
+
+func (m *targetManager) processPutRetentionPolicy() {
+	for target := range m.prp {
+		var updated bool
+		for attempt := 0; attempt < numBackoffRetries; attempt++ {
+			err := m.updateRetentionPolicy(target)
+			if err == nil {
+				updated = true
+				break
+			}
+
+			m.logger.Debugf("retrying to update retention policy for target (%v) %v: %v", attempt, target, err)
+			time.Sleep(m.calculateBackoff(attempt))
 		}
+
+		if !updated {
+			m.logger.Errorf("failed to update retention policy for target %v after %d attempts", target, numBackoffRetries)
+		}
+	}
+}
+
+func (m *targetManager) updateRetentionPolicy(target Target) error {
+	input := &cloudwatchlogs.PutRetentionPolicyInput{
+		LogGroupName:    aws.String(target.Group),
+		RetentionInDays: aws.Int64(int64(target.Retention)),
+	}
+
+	_, err := m.service.PutRetentionPolicy(input)
+	if err != nil {
+		return fmt.Errorf("put retention policy failed: %w", err)
+	}
+	m.logger.Debugf("successfully updated retention policy for log group %v", target.Group)
+	return nil
+}
+
+func (m *targetManager) calculateBackoff(retryCount int) time.Duration {
+	delay := baseRetryDelay
+	if retryCount < numBackoffRetries {
+		delay = baseRetryDelay * time.Duration(1<<int64(retryCount))
+	}
+	if delay > maxRetryDelayTarget {
+		delay = maxRetryDelayTarget
 	}
+	return time.Duration(seededRand.Int63n(int64(delay/2)) + int64(delay/2))
 }