This is how you can work with cloudwatch sdk and handle ThrottlingException

aws_error.go

package aws

// Error implements AWS Error interface
type Error struct {
	error
	ErrorCode     string
	ErrorMessage  string
	OriginalError error
}

// Code Returns the short phrase depicting the classification of the error.
func (a Error) Code() string {
	return a.ErrorCode
}

// Message Returns the error details message.
func (a Error) Message() string {
	return a.ErrorMessage
}

// OrigErr Returns the original error if one was set.  Nil is returned if not set.
func (a Error) OrigErr() error {
	return a.OriginalError
}

cloudwatch.go

package aws
func isThrottlingError(err error) bool {
	if aerr, ok := err.(awserr.Error); ok {
		return aerr.Code() == "ThrottlingException"
	}
	return false
}

func describeLogGroups(cloudwatchClient cloudwatchlogsiface.CloudWatchLogsAPI, request cloudwatchlogs.DescribeLogGroupsInput, numberOfThrottlingRetries int) (*cloudwatchlogs.DescribeLogGroupsOutput, error) {
	res, err := cloudwatchClient.DescribeLogGroupsRequest(&request).Send()
	if err != nil {
		if isThrottlingError(err) && numberOfThrottlingRetries != 10 {
			sleep := time.Second
			jitter := time.Duration(rand.Int63n(int64(sleep)))
			sleep = sleep + jitter/2

			time.Sleep(sleep)
			return describeLogGroups(cloudwatchClient, request, numberOfThrottlingRetries+1)
		}
		return nil, err

	}
	return res, nil
}

func describeSubscriptionFilter(cloudwatchClient cloudwatchlogsiface.CloudWatchLogsAPI, request cloudwatchlogs.DescribeSubscriptionFiltersInput, numberOfThrottlingRetries int) (*cloudwatchlogs.DescribeSubscriptionFiltersOutput, error) {
	res, err := cloudwatchClient.DescribeSubscriptionFiltersRequest(&request).Send()
	if err != nil {
		if isThrottlingError(err) && numberOfThrottlingRetries != 10 {
			sleep := time.Second
			jitter := time.Duration(rand.Int63n(int64(sleep)))
			sleep = sleep + jitter/2

			time.Sleep(sleep)
			return describeSubscriptionFilter(cloudwatchClient, request, numberOfThrottlingRetries+1)
		}
		return nil, err

	}
	return res, nil
}

func putSubscriptionFilter(cloudwatchClient cloudwatchlogsiface.CloudWatchLogsAPI, logGroupName *string, destArn string, roleName string, numberOfThrottlingRetries int) error {
	var input cloudwatchlogs.PutSubscriptionFilterInput
	if roleName != "" {
		input = cloudwatchlogs.PutSubscriptionFilterInput{
			LogGroupName:   logGroupName,
			DestinationArn: aws.String(destArn),
			FilterName:     aws.String("Datadog"),
			FilterPattern:  aws.String(""),
			RoleArn:        aws.String(roleName),
		}
	} else {
		input = cloudwatchlogs.PutSubscriptionFilterInput{
			LogGroupName:   logGroupName,
			DestinationArn: aws.String(destArn),
			FilterName:     aws.String("Datadog"),
			FilterPattern:  aws.String(""),
		}
	}
	_, err := cloudwatchClient.PutSubscriptionFilterRequest(&input).Send()

	if err != nil {
		if isThrottlingError(err) && numberOfThrottlingRetries != 10 {
			sleep := time.Second
			jitter := time.Duration(rand.Int63n(int64(sleep)))
			sleep = sleep + jitter/2

			time.Sleep(sleep)
			return putSubscriptionFilter(cloudwatchClient, logGroupName, destArn, roleName, numberOfThrottlingRetries+1)
		}
		return err

	}
	return nil
}

