Merge pull request #912 from mackerelio/update-aws-sdk-go-v2

replace to aws-sdk-go-v2

Author: yseto

check-aws-cloudwatch-logs/lib/check-aws-cloudwatch-logs.go

@@ -13,10 +13,9 @@ import (
 	"strings"
 	"time"
 
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/aws/session"
-	"github.com/aws/aws-sdk-go/service/cloudwatchlogs"
-	"github.com/aws/aws-sdk-go/service/cloudwatchlogs/cloudwatchlogsiface"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
 	"github.com/jessevdk/go-flags"
 
 	"github.com/mackerelio/checkers"
@@ -50,15 +49,15 @@ func Do() {
 }
 
 type awsCloudwatchLogsPlugin struct {
-	Service   cloudwatchlogsiface.CloudWatchLogsAPI
+	Service   cloudwatchlogs.FilterLogEventsAPIClient
 	StateFile string
 	*logOpts
 }
 
-func newCloudwatchLogsPlugin(opts *logOpts, args []string) (*awsCloudwatchLogsPlugin, error) {
+func newCloudwatchLogsPlugin(ctx context.Context, opts *logOpts, args []string) (*awsCloudwatchLogsPlugin, error) {
 	var err error
 	p := &awsCloudwatchLogsPlugin{logOpts: opts}
-	p.Service, err = createService(opts)
+	p.Service, err = createService(ctx, opts)
 	if err != nil {
 		return nil, err
 	}
@@ -94,20 +93,22 @@ func getStateFile(stateDir, logGroupName, logStreamNamePrefix string, args []str
 	)
 }
 
-func createAWSConfig(opts *logOpts) *aws.Config {
-	conf := aws.NewConfig()
+func createCloudwatchlogsOptions(opts *logOpts) (optFns []func(*cloudwatchlogs.Options)) {
 	if opts.MaxRetries > 0 {
-		return conf.WithMaxRetries(opts.MaxRetries)
+		optFns = append(optFns, func(o *cloudwatchlogs.Options) {
+			o.RetryMaxAttempts = opts.MaxRetries
+		})
 	}
-	return conf
+	return
 }
 
-func createService(opts *logOpts) (*cloudwatchlogs.CloudWatchLogs, error) {
-	sess, err := session.NewSession()
+func createService(ctx context.Context, opts *logOpts) (*cloudwatchlogs.Client, error) {
+	cfg, err := config.LoadDefaultConfig(ctx)
 	if err != nil {
 		return nil, err
 	}
-	return cloudwatchlogs.New(sess, createAWSConfig(opts)), nil
+
+	return cloudwatchlogs.NewFromConfig(cfg, createCloudwatchlogsOptions(opts)...), nil
 }
 
 type logState struct {
@@ -134,9 +135,12 @@ func (p *awsCloudwatchLogsPlugin) collect(ctx context.Context, now time.Time) ([
 	if p.LogStreamNamePrefix != "" {
 		input.LogStreamNamePrefix = aws.String(p.LogStreamNamePrefix)
 	}
-	err = p.Service.FilterLogEventsPages(input, func(output *cloudwatchlogs.FilterLogEventsOutput, lastPage bool) bool {
-		if ctx.Err() != nil {
-			return false
+
+	paginator := cloudwatchlogs.NewFilterLogEventsPaginator(p.Service, input)
+	for paginator.HasMorePages() {
+		output, err := paginator.NextPage(ctx)
+		if err != nil {
+			return nil, err
 		}
 		for _, event := range output.Events {
 			messages = append(messages, *event.Message)
@@ -145,18 +149,11 @@ func (p *awsCloudwatchLogsPlugin) collect(ctx context.Context, now time.Time) ([
 			}
 		}
 		s.NextToken = output.NextToken
-		if lastPage {
-			s.NextToken = nil
-		}
-		err = p.saveState(s)
-		if err != nil {
-			return false
+
+		if err = p.saveState(s); err != nil {
+			return nil, err
 		}
 		time.Sleep(150 * time.Millisecond)
-		return true
-	})
-	if err != nil {
-		return nil, err
 	}
 	return messages, nil
 }
@@ -220,7 +217,7 @@ func run(ctx context.Context, args []string) *checkers.Checker {
 	if err != nil {
 		os.Exit(1)
 	}
-	p, err := newCloudwatchLogsPlugin(opts, args)
+	p, err := newCloudwatchLogsPlugin(ctx, opts, args)
 	if err != nil {
 		return checkers.Unknown(fmt.Sprint(err))
 	}

check-aws-cloudwatch-logs/lib/check-aws-cloudwatch-logs_test.go

@@ -7,41 +7,44 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
 	"os"
+	"path/filepath"
+	"strconv"
 	"testing"
 	"time"
 
 	"github.com/jessevdk/go-flags"
 	"github.com/mackerelio/checkers"
 	"github.com/stretchr/testify/assert"
 
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/service/cloudwatchlogs"
-	"github.com/aws/aws-sdk-go/service/cloudwatchlogs/cloudwatchlogsiface"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types"
 )
 
 type mockAWSCloudWatchLogsClient struct {
-	cloudwatchlogsiface.CloudWatchLogsAPI
+	cloudwatchlogs.FilterLogEventsAPIClient
 	outputs []*cloudwatchlogs.FilterLogEventsOutput
 }
 
-func (c *mockAWSCloudWatchLogsClient) FilterLogEventsPages(input *cloudwatchlogs.FilterLogEventsInput, fn func(*cloudwatchlogs.FilterLogEventsOutput, bool) bool) error {
-	for i, output := range c.outputs {
-		lastPage := i == len(c.outputs)-1
-		if !fn(output, lastPage) {
-			break
-		}
+func (c *mockAWSCloudWatchLogsClient) FilterLogEvents(ctx context.Context, input *cloudwatchlogs.FilterLogEventsInput, _ ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.FilterLogEventsOutput, error) {
+	if ctx.Err() != nil {
+		return nil, ctx.Err()
+	}
+
+	var pageNo = 0
+	if input.NextToken != nil {
+		pageNo, _ = strconv.Atoi(*input.NextToken)
 	}
-	return nil
+	return c.outputs[pageNo], nil
 }
 
-func createMockService() cloudwatchlogsiface.CloudWatchLogsAPI {
+func createMockService() cloudwatchlogs.FilterLogEventsAPIClient {
 	return &mockAWSCloudWatchLogsClient{
 		outputs: []*cloudwatchlogs.FilterLogEventsOutput{
 			{
 				NextToken: aws.String("1"),
-				Events: []*cloudwatchlogs.FilteredLogEvent{
+				Events: []types.FilteredLogEvent{
 					{
 						EventId:   aws.String("event-id-0"),
 						Message:   aws.String("message-0"),
@@ -56,7 +59,7 @@ func createMockService() cloudwatchlogsiface.CloudWatchLogsAPI {
 			},
 			{
 				NextToken: aws.String("2"),
-				Events: []*cloudwatchlogs.FilteredLogEvent{
+				Events: []types.FilteredLogEvent{
 					{
 						EventId:   aws.String("event-id-2"),
 						Message:   aws.String("message-2"),
@@ -75,7 +78,7 @@ func createMockService() cloudwatchlogsiface.CloudWatchLogsAPI {
 				},
 			},
 			{
-				Events: []*cloudwatchlogs.FilteredLogEvent{
+				Events: []types.FilteredLogEvent{
 					{
 						EventId:   aws.String("event-id-5"),
 						Message:   aws.String("message-5"),
@@ -88,13 +91,11 @@ func createMockService() cloudwatchlogsiface.CloudWatchLogsAPI {
 }
 
 func Test_cloudwatchLogsPlugin_collect(t *testing.T) {
-	file, _ := ioutil.TempFile("", "check-cloudwatch-logs-test-collect")
-	os.Remove(file.Name())
-	file.Close()
-	defer os.Remove(file.Name())
+	stateFile := filepath.Join(t.TempDir(), "check-cloudwatch-logs-test-collect")
+
 	p := &awsCloudwatchLogsPlugin{
 		Service:   createMockService(),
-		StateFile: file.Name(),
+		StateFile: stateFile,
 		logOpts: &logOpts{
 			LogGroupName: "test-group",
 		},
@@ -104,7 +105,7 @@ func Test_cloudwatchLogsPlugin_collect(t *testing.T) {
 		messages, err := p.collect(context.Background(), time.Unix(0, 0))
 		assert.Equal(t, err, nil, "err should be nil")
 		assert.Equal(t, len(messages), 6)
-		cnt, _ := ioutil.ReadFile(file.Name())
+		cnt, _ := os.ReadFile(stateFile)
 		var s logState
 		json.NewDecoder(bytes.NewReader(cnt)).Decode(&s)
 		assert.Equal(t, s, logState{StartTime: aws.Int64(5 + 1)})
@@ -115,7 +116,7 @@ func Test_cloudwatchLogsPlugin_collect(t *testing.T) {
 		cancel()
 
 		messages, err := p.collect(ctx, time.Unix(0, 0))
-		assert.Equal(t, err, nil, "err should be nil")
+		assert.NotEqual(t, err, nil, "err should be someting")
 		assert.Equal(t, len(messages), 0)
 	})
 }
@@ -221,25 +222,32 @@ func Test_cloudwatchLogsPlugin_options(t *testing.T) {
 	}
 }
 
-func Test_createAWSConfig(t *testing.T) {
+func Test_createCloudwatchlogsOptions(t *testing.T) {
 	tests := []struct {
-		opts *logOpts
-		want *aws.Config
+		opts   *logOpts
+		want   cloudwatchlogs.Options
+		length int
 	}{
 		{
-			opts: &logOpts{MaxRetries: 0},
-			want: aws.NewConfig(),
+			opts:   &logOpts{MaxRetries: 0},
+			length: 0,
 		},
 		{
-			opts: &logOpts{MaxRetries: 1},
-			want: aws.NewConfig().WithMaxRetries(1),
+			opts:   &logOpts{MaxRetries: 1},
+			want:   cloudwatchlogs.Options{RetryMaxAttempts: 1},
+			length: 1,
 		},
 	}
 
 	for i, tt := range tests {
 		t.Run(fmt.Sprintf("case:%d", i), func(t *testing.T) {
-			res := createAWSConfig(tt.opts)
-			assert.Equal(t, tt.want, res)
+			res := createCloudwatchlogsOptions(tt.opts)
+			assert.Equal(t, len(res), tt.length)
+			if tt.length > 0 {
+				opts := cloudwatchlogs.Options{}
+				res[0](&opts)
+				assert.Equal(t, tt.want, opts)
+			}
 		})
 	}
 }

check-aws-sqs-queue-size/lib/check-aws-sqs-queue-size.go

@@ -1,22 +1,31 @@
 package checkawssqsqueuesize
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"os"
+	"os/signal"
 	"strconv"
 
-	"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/aws-sdk-go/service/sqs"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
 	"github.com/jessevdk/go-flags"
 	"github.com/mackerelio/checkers"
 )
 
+// overwritten with syscall.SIGTERM on unix environment (see check-log_unix.go)
+var defaultSignal = os.Interrupt
+
 // Do the plugin
 func Do() {
-	ckr := run(os.Args[1:])
+	ctx, stop := signal.NotifyContext(context.Background(), defaultSignal)
+	defer stop()
+
+	ckr := run(ctx, os.Args[1:])
 	ckr.Name = "SQSQueueSize"
 	ckr.Exit()
 }
@@ -32,65 +41,68 @@ var opts struct {
 
 const sqsAttributeOfQueueSize = "ApproximateNumberOfMessages"
 
-func createService(region, awsAccessKeyID, awsSecretAccessKey string) (*sqs.SQS, error) {
-	sess, err := session.NewSession()
-	if err != nil {
-		return nil, err
-	}
-
-	config := aws.NewConfig()
+func createService(ctx context.Context, region, awsAccessKeyID, awsSecretAccessKey string) (*sqs.Client, error) {
+	var opts []func(*config.LoadOptions) error
 	if awsAccessKeyID != "" && awsSecretAccessKey != "" {
-		config = config.WithCredentials(credentials.NewStaticCredentials(awsAccessKeyID, awsSecretAccessKey, ""))
+		opts = append(opts, config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(awsAccessKeyID, awsSecretAccessKey, "")))
 	}
 	if region != "" {
-		config = config.WithRegion(region)
+		opts = append(opts, config.WithRegion(region))
 	}
-	return sqs.New(sess, config), nil
+
+	cfg, err := config.LoadDefaultConfig(ctx, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return sqs.NewFromConfig(cfg), nil
 }
 
-func getSqsQueueSize(region, awsAccessKeyID, awsSecretAccessKey, queueName string) (int, error) {
-	sqsClient, err := createService(region, awsAccessKeyID, awsSecretAccessKey)
+func getSqsQueueSize(ctx context.Context, region, awsAccessKeyID, awsSecretAccessKey, queueName string) (int, error) {
+	sqsClient, err := createService(ctx, region, awsAccessKeyID, awsSecretAccessKey)
 	if err != nil {
 		return -1, err
 	}
 
 	// Get queue url
-	q, err := sqsClient.GetQueueUrl(&sqs.GetQueueUrlInput{
+	q, err := sqsClient.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{
 		QueueName: aws.String(queueName),
 	})
 	if err != nil {
 		return -1, err
 	}
 
 	// Get queue attribute
-	attr, err := sqsClient.GetQueueAttributes(&sqs.GetQueueAttributesInput{
-		QueueUrl:       q.QueueUrl,
-		AttributeNames: []*string{aws.String(sqsAttributeOfQueueSize)},
+	attr, err := sqsClient.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{
+		QueueUrl: q.QueueUrl,
+		AttributeNames: []types.QueueAttributeName{
+			types.QueueAttributeNameApproximateNumberOfMessages,
+		},
 	})
 	if err != nil {
 		return -1, err
 	}
 
 	// Queue size
 	sizeStr, ok := attr.Attributes[sqsAttributeOfQueueSize]
-	if !ok || sizeStr == nil {
+	if !ok {
 		return -1, errors.New("attribute not found")
 	}
-	size, err := strconv.Atoi(*sizeStr)
+	size, err := strconv.Atoi(sizeStr)
 	if err != nil {
 		return -1, err
 	}
 
 	return size, nil
 }
 
-func run(args []string) *checkers.Checker {
+func run(ctx context.Context, args []string) *checkers.Checker {
 	_, err := flags.ParseArgs(&opts, args)
 	if err != nil {
 		os.Exit(1)
 	}
 
-	size, err := getSqsQueueSize(opts.Region, opts.AccessKeyID, opts.SecretAccessKey, opts.QueueName)
+	size, err := getSqsQueueSize(ctx, opts.Region, opts.AccessKeyID, opts.SecretAccessKey, opts.QueueName)
 	if err != nil {
 		return checkers.NewChecker(checkers.UNKNOWN, err.Error())
 	}

check-aws-sqs-queue-size/lib/check-aws-sqs-queue-size_unix.go

@@ -0,0 +1,9 @@
+package checkawssqsqueuesize
+
+import (
+	"syscall"
+)
+
+func init() {
+	defaultSignal = syscall.SIGTERM
+}