Bump up aws sdk to aws-sdk-go-v2

Signed-off-by: Daniel Jiang <jiangd@vmware.com>

test/util/providers/aws_utils.go

@@ -17,19 +17,24 @@ limitations under the License.
 package providers
 
 import (
+	"crypto/tls"
+	"crypto/x509"
 	"fmt"
+	"net/http"
 	"net/url"
 	"os"
-	"strconv"
 	"strings"
 
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/aws/credentials"
-	"github.com/aws/aws-sdk-go/aws/endpoints"
-	"github.com/aws/aws-sdk-go/aws/session"
-	"github.com/aws/aws-sdk-go/service/ec2"
-	"github.com/aws/aws-sdk-go/service/s3"
-	"github.com/aws/aws-sdk-go/service/s3/s3manager"
+	"github.com/aws/aws-sdk-go-v2/config"
+	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
+	s3manager "github.com/aws/aws-sdk-go-v2/feature/s3/manager"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+
+	"github.com/aws/aws-sdk-go-v2/service/ec2"
+	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
 	"github.com/pkg/errors"
 	"golang.org/x/net/context"
 
@@ -55,10 +60,26 @@ const (
 	enableSharedConfigKey        = "enableSharedConfig"
 )
 
-func newSessionOptions(config aws.Config, profile, caCert, credentialsFile, enableSharedConfig string) (session.Options, error) {
-	sessionOptions := session.Options{Config: config, Profile: profile}
-	if caCert != "" {
-		sessionOptions.CustomCABundle = strings.NewReader(caCert)
+func newAWSConfig(region, profile, credentialsFile string, insecureSkipTLSVerify bool, caCert string) (aws.Config, error) {
+	empty := aws.Config{}
+	client := awshttp.NewBuildableClient().WithTransportOptions(func(tr *http.Transport) {
+		if len(caCert) > 0 {
+			caCertPool := x509.NewCertPool()
+			caCertPool.AppendCertsFromPEM([]byte(caCert))
+			if tr.TLSClientConfig == nil {
+				tr.TLSClientConfig = &tls.Config{
+					RootCAs: caCertPool,
+				}
+			} else {
+				tr.TLSClientConfig.RootCAs = caCertPool
+			}
+		}
+		tr.TLSClientConfig.InsecureSkipVerify = insecureSkipTLSVerify
+	})
+	opts := []func(*config.LoadOptions) error{
+		config.WithRegion(region),
+		config.WithSharedConfigProfile(profile),
+		config.WithHTTPClient(client),
 	}
 
 	if credentialsFile == "" && os.Getenv("AWS_SHARED_CREDENTIALS_FILE") != "" {
@@ -68,104 +89,60 @@ func newSessionOptions(config aws.Config, profile, caCert, credentialsFile, enab
 	if credentialsFile != "" {
 		if _, err := os.Stat(credentialsFile); err != nil {
 			if os.IsNotExist(err) {
-				return session.Options{}, errors.Wrapf(err, "provided credentialsFile does not exist")
+				return empty, errors.Wrapf(err, "provided credentialsFile does not exist")
 			}
-			return session.Options{}, errors.Wrapf(err, "could not get credentialsFile info")
+			return empty, errors.Wrapf(err, "could not get credentialsFile info")
 		}
-		sessionOptions.SharedConfigFiles = append(sessionOptions.SharedConfigFiles, credentialsFile)
-		sessionOptions.SharedConfigState = session.SharedConfigEnable
+		opts = append(opts, config.WithSharedCredentialsFiles([]string{credentialsFile}),
+			config.WithSharedConfigFiles([]string{credentialsFile}))
 	}
 
-	return sessionOptions, nil
+	awsConfig, err := config.LoadDefaultConfig(context.Background(), opts...)
+	if err != nil {
+		return empty, errors.Wrapf(err, "could not load config")
+	}
+	if _, err := awsConfig.Credentials.Retrieve(context.Background()); err != nil {
+		return empty, errors.WithStack(err)
+	}
+
+	return awsConfig, nil
 }
 
-// takes AWS session options to create a new session
-func getSession(options session.Options) (*session.Session, error) {
-	sess, err := session.NewSessionWithOptions(options)
-	if err != nil {
-		return nil, errors.WithStack(err)
+func newS3Client(cfg aws.Config, url string, forcePathStyle bool) (*s3.Client, error) {
+	opts := []func(*s3.Options){
+		func(o *s3.Options) {
+			o.UsePathStyle = forcePathStyle
+		},
+	}
+	if url != "" {
+		if !IsValidS3URLScheme(url) {
+			return nil, errors.Errorf("Invalid s3 url %s, URL must be valid according to https://golang.org/pkg/net/url/#Parse and start with http:// or https://", url)
+		}
+		opts = append(opts, func(o *s3.Options) {
+			o.BaseEndpoint = aws.String(url)
+		})
 	}
 
-	if _, err := sess.Config.Credentials.Get(); err != nil {
-		return nil, errors.WithStack(err)
-	}
-
-	return sess, nil
+	return s3.NewFromConfig(cfg, opts...), nil
 }
 
 // GetBucketRegion returns the AWS region that a bucket is in, or an error
 // if the region cannot be determined.
 func GetBucketRegion(bucket string) (string, error) {
-	var region string
 
-	session, err := session.NewSession()
+	cfg, err := config.LoadDefaultConfig(context.Background())
 	if err != nil {
 		return "", errors.WithStack(err)
 	}
-
-	for _, partition := range endpoints.DefaultPartitions() {
-		for regionHint := range partition.Regions() {
-			region, _ = s3manager.GetBucketRegion(context.Background(), session, bucket, regionHint)
-
-			// we only need to try a single region hint per partition, so break after the first
-			break
-		}
-
-		if region != "" {
-			return region, nil
-		}
-	}
-
-	return "", errors.New("unable to determine bucket's region")
-}
-
-func (s AWSStorage) CreateSession(credentialProfile, credentialsFile, enableSharedConfig, caCert, bucket, bslPrefix, bslConfig string) (*session.Session, error) {
-	var err error
-	config := flag.NewMap()
-	config.Set(bslConfig)
-	region := config.Data()["region"]
-	objectsInput := s3.ListObjectsV2Input{}
-	objectsInput.Bucket = aws.String(bucket)
-	objectsInput.Delimiter = aws.String("/")
-	s3URL := ""
-	s3ForcePathStyleVal := ""
-	s3ForcePathStyle := false
-
-	if s3ForcePathStyleVal != "" {
-		if s3ForcePathStyle, err = strconv.ParseBool(s3ForcePathStyleVal); err != nil {
-			return nil, errors.Wrapf(err, "could not parse %s (expected bool)", s3ForcePathStyleKey)
-		}
-	}
-
-	// AWS (not an alternate S3-compatible API) and region not
-	// explicitly specified: determine the bucket's region
-	if s3URL == "" && region == "" {
-		var err error
-
-		region, err = GetBucketRegion(bucket)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	serverConfig, err := newAWSConfig(s3URL, region, s3ForcePathStyle)
+	client := s3.NewFromConfig(cfg)
+	region, err := s3manager.GetBucketRegion(context.Background(), client, bucket)
 	if err != nil {
-		return nil, err
+		return "", errors.WithStack(err)
 	}
-
-	sessionOptions, err := newSessionOptions(*serverConfig, credentialProfile, caCert, credentialsFile, enableSharedConfig)
-	if err != nil {
-		return nil, err
+	if region == "" {
+		return "", errors.New("unable to determine bucket's region")
 	}
-
-	serverSession, err := getSession(sessionOptions)
-
-	if err != nil {
-		fmt.Println(err)
-		return nil, err
-	}
-
-	return serverSession, nil
+	return region, nil
 }
 
 // IsValidS3URLScheme returns true if the scheme is http:// or https://
@@ -180,34 +157,9 @@ func IsValidS3URLScheme(s3URL string) bool {
 	}
 	return true
 }
-func newAWSConfig(url, region string, forcePathStyle bool) (*aws.Config, error) {
-	awsConfig := aws.NewConfig().
-		WithRegion(region).
-		WithS3ForcePathStyle(forcePathStyle)
 
-	if url != "" {
-		if !IsValidS3URLScheme(url) {
-			return nil, errors.Errorf("Invalid s3 url %s, URL must be valid according to https://golang.org/pkg/net/url/#Parse and start with http:// or https://", url)
-		}
-
-		awsConfig = awsConfig.WithEndpointResolver(
-			endpoints.ResolverFunc(func(service, region string, optFns ...func(*endpoints.Options)) (endpoints.ResolvedEndpoint, error) {
-				if service == s3.EndpointsID {
-					return endpoints.ResolvedEndpoint{
-						URL: url,
-					}, nil
-				}
-
-				return endpoints.DefaultResolver().EndpointFor(service, region, optFns...)
-			}),
-		)
-	}
-
-	return awsConfig, nil
-}
-
-func (s AWSStorage) ListItems(client *s3.S3, objectsV2Input *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) {
-	res, err := client.ListObjectsV2(objectsV2Input)
+func (s AWSStorage) ListItems(client *s3.Client, objectsV2Input *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) {
+	res, err := client.ListObjectsV2(context.Background(), objectsV2Input)
 	if err != nil {
 		return nil, err
 	}
@@ -215,14 +167,15 @@ func (s AWSStorage) ListItems(client *s3.S3, objectsV2Input *s3.ListObjectsV2Inp
 	return res, nil
 }
 
-func (s AWSStorage) DeleteItem(client *s3.S3, deleteObjectV2Input *s3.DeleteObjectInput) (*s3.DeleteObjectOutput, error) {
-	res, err := client.DeleteObject(deleteObjectV2Input)
+func (s AWSStorage) DeleteItem(client *s3.Client, deleteObjectV2Input *s3.DeleteObjectInput) (*s3.DeleteObjectOutput, error) {
+	res, err := client.DeleteObject(context.Background(), deleteObjectV2Input)
 	if err != nil {
 		return nil, err
 	}
 	fmt.Println(res)
 	return res, nil
 }
+
 func (s AWSStorage) IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string) (bool, error) {
 	config := flag.NewMap()
 	config.Set(bslConfig)
@@ -235,38 +188,42 @@ func (s AWSStorage) IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix
 	}
 
 	var err error
-	var s3Config *aws.Config
-	var sess *session.Session
+	var s3Config aws.Config
+	var s3Client *s3.Client
 	region := config.Data()["region"]
 	s3url := ""
-
+	if region == "" {
+		region, err = GetBucketRegion(bslBucket)
+		if err != nil {
+			return false, errors.Wrapf(err, "failed to get region for bucket %s", bslBucket)
+		}
+	}
 	if region == "minio" {
 		s3url = config.Data()["s3Url"]
-		s3Config = &aws.Config{
-			Credentials:      credentials.NewSharedCredentials(cloudCredentialsFile, ""),
-			Endpoint:         aws.String(s3url),
-			Region:           aws.String(region),
-			DisableSSL:       aws.Bool(true),
-			S3ForcePathStyle: aws.Bool(true),
+		s3Config, err = newAWSConfig(region, "", cloudCredentialsFile, true, "")
+		if err != nil {
+			return false, errors.Wrapf(err, "Failed to create AWS config of region %s", region)
 		}
-		sess, err = session.NewSession(s3Config)
+		s3Client, err = newS3Client(s3Config, s3url, true)
 	} else {
-		sess, err = s.CreateSession("", cloudCredentialsFile, "false", "", "", "", bslConfig)
+		s3Config, err = newAWSConfig(region, "", cloudCredentialsFile, false, "")
+		if err != nil {
+			return false, errors.Wrapf(err, "Failed to create AWS config of region %s", region)
+		}
+		s3Client, err = newS3Client(s3Config, s3url, true)
 	}
-
 	if err != nil {
-		return false, errors.Wrapf(err, fmt.Sprintf("Failed to create AWS session of region %s", region))
+		return false, errors.Wrapf(err, "failed to create S3 client of region %s", region)
 	}
-	svc := s3.New(sess)
 
-	bucketObjects, err := s.ListItems(svc, &objectsInput)
+	bucketObjects, err := s.ListItems(s3Client, &objectsInput)
 	if err != nil {
 		fmt.Println("Couldn't retrieve bucket items!")
 		return false, errors.Wrapf(err, "Couldn't retrieve bucket items")
 	}
 
 	for _, item := range bucketObjects.Contents {
-		fmt.Println(*item)
+		fmt.Println(item)
 	}
 	var backupNameInStorage string
 	for _, item := range bucketObjects.CommonPrefixes {
@@ -288,50 +245,64 @@ func (s AWSStorage) DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bslPr
 	config.Set(bslConfig)
 
 	var err error
-	var sess *session.Session
+	var s3Config aws.Config
+	var s3Client *s3.Client
 	region := config.Data()["region"]
 	s3url := ""
-
-	s3Config := &aws.Config{
-		Region:      aws.String(region),
-		Credentials: credentials.NewSharedCredentials(cloudCredentialsFile, ""),
+	if region == "" {
+		region, err = GetBucketRegion(bslBucket)
+		if err != nil {
+			return errors.Wrapf(err, "failed to get region for bucket %s", bslBucket)
+		}
 	}
 
 	if region == "minio" {
 		s3url = config.Data()["s3Url"]
-		s3Config = &aws.Config{
-			Credentials:      credentials.NewSharedCredentials(cloudCredentialsFile, ""),
-			Endpoint:         aws.String(s3url),
-			Region:           aws.String(region),
-			DisableSSL:       aws.Bool(true),
-			S3ForcePathStyle: aws.Bool(true),
+		s3Config, err = newAWSConfig(region, "", cloudCredentialsFile, true, "")
+		if err != nil {
+			return errors.Wrapf(err, "Failed to create AWS config of region %s", region)
 		}
-		sess, err = session.NewSession(s3Config)
+		s3Client, err = newS3Client(s3Config, s3url, true)
 	} else {
-		sess, err = s.CreateSession("", cloudCredentialsFile, "false", "", "", "", bslConfig)
+		s3Config, err = newAWSConfig(region, "", cloudCredentialsFile, false, "")
+		if err != nil {
+			return errors.Wrapf(err, "Failed to create AWS config of region %s", region)
+		}
+		s3Client, err = newS3Client(s3Config, s3url, false)
 	}
 
 	if err != nil {
-		return errors.Wrapf(err, fmt.Sprintf("Failed to create AWS session of region %s", region))
+		return errors.Wrapf(err, "Failed to create S3 client of region %s", region)
 	}
-
-	svc := s3.New(sess)
-
 	fullPrefix := strings.Trim(bslPrefix, "/") + "/" + strings.Trim(backupObject, "/") + "/"
-	iter := s3manager.NewDeleteListIterator(svc, &s3.ListObjectsInput{
+	listInput := &s3.ListObjectsV2Input{
 		Bucket: aws.String(bslBucket),
 		Prefix: aws.String(fullPrefix),
+	}
+	// list all keys
+	var objectIds []s3types.ObjectIdentifier
+	p := s3.NewListObjectsV2Paginator(s3Client, listInput)
+	for p.HasMorePages() {
+		page, err := p.NextPage(context.Background())
+		if err != nil {
+			return errors.Wrapf(err, "failed to list objects in bucket %s", bslBucket)
+		}
+		for _, obj := range page.Contents {
+			objectIds = append(objectIds, s3types.ObjectIdentifier{Key: aws.String(*obj.Key)})
+		}
+	}
+	_, err = s3Client.DeleteObjects(context.Background(), &s3.DeleteObjectsInput{
+		Bucket: aws.String(bslBucket),
+		Delete: &s3types.Delete{Objects: objectIds},
 	})
-
-	if err := s3manager.NewBatchDeleteWithClient(svc).Delete(aws.BackgroundContext(), iter); err != nil {
-		return errors.Wrapf(err, "Fail to delete object")
+	if err != nil {
+		return errors.Wrapf(err, "failed to delete objects from bucket %s", bslBucket)
 	}
 	fmt.Printf("Deleted object(s) from bucket: %s %s \n", bslBucket, fullPrefix)
 	return nil
 }
 
 func (s AWSStorage) IsSnapshotExisted(cloudCredentialsFile, bslConfig, backupObject string, snapshotCheck test.SnapshotCheckPoint) error {
-
 	config := flag.NewMap()
 	config.Set(bslConfig)
 	region := config.Data()["region"]
@@ -339,26 +310,22 @@ func (s AWSStorage) IsSnapshotExisted(cloudCredentialsFile, bslConfig, backupObj
 	if region == "minio" {
 		return errors.New("No snapshot for Minio provider")
 	}
-	sess, err := s.CreateSession("", cloudCredentialsFile, "false", "", "", "", bslConfig)
-
+	cfg, err := newAWSConfig(region, "", cloudCredentialsFile, false, "")
 	if err != nil {
-		fmt.Printf("Fail to create session with profile %s and config %s", cloudCredentialsFile, bslConfig)
-		return errors.Wrapf(err, "Fail to create session with profile %s and config %s", cloudCredentialsFile, bslConfig)
+		return errors.Wrapf(err, "Failed to create AWS config of region %s", region)
 	}
-	svc := ec2.New(sess)
-	params := &ec2.DescribeSnapshotsInput{
-		OwnerIds: []*string{aws.String("self")},
-		Filters: []*ec2.Filter{
+	ec2Client := ec2.NewFromConfig(cfg)
+	input := &ec2.DescribeSnapshotsInput{
+		OwnerIds: []string{"self"},
+		Filters: []ec2types.Filter{
 			{
-				Name: aws.String("tag:velero.io/backup"),
-				Values: []*string{
-					aws.String(backupObject),
-				},
+				Name:   aws.String("tag:velero.io/backup"),
+				Values: []string{backupObject},
 			},
 		},
 	}
 
-	result, err := svc.DescribeSnapshots(params)
+	result, err := ec2Client.DescribeSnapshots(context.Background(), input)
 	if err != nil {
 		fmt.Println(err)
 	}
@@ -386,52 +353,46 @@ func (s AWSStorage) GetMinioBucketSize(cloudCredentialsFile, bslBucket, bslPrefi
 	config := flag.NewMap()
 	config.Set(bslConfig)
 	region := config.Data()["region"]
-	s3url := config.Data()["s3Url"]
-	s3Config := &aws.Config{
-		Credentials:      credentials.NewSharedCredentials(cloudCredentialsFile, ""),
-		Endpoint:         aws.String(s3url),
-		Region:           aws.String(region),
-		DisableSSL:       aws.Bool(true),
-		S3ForcePathStyle: aws.Bool(true),
-	}
 	if region != "minio" {
 		return 0, errors.New("it only supported by minio")
 	}
-	sess, err := session.NewSession(s3Config)
+	s3url := config.Data()["s3Url"]
+	s3Config, err := newAWSConfig(region, "", cloudCredentialsFile, true, "")
 	if err != nil {
-		return 0, errors.Wrapf(err, "Error create config session")
+		return 0, errors.Wrapf(err, "failed to create AWS config of region %s", region)
 	}
+	s3Client, err := newS3Client(s3Config, s3url, true)
+	if err != nil {
+		return 0, errors.Wrapf(err, "failed to create S3 client of region %s", region)
+	}
+	/*
+		s3Config := &aws.Config{
+			Credentials:      credentials.NewSharedCredentials(cloudCredentialsFile, ""),
+			Endpoint:         aws.String(s3url),
+			Region:           aws.String(region),
+			DisableSSL:       aws.Bool(true),
+			S3ForcePathStyle: aws.Bool(true),
+		}
+	*/
 
-	svc := s3.New(sess)
 	var totalSize int64
-	var continuationToken *string
 	// Paginate through objects in the bucket
 	objectsInput := &s3.ListObjectsV2Input{
-		Bucket:            aws.String(bslBucket),
-		ContinuationToken: continuationToken,
+		Bucket: aws.String(bslBucket),
 	}
 	if bslPrefix != "" {
 		objectsInput.Prefix = aws.String(bslPrefix)
 	}
 
-	for {
-		resp, err := svc.ListObjectsV2(objectsInput)
+	p := s3.NewListObjectsV2Paginator(s3Client, objectsInput)
+	for p.HasMorePages() {
+		page, err := p.NextPage(context.Background())
 		if err != nil {
-			return 0, errors.Wrapf(err, "Error list objects")
+			return 0, errors.Wrapf(err, "failed to list objects in bucket %s", bslBucket)
 		}
-
-		// Process objects in the current response
-		for _, obj := range resp.Contents {
-			totalSize += *obj.Size
+		for _, obj := range page.Contents {
+			totalSize += obj.Size
 		}
-
-		// Check if there are more objects to retrieve
-		if !*resp.IsTruncated {
-			break
-		}
-
-		// Set the continuation token for the next page
-		continuationToken = resp.NextContinuationToken
 	}
 	return totalSize, nil
 }