func deleteSubscriptionFilter(cloudwatchClient cloudwatchlogsiface.CloudWatchLogsAPI, logGroupName *string, filterName string, numberOfThrottlingRetries int) error {
	_, err := cloudwatchClient.DeleteSubscriptionFilterRequest(&cloudwatchlogs.DeleteSubscriptionFilterInput{
		LogGroupName: logGroupName,
		FilterName:   aws.String(filterName),
	}).Send()

	if err != nil {
		if isThrottlingError(err) && numberOfThrottlingRetries != 10 {
			sleep := time.Second
			jitter := time.Duration(rand.Int63n(int64(sleep)))
			sleep = sleep + jitter/2

			time.Sleep(sleep)
			return deleteSubscriptionFilter(cloudwatchClient, logGroupName, filterName, numberOfThrottlingRetries+1)
		}
		return err

	}
	return nil
}

cloudwatch_mock.go

package aws

import (
	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/cloudwatchlogsiface"
)
// CloudWatchLogsClientMock is a mock for cloudwatch logs client
type CloudWatchLogsClientMock struct {
	cloudwatchlogsiface.CloudWatchLogsAPI
	Psfr func(input *cloudwatchlogs.PutSubscriptionFilterInput) cloudwatchlogs.PutSubscriptionFilterRequest
	Dlgr func(input *cloudwatchlogs.DescribeLogGroupsInput) cloudwatchlogs.DescribeLogGroupsRequest
	Dsfr func(input *cloudwatchlogs.DescribeSubscriptionFiltersInput) cloudwatchlogs.DescribeSubscriptionFiltersRequest
	Delsfr func(input *cloudwatchlogs.DeleteSubscriptionFilterInput) cloudwatchlogs.DeleteSubscriptionFilterRequest
}
// PutSubscriptionFilterRequest mocks cloudwatch logs client method
func (c CloudWatchLogsClientMock) PutSubscriptionFilterRequest(input *cloudwatchlogs.PutSubscriptionFilterInput) cloudwatchlogs.PutSubscriptionFilterRequest {
	return c.Psfr(input)
}

// DescribeLogGroupsRequest mocks cloudwatch logs client method
func (c CloudWatchLogsClientMock) DescribeLogGroupsRequest(input *cloudwatchlogs.DescribeLogGroupsInput) cloudwatchlogs.DescribeLogGroupsRequest {
	return c.Dlgr(input)
}

// DescribeSubscriptionFiltersRequest mocks cloudwatch logs client method
func (c CloudWatchLogsClientMock) DescribeSubscriptionFiltersRequest(input *cloudwatchlogs.DescribeSubscriptionFiltersInput) cloudwatchlogs.DescribeSubscriptionFiltersRequest {
	return c.Dsfr(input)
}

// DeleteSubscriptionFilterRequest mocks cloudwatch logs client method
func (c CloudWatchLogsClientMock) DeleteSubscriptionFilterRequest(input *cloudwatchlogs.DeleteSubscriptionFilterInput) cloudwatchlogs.DeleteSubscriptionFilterRequest {
	return c.Delsfr(input)
}

test_aws_cloudwatch.go

package aws

func TestPutSubscriptionFilter(t *testing.T) {
	t.Run("I put subscribe cloudwatch log group to lambda from the first time when there are no errors", func(t *testing.T) {
		logGroupName := "sls-dev-datadog-adapter-process-send-logs-to-datadog"
		destinationArn := "arn:aws:lambda:eu-west-1:12123123:function:sls-dev-datadog-adapter-process-send-logs-to-datadog"
		filterName := "Datadog"
		filterPattern := ""
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Psfr: func(input *cloudwatchlogs.PutSubscriptionFilterInput) cloudwatchlogs.PutSubscriptionFilterRequest {
				assert.Equal(t, logGroupName, *input.LogGroupName, "Log Group Name")
				assert.Equal(t, *aws.String(destinationArn), *input.DestinationArn, "Destination ARN")
				assert.Equal(t, *aws.String(filterName), *input.FilterName, "Filter Name")
				assert.Equal(t, *aws.String(filterPattern), *input.FilterPattern, "Filter Pattern")
				return cloudwatchlogs.PutSubscriptionFilterRequest{
					Request: &aws.Request{
						Data: &cloudwatchlogs.PutSubscriptionFilterOutput{},
					},
					Input: input,
				}
			},
		}
		roleName := GetDatadogSubscriptionRoleName(AccountDetails{
			Alias:     "dev",
			Region:    "eu-west-1",
			AccountID: "123123",
		})
		err := putSubscriptionFilter(cloudwatchLogClientMock, &logGroupName, destinationArn, roleName, 0)
		assert.Nil(t, err, "Error")
	})
}

