Merge branch 'main' into test-fix-kafka

Author: JoshVanL

.github/pull_request_template.md

@@ -14,4 +14,7 @@ Please make sure you've completed the relevant tasks for this PR, out of the fol
 
 * [ ] Code compiles correctly
 * [ ] Created/updated tests
-* [ ] Extended the documentation / Created issue in the https://github.com/dapr/docs/ repo: dapr/docs#_[issue number]_
+* [ ] Extended the documentation
+    * [ ] Created the dapr/docs PR: <insert PR link here>
+
+**Note:** We expect contributors to open a corresponding documentation PR in the [dapr/docs](https://github.com/dapr/docs/) repository. As the implementer, you are the best person to document your work! Implementation PRs will not be merged until the documentation PR is opened and ready for review.

.github/workflows/components-contrib-all.yml

@@ -68,19 +68,20 @@ jobs:
       GOLANGCI_LINT_VER: "v1.64.6"
     strategy:
       matrix:
-        os: [ubuntu-latest, windows-latest, macOS-latest]
+        # TODO: @nelson-parente Upgrade macos latest once dns is fixed.
+        os: [ubuntu-latest, windows-latest, macos-14]
         target_arch: [arm, amd64]
         include:
           - os: ubuntu-latest
             target_os: linux
           - os: windows-latest
             target_os: windows
-          - os: macOS-latest
+          - os: macos-14
             target_os: darwin
         exclude:
           - os: windows-latest
             target_arch: arm
-          - os: macOS-latest
+          - os: macos-14
             target_arch: arm
     steps:
       - name: Set default payload repo and ref

bindings/aws/kinesis/kinesis.go

@@ -156,8 +156,15 @@ func (a *AWSKinesis) Read(ctx context.Context, handler bindings.Handler) (err er
 	if a.closed.Load() {
 		return errors.New("binding is closed")
 	}
+
 	if a.metadata.KinesisConsumerMode == SharedThroughput {
-		a.worker = worker.NewWorker(a.recordProcessorFactory(ctx, handler), a.authProvider.Kinesis().WorkerCfg(ctx, a.streamName, a.consumerName, a.consumerMode))
+		// Configure the KCL worker with custom endpoints for LocalStack
+		config := a.authProvider.Kinesis().WorkerCfg(ctx, a.streamName, a.consumerName, a.consumerMode)
+		if a.metadata.Endpoint != "" {
+			config.KinesisEndpoint = a.metadata.Endpoint
+			config.DynamoDBEndpoint = a.metadata.Endpoint
+		}
+		a.worker = worker.NewWorker(a.recordProcessorFactory(ctx, handler), config)
 		err = a.worker.Start()
 		if err != nil {
 			return err

bindings/aws/s3/s3.go

@@ -24,20 +24,17 @@ import (
 	"net/http"
 	"os"
 	"reflect"
-	"slices"
 	"strings"
 	"time"
 
-	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/aws/aws-sdk-go/aws"
-	awsCommon "github.com/dapr/components-contrib/common/aws"
-	awsCommonAuth "github.com/dapr/components-contrib/common/aws/auth"
-
-	"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/aws/awserr"
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/aws/aws-sdk-go/service/s3/s3manager"
 	"github.com/google/uuid"
 
 	"github.com/dapr/components-contrib/bindings"
+	awsAuth "github.com/dapr/components-contrib/common/authentication/aws"
 	commonutils "github.com/dapr/components-contrib/common/utils"
 	"github.com/dapr/components-contrib/metadata"
 	"github.com/dapr/kit/logger"
@@ -63,12 +60,9 @@ const (
 
 // AWSS3 is a binding for an AWS S3 storage bucket.
 type AWSS3 struct {
-	metadata        *s3Metadata
-	logger          logger.Logger
-	s3Client        *s3.Client
-	s3Uploader      *manager.Uploader
-	s3Downloader    *manager.Downloader
-	s3PresignClient *s3.PresignClient
+	metadata     *s3Metadata
+	authProvider awsAuth.Provider
+	logger       logger.Logger
 }
 
 type s3Metadata struct {
@@ -112,42 +106,10 @@ func NewAWSS3(logger logger.Logger) bindings.OutputBinding {
 	return &AWSS3{logger: logger}
 }
 
-// Init does metadata parsing and connection creation.
-func (s *AWSS3) Init(ctx context.Context, metadata bindings.Metadata) error {
-	m, err := s.parseMetadata(metadata)
-	if err != nil {
-		return err
-	}
-	s.metadata = m
-
-	authOpts := awsCommonAuth.Options{
-		Logger: s.logger,
-
-		Properties: metadata.Properties,
-
-		Region:       m.Region,
-		Endpoint:     m.Endpoint,
-		AccessKey:    m.AccessKey,
-		SecretKey:    m.SecretKey,
-		SessionToken: m.SessionToken,
-	}
-
-	var configOptions []awsCommon.ConfigOption
-
-	var s3Options []func(options *s3.Options)
-
-	if s.metadata.DisableSSL {
-		s3Options = append(s3Options, func(options *s3.Options) {
-			options.EndpointOptions.DisableHTTPS = true
-		})
-	}
-
-	if !s.metadata.ForcePathStyle {
-		s3Options = append(s3Options, func(options *s3.Options) {
-			options.UsePathStyle = true
-		})
-	}
+func (s *AWSS3) getAWSConfig(opts awsAuth.Options) *aws.Config {
+	cfg := awsAuth.GetConfig(opts).WithS3ForcePathStyle(s.metadata.ForcePathStyle).WithDisableSSL(s.metadata.DisableSSL)
 
+	// Use a custom HTTP client to allow self-signed certs
 	if s.metadata.InsecureSSL {
 		customTransport := http.DefaultTransport.(*http.Transport).Clone()
 		customTransport.TLSClientConfig = &tls.Config{
@@ -157,27 +119,44 @@ func (s *AWSS3) Init(ctx context.Context, metadata bindings.Metadata) error {
 		client := &http.Client{
 			Transport: customTransport,
 		}
-		configOptions = append(configOptions, awsCommon.WithHTTPClient(client))
+		cfg = cfg.WithHTTPClient(client)
 
 		s.logger.Infof("aws s3: you are using 'insecureSSL' to skip server config verify which is unsafe!")
 	}
+	return cfg
+}
 
-	awsConfig, err := awsCommon.NewConfig(ctx, authOpts, configOptions...)
+// Init does metadata parsing and connection creation.
+func (s *AWSS3) Init(ctx context.Context, metadata bindings.Metadata) error {
+	m, err := s.parseMetadata(metadata)
 	if err != nil {
-		return fmt.Errorf("s3 binding error: failed to create AWS config: %w", err)
+		return err
 	}
+	s.metadata = m
 
-	s.s3Client = s3.NewFromConfig(awsConfig, s3Options...)
-
-	s.s3Uploader = manager.NewUploader(s.s3Client)
-	s.s3Downloader = manager.NewDownloader(s.s3Client)
-
-	s.s3PresignClient = s3.NewPresignClient(s.s3Client)
+	opts := awsAuth.Options{
+		Logger:       s.logger,
+		Properties:   metadata.Properties,
+		Region:       m.Region,
+		Endpoint:     m.Endpoint,
+		AccessKey:    m.AccessKey,
+		SecretKey:    m.SecretKey,
+		SessionToken: m.SessionToken,
+	}
+	// extra configs needed per component type
+	provider, err := awsAuth.NewProvider(ctx, opts, s.getAWSConfig(opts))
+	if err != nil {
+		return err
+	}
+	s.authProvider = provider
 
 	return nil
 }
 
 func (s *AWSS3) Close() error {
+	if s.authProvider != nil {
+		return s.authProvider.Close()
+	}
 	return nil
 }
 
@@ -236,25 +215,19 @@ func (s *AWSS3) create(ctx context.Context, req *bindings.InvokeRequest) (*bindi
 		r = b64.NewDecoder(b64.StdEncoding, r)
 	}
 
-	var storageClass types.StorageClass
+	var storageClass *string
 	if metadata.StorageClass != "" {
-		// assert storageclass exists in the types.storageclass.values() slice
-		storageClass = types.StorageClass(strings.ToUpper(metadata.StorageClass))
-		if !slices.Contains(storageClass.Values(), storageClass) {
-			return nil, fmt.Errorf("s3 binding error: invalid storage class '%s' provided", metadata.StorageClass)
-		}
+		storageClass = aws.String(metadata.StorageClass)
 	}
 
-	s3UploaderPutObjectInput := &s3.PutObjectInput{
+	resultUpload, err := s.authProvider.S3().Uploader.UploadWithContext(ctx, &s3manager.UploadInput{
 		Bucket:       ptr.Of(metadata.Bucket),
 		Key:          ptr.Of(key),
 		Body:         r,
 		ContentType:  contentType,
 		StorageClass: storageClass,
 		Tagging:      tagging,
-	}
-
-	resultUpload, err := s.s3Uploader.Upload(ctx, s3UploaderPutObjectInput)
+	})
 	if err != nil {
 		return nil, fmt.Errorf("s3 binding error: uploading failed: %w", err)
 	}
@@ -323,21 +296,16 @@ func (s *AWSS3) presignObject(ctx context.Context, bucket, key, ttl string) (str
 	if err != nil {
 		return "", fmt.Errorf("s3 binding error: cannot parse duration %s: %w", ttl, err)
 	}
-	s3GetObjectInput := &s3.GetObjectInput{
+	objReq, _ := s.authProvider.S3().S3.GetObjectRequest(&s3.GetObjectInput{
 		Bucket: ptr.Of(bucket),
 		Key:    ptr.Of(key),
-	}
-
-	presignedObjectRequest, err := s.s3PresignClient.PresignGetObject(
-		ctx,
-		s3GetObjectInput,
-		s3.WithPresignExpires(d),
-	)
+	})
+	url, err := objReq.Presign(d)
 	if err != nil {
 		return "", fmt.Errorf("s3 binding error: failed to presign URL: %w", err)
 	}
 
-	return presignedObjectRequest.URL, nil
+	return url, nil
 }
 
 func (s *AWSS3) get(ctx context.Context, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error) {
@@ -352,16 +320,16 @@ func (s *AWSS3) get(ctx context.Context, req *bindings.InvokeRequest) (*bindings
 	}
 
 	buff := &aws.WriteAtBuffer{}
-	_, err = s.s3Downloader.Download(ctx,
+	_, err = s.authProvider.S3().Downloader.DownloadWithContext(ctx,
 		buff,
 		&s3.GetObjectInput{
 			Bucket: ptr.Of(s.metadata.Bucket),
 			Key:    ptr.Of(key),
 		},
 	)
 	if err != nil {
-		var awsErr *types.NoSuchKey
-		if errors.As(err, &awsErr) {
+		var awsErr awserr.Error
+		if errors.As(err, &awsErr) && awsErr.Code() == s3.ErrCodeNoSuchKey {
 			return nil, errors.New("object not found")
 		}
 		return nil, fmt.Errorf("s3 binding error: error downloading S3 object: %w", err)
@@ -386,16 +354,16 @@ func (s *AWSS3) delete(ctx context.Context, req *bindings.InvokeRequest) (*bindi
 	if key == "" {
 		return nil, fmt.Errorf("s3 binding error: required metadata '%s' missing", metadataKey)
 	}
-	_, err := s.s3Client.DeleteObject(
+	_, err := s.authProvider.S3().S3.DeleteObjectWithContext(
 		ctx,
 		&s3.DeleteObjectInput{
 			Bucket: ptr.Of(s.metadata.Bucket),
 			Key:    ptr.Of(key),
 		},
 	)
 	if err != nil {
-		var awsErr *types.NoSuchKey
-		if errors.As(err, &awsErr) {
+		var awsErr awserr.Error
+		if errors.As(err, &awsErr) && awsErr.Code() == s3.ErrCodeNoSuchKey {
 			return nil, errors.New("object not found")
 		}
 		return nil, fmt.Errorf("s3 binding error: delete operation failed: %w", err)
@@ -415,9 +383,9 @@ func (s *AWSS3) list(ctx context.Context, req *bindings.InvokeRequest) (*binding
 	if payload.MaxResults < 1 {
 		payload.MaxResults = defaultMaxResults
 	}
-	result, err := s.s3Client.ListObjects(ctx, &s3.ListObjectsInput{
+	result, err := s.authProvider.S3().S3.ListObjectsWithContext(ctx, &s3.ListObjectsInput{
 		Bucket:    ptr.Of(s.metadata.Bucket),
-		MaxKeys:   ptr.Of(payload.MaxResults),
+		MaxKeys:   ptr.Of(int64(payload.MaxResults)),
 		Marker:    ptr.Of(payload.Marker),
 		Prefix:    ptr.Of(payload.Prefix),
 		Delimiter: ptr.Of(payload.Delimiter),

bindings/azure/servicebusqueues/servicebusqueues.go

@@ -62,7 +62,7 @@ func (a *AzureServiceBusQueues) Init(ctx context.Context, metadata bindings.Meta
 		return err
 	}
 
-	a.client, err = impl.NewClient(a.metadata, metadata.Properties)
+	a.client, err = impl.NewClient(a.metadata, metadata.Properties, a.logger)
 	if err != nil {
 		return err
 	}

bindings/gcp/bucket/bucket.go

@@ -110,13 +110,7 @@ func (g *GCPStorage) Init(ctx context.Context, metadata bindings.Metadata) error
 		return err
 	}
 
-	b, err := json.Marshal(m)
-	if err != nil {
-		return err
-	}
-
-	clientOptions := option.WithCredentialsJSON(b)
-	client, err := storage.NewClient(ctx, clientOptions)
+	client, err := g.getClient(ctx, m)
 	if err != nil {
 		return err
 	}
@@ -127,6 +121,41 @@ func (g *GCPStorage) Init(ctx context.Context, metadata bindings.Metadata) error
 	return nil
 }
 
+func (g *GCPStorage) getClient(ctx context.Context, m *gcpMetadata) (*storage.Client, error) {
+	var client *storage.Client
+	var err error
+
+	if m.Bucket == "" {
+		return nil, errors.New("missing property `bucket` in metadata")
+	}
+	if m.ProjectID == "" {
+		return nil, errors.New("missing property `project_id` in metadata")
+	}
+
+	// Explicit authentication
+	if m.PrivateKeyID != "" {
+		var b []byte
+		b, err = json.Marshal(m)
+		if err != nil {
+			return nil, err
+		}
+
+		clientOptions := option.WithCredentialsJSON(b)
+		client, err = storage.NewClient(ctx, clientOptions)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		// Implicit authentication, using GCP Application Default Credentials (ADC)
+		// Credentials search order: https://cloud.google.com/docs/authentication/application-default-credentials#order
+		client, err = storage.NewClient(ctx)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return client, nil
+}
+
 func (g *GCPStorage) parseMetadata(meta bindings.Metadata) (*gcpMetadata, error) {
 	m := gcpMetadata{}
 	err := kitmd.DecodeMetadata(meta.Properties, &m)