func TestPutSubscriptionFilterAfterRetry(t *testing.T) {
	t.Run("I put subscribe cloudwatch log group to lambda from the second try because of Throttling error", func(t *testing.T) {
		logGroupName := "sls-dev-datadog-adapter-process-send-logs-to-datadog"
		destinationArn := "arn:aws:lambda:eu-west-1:12123123:function:sls-dev-datadog-adapter-process-send-logs-to-datadog"
		filterName := "Datadog"
		filterPattern := ""
		errorsCount := 0
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Psfr: func(input *cloudwatchlogs.PutSubscriptionFilterInput) cloudwatchlogs.PutSubscriptionFilterRequest {
				assert.Equal(t, logGroupName, *input.LogGroupName, "Log Group Name")
				assert.Equal(t, *aws.String(destinationArn), *input.DestinationArn, "Destination ARN")
				assert.Equal(t, *aws.String(filterName), *input.FilterName, "Filter Name")
				assert.Equal(t, *aws.String(filterPattern), *input.FilterPattern, "Filter Pattern")
				var err error
				if errorsCount == 0 {
					errorsCount++
					err = Error{
						error:         errors.New("I have failed you"),
						ErrorCode:     "ThrottlingException",
						ErrorMessage:  "Sorry i have failed you",
						OriginalError: nil,
					}
				} else {
					err = nil
				}
				return cloudwatchlogs.PutSubscriptionFilterRequest{
					Request: &aws.Request{
						Data:  &cloudwatchlogs.PutSubscriptionFilterOutput{},
						Error: err,
					},
					Input: input,
				}
			},
		}
		roleName := GetDatadogSubscriptionRoleName(AccountDetails{
			Alias:     "dev",
			Region:    "eu-west-1",
			AccountID: "123123",
		})
		err := putSubscriptionFilter(cloudwatchLogClientMock, &logGroupName, destinationArn, roleName, 0)
		assert.Nil(t, err, "Error")
	})
}

func TestPutSubscriptionFilterFailAfterMaxRetries(t *testing.T) {
	t.Run("I fail to subscribe cloudwatch log group to lambda because too many ThrottlingException retries", func(t *testing.T) {
		logGroupName := "sls-dev-datadog-adapter-process-send-logs-to-datadog"
		destinationArn := "arn:aws:lambda:eu-west-1:12123123:function:sls-dev-datadog-adapter-process-send-logs-to-datadog"
		filterName := "Datadog"
		filterPattern := ""
		errorsCount := 0
		errorMsg := "I have failed you"
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Psfr: func(input *cloudwatchlogs.PutSubscriptionFilterInput) cloudwatchlogs.PutSubscriptionFilterRequest {
				assert.Equal(t, logGroupName, *input.LogGroupName, "Log Group Name")
				assert.Equal(t, *aws.String(destinationArn), *input.DestinationArn, "Destination ARN")
				assert.Equal(t, *aws.String(filterName), *input.FilterName, "Filter Name")
				assert.Equal(t, *aws.String(filterPattern), *input.FilterPattern, "Filter Pattern")
				var err error
				if errorsCount <= 10 {
					errorsCount++

					err = Error{
						error:         errors.New(errorMsg),
						ErrorCode:     "ThrottlingException",
						ErrorMessage:  "Sorry i have failed you",
					}
				} else {
					err = nil
				}
				return cloudwatchlogs.PutSubscriptionFilterRequest{
					Request: &aws.Request{
						Data:  &cloudwatchlogs.PutSubscriptionFilterOutput{},
						Error: err,
					},
					Input: input,
				}
			},
		}
		roleName := GetDatadogSubscriptionRoleName(AccountDetails{
			Alias:     "dev",
			Region:    "eu-west-1",
			AccountID: "123123",
		})
		err := putSubscriptionFilter(cloudwatchLogClientMock, &logGroupName, destinationArn, roleName, 0)
		assert.NotNil(t, err, "Error")
		assert.Equal(t, errorMsg, err.Error(), "Error message")
		assert.Equal(t, 11, errorsCount, "Errors count")
	})
}
func TestPutSubscriptionFilterFailAfterNonThrottlingError(t *testing.T) {
	t.Run("I do not retry to subscribe cloudwatch group to lambda if error is not ThrottlingException", func(t *testing.T) {
		logGroupName := "sls-dev-datadog-adapter-process-send-logs-to-datadog"
		destinationArn := "arn:aws:lambda:eu-west-1:12123123:function:sls-dev-datadog-adapter-process-send-logs-to-datadog"
		filterName := "Datadog"
		filterPattern := ""
		errorsCount := 0
		errorMsg := "I have failed you"
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Psfr: func(input *cloudwatchlogs.PutSubscriptionFilterInput) cloudwatchlogs.PutSubscriptionFilterRequest {
				assert.Equal(t, logGroupName, *input.LogGroupName, "Log Group Name")
				assert.Equal(t, *aws.String(destinationArn), *input.DestinationArn, "Destination ARN")
				assert.Equal(t, *aws.String(filterName), *input.FilterName, "Filter Name")
				assert.Equal(t, *aws.String(filterPattern), *input.FilterPattern, "Filter Pattern")
				var err error
				if errorsCount <= 5 {
					errorsCount++
					err = Error{
						error:         errors.New(errorMsg),
						ErrorCode:     "SomeBadError",
						ErrorMessage:  "Sorry i have failed you",
					}
				} else {
					err = nil
				}
				return cloudwatchlogs.PutSubscriptionFilterRequest{
					Request: &aws.Request{
						Data:  &cloudwatchlogs.PutSubscriptionFilterOutput{},
						Error: err,
					},
					Input: input,
				}
			},
		}
		roleName := GetDatadogSubscriptionRoleName(AccountDetails{
			Alias:     "dev",
			Region:    "eu-west-1",
			AccountID: "123123",
		})
		err := putSubscriptionFilter(cloudwatchLogClientMock, &logGroupName, destinationArn, roleName, 0)
		assert.NotNil(t, err, "Error")
		assert.Equal(t, errorMsg, err.Error(), "Error message")
		assert.Equal(t, 1, errorsCount, "Errors count")
	})
}

func TestDescribeLogGroups(t *testing.T) {
	t.Run("I describe log groups from the first time when there are no errors", func(t *testing.T) {
		logGroupNamePrefix := "sls-dev-datadog-adapter"
		logGroups := []cloudwatchlogs.LogGroup{{LogGroupName: aws.String(logGroupNamePrefix + "bla")}}
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Dlgr: func(input *cloudwatchlogs.DescribeLogGroupsInput) cloudwatchlogs.DescribeLogGroupsRequest {
				assert.Equal(t, logGroupNamePrefix, *input.LogGroupNamePrefix, "Log Group Prefix")
				return cloudwatchlogs.DescribeLogGroupsRequest{
					Request: &aws.Request{
						Data: &cloudwatchlogs.DescribeLogGroupsOutput{
							LogGroups: logGroups,
						},
					},
					Input: input,
				}
			},
		}

		describeLogGroupsInput := cloudwatchlogs.DescribeLogGroupsInput{
			LogGroupNamePrefix: aws.String(logGroupNamePrefix),
		}
		res, err := describeLogGroups(cloudwatchLogClientMock, describeLogGroupsInput, 0)
		assert.Equal(t, logGroups, res.LogGroups, "Log Groups")
		assert.Nil(t, err, "Error")
	})
}

func TestDescribeLogGroupsAfterRetry(t *testing.T) {
	t.Run("I describe log groups from the second try because of Throttling error", func(t *testing.T) {
		logGroupNamePrefix := "sls-dev-datadog-adapter"
		logGroups := []cloudwatchlogs.LogGroup{{LogGroupName: aws.String(logGroupNamePrefix + "bla")}}
		errorsCount := 0
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Dlgr: func(input *cloudwatchlogs.DescribeLogGroupsInput) cloudwatchlogs.DescribeLogGroupsRequest {
				assert.Equal(t, logGroupNamePrefix, *input.LogGroupNamePrefix, "Log Group Prefix")
				var err error
				if errorsCount == 0 {
					errorsCount++
					err = Error{
						error:         errors.New("I have failed you"),
						ErrorCode:     "ThrottlingException",
						ErrorMessage:  "Sorry i have failed you",
					}
				} else {
					err = nil
				}
				return cloudwatchlogs.DescribeLogGroupsRequest{
					Request: &aws.Request{
						Data: &cloudwatchlogs.DescribeLogGroupsOutput{
							LogGroups: logGroups,
						},
						Error: err,
					},
					Input: input,
				}
			},
		}

		describeLogGroupsInput := cloudwatchlogs.DescribeLogGroupsInput{
			LogGroupNamePrefix: aws.String(logGroupNamePrefix),
		}
		res, err := describeLogGroups(cloudwatchLogClientMock, describeLogGroupsInput, 0)
		assert.Equal(t, logGroups, res.LogGroups, "Log Groups")
		assert.Nil(t, err, "Error")
	})
}

func TestDescribeLogGroupsFailAfterMaxRetries(t *testing.T) {
	t.Run("I fail to describe log group because too many ThrottlingException retries", func(t *testing.T) {
		logGroupNamePrefix := "sls-dev-datadog-adapter"
		logGroups := []cloudwatchlogs.LogGroup{{LogGroupName: aws.String(logGroupNamePrefix + "bla")}}
		errorsCount := 0
		errorMsg := "I have failed you"
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Dlgr: func(input *cloudwatchlogs.DescribeLogGroupsInput) cloudwatchlogs.DescribeLogGroupsRequest {
				assert.Equal(t, logGroupNamePrefix, *input.LogGroupNamePrefix, "Log Group Prefix")
				var err error
				if errorsCount <= 10 {
					errorsCount++

					err = Error{
						error:         errors.New(errorMsg),
						ErrorCode:     "ThrottlingException",
						ErrorMessage:  "Sorry i have failed you",
					}
				} else {
					err = nil
				}
				return cloudwatchlogs.DescribeLogGroupsRequest{
					Request: &aws.Request{
						Data: &cloudwatchlogs.DescribeLogGroupsOutput{
							LogGroups: logGroups,
						},
						Error: err,
					},
					Input: input,
				}
			},
		}

		describeLogGroupsInput := cloudwatchlogs.DescribeLogGroupsInput{
			LogGroupNamePrefix: aws.String(logGroupNamePrefix),
		}
		res, err := describeLogGroups(cloudwatchLogClientMock, describeLogGroupsInput, 0)
		assert.Nil(t, res, "Response")
		assert.NotNil(t, err, "Error")
		assert.Equal(t, errorMsg, err.Error(), "Error message")
		assert.Equal(t, 11, errorsCount, "Errors count")
	})
}
func TestDescribeLogGroupsFailAfterNonThrottlingError(t *testing.T) {
	t.Run("I do not retry to subscribe cloudwatch group to lambda if error is not ThrottlingException", func(t *testing.T) {
		logGroupNamePrefix := "sls-dev-datadog-adapter"
		logGroups := []cloudwatchlogs.LogGroup{{LogGroupName: aws.String(logGroupNamePrefix + "bla")}}
		errorsCount := 0
		errorMsg := "I have failed you"
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Dlgr: func(input *cloudwatchlogs.DescribeLogGroupsInput) cloudwatchlogs.DescribeLogGroupsRequest {
				assert.Equal(t, logGroupNamePrefix, *input.LogGroupNamePrefix, "Log Group Prefix")
				var err error
				if errorsCount <= 5 {
					errorsCount++
					err = Error{
						error:         errors.New(errorMsg),
						ErrorCode:     "SomeBadError",
						ErrorMessage:  "Sorry i have failed you",
						OriginalError: nil,
					}
				} else {
					err = nil
				}
				return cloudwatchlogs.DescribeLogGroupsRequest{
					Request: &aws.Request{
						Data: &cloudwatchlogs.DescribeLogGroupsOutput{
							LogGroups: logGroups,
						},
						Error: err,
					},
					Input: input,
				}
			},
		}

		describeLogGroupsInput := cloudwatchlogs.DescribeLogGroupsInput{
			LogGroupNamePrefix: aws.String(logGroupNamePrefix),
		}
		res, err := describeLogGroups(cloudwatchLogClientMock, describeLogGroupsInput, 0)
		assert.Nil(t, res, "Response")
		assert.NotNil(t, err, "Error")
		assert.Equal(t, errorMsg, err.Error(), "Error message")
		assert.Equal(t, 1, errorsCount, "Errors count")
	})
}


func TestDescribeSubscriptionFilterRequest(t *testing.T) {
	t.Run("I describe subscription filter request from the first time when there are no errors", func(t *testing.T) {
		logGroupName := "sls-dev-datadog-adapter"
		filterName := "datadog-adapter"
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Dsfr: func(input *cloudwatchlogs.DescribeSubscriptionFiltersInput) cloudwatchlogs.DescribeSubscriptionFiltersRequest {
				assert.Equal(t, logGroupName, *input.LogGroupName, "Log Group")
				return cloudwatchlogs.DescribeSubscriptionFiltersRequest{
					Request: &aws.Request{
						Data: &cloudwatchlogs.DescribeSubscriptionFiltersOutput{
							SubscriptionFilters: []cloudwatchlogs.SubscriptionFilter{{FilterName: aws.String(filterName)}},
						},
					},
					Input: input,
				}
			},
		}

		describeSubscriptionFilterInput := cloudwatchlogs.DescribeSubscriptionFiltersInput{
			LogGroupName: aws.String(logGroupName),
		}
		res, err := describeSubscriptionFilter(cloudwatchLogClientMock, describeSubscriptionFilterInput, 0)
		assert.Equal(t, []cloudwatchlogs.SubscriptionFilter{{FilterName: aws.String(filterName)}}, res.SubscriptionFilters, "Log Groups")
		assert.Nil(t, err, "Error")
	})
}

func TestDescribeSubscriptionFilterRequestAfterRetry(t *testing.T) {
	t.Run("I describe subscription filter from the second try because of Throttling error", func(t *testing.T) {
		logGroupName := "sls-dev-datadog-adapter"
		filterName := "datadog-adapter"
		errorsCount := 0
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Dsfr: func(input *cloudwatchlogs.DescribeSubscriptionFiltersInput) cloudwatchlogs.DescribeSubscriptionFiltersRequest {
				assert.Equal(t, logGroupName, *input.LogGroupName, "Log Group")
				var err error
				if errorsCount == 0 {
					errorsCount++
					err = Error{
						error:         errors.New("I have failed you"),
						ErrorCode:     "ThrottlingException",
						ErrorMessage:  "Sorry i have failed you",
					}
				} else {
					err = nil
				}
				return cloudwatchlogs.DescribeSubscriptionFiltersRequest{
					Request: &aws.Request{
						Data: &cloudwatchlogs.DescribeSubscriptionFiltersOutput{
							SubscriptionFilters: []cloudwatchlogs.SubscriptionFilter{{FilterName: aws.String(filterName)}},
						},
						Error: err,
					},
					Input: input,
				}
			},
		}

		describeSubscriptionFilterInput := cloudwatchlogs.DescribeSubscriptionFiltersInput{
			LogGroupName: aws.String(logGroupName),
		}
		res, err := describeSubscriptionFilter(cloudwatchLogClientMock, describeSubscriptionFilterInput, 0)
		assert.Equal(t, []cloudwatchlogs.SubscriptionFilter{{FilterName: aws.String(filterName)}}, res.SubscriptionFilters, "Log Groups")
		assert.Nil(t, err, "Error")
	})
}

func TestDescribeSubscriptionFilterRequestFailAfterMaxRetries(t *testing.T) {
	t.Run("I fail to describe subscription filter because too many ThrottlingException retries", func(t *testing.T) {
		logGroupName := "sls-dev-datadog-adapter"
		filterName := "datadog-adapter"
		errorsCount := 0
		errorMsg := "I have failed you"
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Dsfr: func(input *cloudwatchlogs.DescribeSubscriptionFiltersInput) cloudwatchlogs.DescribeSubscriptionFiltersRequest {
				assert.Equal(t, logGroupName, *input.LogGroupName, "Log Group")
				var err error
				if errorsCount <= 10 {
					errorsCount++

					err = Error{
						error:         errors.New(errorMsg),
						ErrorCode:     "ThrottlingException",
						ErrorMessage:  "Sorry i have failed you",
					}
				} else {
					err = nil
				}
				return cloudwatchlogs.DescribeSubscriptionFiltersRequest{
					Request: &aws.Request{
						Data: &cloudwatchlogs.DescribeSubscriptionFiltersOutput{
							SubscriptionFilters: []cloudwatchlogs.SubscriptionFilter{{FilterName: aws.String(filterName)}},
						},
						Error: err,
					},
					Input: input,
				}
			},
		}

		describeSubscriptionFilterInput := cloudwatchlogs.DescribeSubscriptionFiltersInput{
			LogGroupName: aws.String(logGroupName),
		}
		res, err := describeSubscriptionFilter(cloudwatchLogClientMock, describeSubscriptionFilterInput, 0)
		assert.Nil(t, res, "Response")
		assert.NotNil(t, err, "Error")
		assert.Equal(t, errorMsg, err.Error(), "Error message")
		assert.Equal(t, 11, errorsCount, "Errors count")
	})
}
func TestDescribeSubscriptionFilterRequestFailAfterNonThrottlingError(t *testing.T) {
	t.Run("I do not retry to subscribe cloudwatch group to lambda if error is not ThrottlingException", func(t *testing.T) {
		logGroupName := "sls-dev-datadog-adapter"
		filterName := "datadog-adapter"
		errorsCount := 0
		errorMsg := "I have failed you"
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Dsfr: func(input *cloudwatchlogs.DescribeSubscriptionFiltersInput) cloudwatchlogs.DescribeSubscriptionFiltersRequest {
				assert.Equal(t, logGroupName, *input.LogGroupName, "Log Group")
				var err error
				if errorsCount <= 5 {
					errorsCount++
					err = Error{
						error:         errors.New(errorMsg),
						ErrorCode:     "SomeBadError",
						ErrorMessage:  "Sorry i have failed you",
						OriginalError: nil,
					}
				} else {
					err = nil
				}
				return cloudwatchlogs.DescribeSubscriptionFiltersRequest{
					Request: &aws.Request{
						Data: &cloudwatchlogs.DescribeSubscriptionFiltersOutput{
							SubscriptionFilters: []cloudwatchlogs.SubscriptionFilter{{FilterName: aws.String(filterName)}},
						},
						Error: err,
					},
					Input: input,
				}
			},
		}

		describeSubscriptionFilterInput := cloudwatchlogs.DescribeSubscriptionFiltersInput{
			LogGroupName: aws.String(logGroupName),
		}
		res, err := describeSubscriptionFilter(cloudwatchLogClientMock, describeSubscriptionFilterInput, 0)
		assert.Nil(t, res, "Response")
		assert.NotNil(t, err, "Error")
		assert.Equal(t, errorMsg, err.Error(), "Error message")
		assert.Equal(t, 1, errorsCount, "Errors count")
	})
}



func TestDeleteSubscriptionFilterRequest(t *testing.T) {
	t.Run("I delete subscription filter request from the first time when there are no errors", func(t *testing.T) {
		logGroupName := "sls-dev-datadog-adapter"
		filterName := "datadog-adapter"
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Delsfr: func(input *cloudwatchlogs.DeleteSubscriptionFilterInput) cloudwatchlogs.DeleteSubscriptionFilterRequest {
				assert.Equal(t, logGroupName, *input.LogGroupName, "Log Group")
				assert.Equal(t, filterName, *input.FilterName, "Filter Name")
				return cloudwatchlogs.DeleteSubscriptionFilterRequest{
					Request: &aws.Request{
						Data: &cloudwatchlogs.DeleteSubscriptionFilterOutput{},
					},
					Input: input,
				}
			},
		}

		err := deleteSubscriptionFilter(cloudwatchLogClientMock, &logGroupName, filterName, 0)
		assert.Nil(t, err, "Error")
	})
}

func TestDeleteSubscriptionFilterRequestAfterRetry(t *testing.T) {
	t.Run("I delete subscription filter from the second try because of Throttling error", func(t *testing.T) {
		logGroupName := "sls-dev-datadog-adapter"
		filterName := "datadog-adapter"
		errorsCount := 0
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Delsfr: func(input *cloudwatchlogs.DeleteSubscriptionFilterInput) cloudwatchlogs.DeleteSubscriptionFilterRequest {
				assert.Equal(t, logGroupName, *input.LogGroupName, "Log Group")
				var err error
				if errorsCount == 0 {
					errorsCount++
					err = Error{
						error:         errors.New("I have failed you"),
						ErrorCode:     "ThrottlingException",
						ErrorMessage:  "Sorry i have failed you",
					}
				} else {
					err = nil
				}
				return cloudwatchlogs.DeleteSubscriptionFilterRequest{
					Request: &aws.Request{
						Data: &cloudwatchlogs.DeleteSubscriptionFilterOutput{},
						Error: err,
					},

					Input: input,
				}
			},
		}

		err := deleteSubscriptionFilter(cloudwatchLogClientMock, &logGroupName, filterName, 0)
		assert.Nil(t, err, "Error")
	})
}

func TestDeleteSubscriptionFilterRequestFailAfterMaxRetries(t *testing.T) {
	t.Run("I fail to delete subscription filter because too many ThrottlingException retries", func(t *testing.T) {
		logGroupName := "sls-dev-datadog-adapter"
		filterName := "datadog-adapter"
		errorsCount := 0
		errorMsg := "I have failed you"
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Delsfr: func(input *cloudwatchlogs.DeleteSubscriptionFilterInput) cloudwatchlogs.DeleteSubscriptionFilterRequest {
				assert.Equal(t, logGroupName, *input.LogGroupName, "Log Group")
				var err error
				if errorsCount <= 10 {
					errorsCount++

					err = Error{
						error:         errors.New(errorMsg),
						ErrorCode:     "ThrottlingException",
						ErrorMessage:  "Sorry i have failed you",
					}
				} else {
					err = nil
				}
				return cloudwatchlogs.DeleteSubscriptionFilterRequest{
					Request: &aws.Request{
						Data: &cloudwatchlogs.DeleteSubscriptionFilterOutput{},
						Error: err,
					},

					Input: input,
				}
			},
		}

		err := deleteSubscriptionFilter(cloudwatchLogClientMock, &logGroupName, filterName, 0)
		assert.NotNil(t, err, "Error")
		assert.Equal(t, errorMsg, err.Error(), "Error message")
		assert.Equal(t, 11, errorsCount, "Errors count")
	})
}
func TestDeleteSubscriptionFilterRequestFailAfterNonThrottlingError(t *testing.T) {
	t.Run("I do not retry to delete subsription filter if error is not ThrottlingException", func(t *testing.T) {
		logGroupName := "sls-dev-datadog-adapter"
		filterName := "datadog-adapter"
		errorsCount := 0
		errorMsg := "I have failed you"
		cloudwatchLogClientMock := CloudWatchLogsClientMock{
			Delsfr: func(input *cloudwatchlogs.DeleteSubscriptionFilterInput) cloudwatchlogs.DeleteSubscriptionFilterRequest {
				assert.Equal(t, logGroupName, *input.LogGroupName, "Log Group")
				var err error
				if errorsCount <= 5 {
					errorsCount++
					err = Error{
						error:         errors.New(errorMsg),
						ErrorCode:     "SomeBadError",
						ErrorMessage:  "Sorry i have failed you",
						OriginalError: nil,
					}
				} else {
					err = nil
				}
				return cloudwatchlogs.DeleteSubscriptionFilterRequest{
					Request: &aws.Request{
						Data: &cloudwatchlogs.DeleteSubscriptionFilterOutput{},
						Error: err,
					},

					Input: input,
				}
			},
		}

		err := deleteSubscriptionFilter(cloudwatchLogClientMock, &logGroupName, filterName, 0)
		assert.NotNil(t, err, "Error")
		assert.Equal(t, errorMsg, err.Error(), "Error message")
		assert.Equal(t, 1, errorsCount, "Errors count")

	})
}