Merge branch 'main' into 3318-RavenDB-state-store-new

Author: nmalocic

.build-tools/go.mod

@@ -1,6 +1,6 @@
 module github.com/dapr/components-contrib/build-tools
 
-go 1.24.1
+go 1.24.4
 
 require (
 	github.com/dapr/components-contrib v0.0.0

.github/infrastructure/docker-compose-coherence.yml

@@ -4,7 +4,7 @@ services:
     environment:
       - coherence.management.http=all
       - coherence.management.http.port=30000
-      - Dcoherence.health.http.port=6676
+      - coherence.health.http.port=6676
       - coherence.wka=127.0.0.1
     ports:
       - 30000:30000

bindings/aws/s3/s3.go

@@ -24,17 +24,20 @@ 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"
-	"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"
+	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/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"
@@ -60,9 +63,12 @@ const (
 
 // AWSS3 is a binding for an AWS S3 storage bucket.
 type AWSS3 struct {
-	metadata     *s3Metadata
-	authProvider awsAuth.Provider
-	logger       logger.Logger
+	metadata        *s3Metadata
+	logger          logger.Logger
+	s3Client        *s3.Client
+	s3Uploader      *manager.Uploader
+	s3Downloader    *manager.Downloader
+	s3PresignClient *s3.PresignClient
 }
 
 type s3Metadata struct {
@@ -106,10 +112,42 @@ func NewAWSS3(logger logger.Logger) bindings.OutputBinding {
 	return &AWSS3{logger: logger}
 }
 
-func (s *AWSS3) getAWSConfig(opts awsAuth.Options) *aws.Config {
-	cfg := awsAuth.GetConfig(opts).WithS3ForcePathStyle(s.metadata.ForcePathStyle).WithDisableSSL(s.metadata.DisableSSL)
+// 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
+		})
+	}
 
-	// Use a custom HTTP client to allow self-signed certs
 	if s.metadata.InsecureSSL {
 		customTransport := http.DefaultTransport.(*http.Transport).Clone()
 		customTransport.TLSClientConfig = &tls.Config{
@@ -119,44 +157,27 @@ func (s *AWSS3) getAWSConfig(opts awsAuth.Options) *aws.Config {
 		client := &http.Client{
 			Transport: customTransport,
 		}
-		cfg = cfg.WithHTTPClient(client)
+		configOptions = append(configOptions, awsCommon.WithHTTPClient(client))
 
 		s.logger.Infof("aws s3: you are using 'insecureSSL' to skip server config verify which is unsafe!")
 	}
-	return cfg
-}
 
-// Init does metadata parsing and connection creation.
-func (s *AWSS3) Init(ctx context.Context, metadata bindings.Metadata) error {
-	m, err := s.parseMetadata(metadata)
+	awsConfig, err := awsCommon.NewConfig(ctx, authOpts, configOptions...)
 	if err != nil {
-		return err
+		return fmt.Errorf("s3 binding error: failed to create AWS config: %w", err)
 	}
-	s.metadata = m
 
-	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
+	s.s3Client = s3.NewFromConfig(awsConfig, s3Options...)
+
+	s.s3Uploader = manager.NewUploader(s.s3Client)
+	s.s3Downloader = manager.NewDownloader(s.s3Client)
+
+	s.s3PresignClient = s3.NewPresignClient(s.s3Client)
 
 	return nil
 }
 
 func (s *AWSS3) Close() error {
-	if s.authProvider != nil {
-		return s.authProvider.Close()
-	}
 	return nil
 }
 
@@ -215,19 +236,25 @@ func (s *AWSS3) create(ctx context.Context, req *bindings.InvokeRequest) (*bindi
 		r = b64.NewDecoder(b64.StdEncoding, r)
 	}
 
-	var storageClass *string
+	var storageClass types.StorageClass
 	if metadata.StorageClass != "" {
-		storageClass = aws.String(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)
+		}
 	}
 
-	resultUpload, err := s.authProvider.S3().Uploader.UploadWithContext(ctx, &s3manager.UploadInput{
+	s3UploaderPutObjectInput := &s3.PutObjectInput{
 		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)
 	}
@@ -296,16 +323,21 @@ 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)
 	}
-	objReq, _ := s.authProvider.S3().S3.GetObjectRequest(&s3.GetObjectInput{
+	s3GetObjectInput := &s3.GetObjectInput{
 		Bucket: ptr.Of(bucket),
 		Key:    ptr.Of(key),
-	})
-	url, err := objReq.Presign(d)
+	}
+
+	presignedObjectRequest, err := s.s3PresignClient.PresignGetObject(
+		ctx,
+		s3GetObjectInput,
+		s3.WithPresignExpires(d),
+	)
 	if err != nil {
 		return "", fmt.Errorf("s3 binding error: failed to presign URL: %w", err)
 	}
 
-	return url, nil
+	return presignedObjectRequest.URL, nil
 }
 
 func (s *AWSS3) get(ctx context.Context, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error) {
@@ -320,16 +352,16 @@ func (s *AWSS3) get(ctx context.Context, req *bindings.InvokeRequest) (*bindings
 	}
 
 	buff := &aws.WriteAtBuffer{}
-	_, err = s.authProvider.S3().Downloader.DownloadWithContext(ctx,
+	_, err = s.s3Downloader.Download(ctx,
 		buff,
 		&s3.GetObjectInput{
 			Bucket: ptr.Of(s.metadata.Bucket),
 			Key:    ptr.Of(key),
 		},
 	)
 	if err != nil {
-		var awsErr awserr.Error
-		if errors.As(err, &awsErr) && awsErr.Code() == s3.ErrCodeNoSuchKey {
+		var awsErr *types.NoSuchKey
+		if errors.As(err, &awsErr) {
 			return nil, errors.New("object not found")
 		}
 		return nil, fmt.Errorf("s3 binding error: error downloading S3 object: %w", err)
@@ -354,16 +386,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.authProvider.S3().S3.DeleteObjectWithContext(
+	_, err := s.s3Client.DeleteObject(
 		ctx,
 		&s3.DeleteObjectInput{
 			Bucket: ptr.Of(s.metadata.Bucket),
 			Key:    ptr.Of(key),
 		},
 	)
 	if err != nil {
-		var awsErr awserr.Error
-		if errors.As(err, &awsErr) && awsErr.Code() == s3.ErrCodeNoSuchKey {
+		var awsErr *types.NoSuchKey
+		if errors.As(err, &awsErr) {
 			return nil, errors.New("object not found")
 		}
 		return nil, fmt.Errorf("s3 binding error: delete operation failed: %w", err)
@@ -383,9 +415,9 @@ func (s *AWSS3) list(ctx context.Context, req *bindings.InvokeRequest) (*binding
 	if payload.MaxResults < 1 {
 		payload.MaxResults = defaultMaxResults
 	}
-	result, err := s.authProvider.S3().S3.ListObjectsWithContext(ctx, &s3.ListObjectsInput{
+	result, err := s.s3Client.ListObjects(ctx, &s3.ListObjectsInput{
 		Bucket:    ptr.Of(s.metadata.Bucket),
-		MaxKeys:   ptr.Of(int64(payload.MaxResults)),
+		MaxKeys:   ptr.Of(payload.MaxResults),
 		Marker:    ptr.Of(payload.Marker),
 		Prefix:    ptr.Of(payload.Prefix),
 		Delimiter: ptr.Of(payload.Delimiter),

bindings/kafka/metadata.yaml

@@ -372,3 +372,10 @@ metadata:
       - "range"
       - "sticky"
       - "roundrobin"
+  - name: excludeHeaderMetaRegex
+    type: string
+    required: false
+    description: |
+      A regular expression to exclude keys from being converted to/from headers from/to metadata to avoid unwanted downstream side effects.
+    example: '"^rawPayload|valueSchemaType$"'
+    default: '""'

bindings/zeebe/command/metadata.yaml

@@ -46,9 +46,10 @@ metadata:
     type: string
   - name: gatewayKeepAlive
     required: false
-    description: Sets how often keep alive messages should be sent to the gateway. Defaults to 45 seconds
+    description: Sets how often keep alive messages should be sent to the gateway.
     example: "45s"
     type: duration
+    default: "45s"
   - name: usePlainTextConnection	
     required: false
     description: Whether to use a plain text connection or not

bindings/zeebe/jobworker/metadata.yaml

@@ -20,9 +20,10 @@ metadata:
     type: string
   - name: gatewayKeepAlive
     required: false
-    description: Sets how often keep alive messages should be sent to the gateway. Defaults to 45 seconds
+    description: Sets how often keep alive messages should be sent to the gateway.
     example: "45s"
     type: duration
+    default: "45s"
   - name: usePlainTextConnection	
     required: false
     description: Whether to use a plain text connection or not
@@ -40,39 +41,44 @@ metadata:
     type: string
   - name: workerTimeout
     required: false
-    description: A job returned after this call will not be activated by another call until the timeout has been reached; defaults to 5 minutes
+    description: A job returned after this call will not be activated by another call until the timeout has been reached.
     example: "5m"
     type: duration
   - name: requestTimeout
     required: false
     description: The request will be completed when at least one job is activated or after the requestTimeout. If the requestTimeout = 0, a default timeout is used. If the requestTimeout < 0, long polling is disabled and the request is completed immediately, even when no job is activated. Defaults to 10 seconds
     example: "30s"
     type: duration
+    default: "10s"
   - name: jobType
     required: true
     description: the job type, as defined in the BPMN process (e.g. <zeebe:taskDefinition type=\"fetch-products\" />)
     example: "fetch-products"
     type: string
   - name: maxJobsActive
     required: false
-    description: Set the maximum number of jobs which will be activated for this worker at the same time. Defaults to 32
+    description: Set the maximum number of jobs which will be activated for this worker at the same time.
     example: "32"
     type: number
+    default: "32"
   - name: concurrency
     required: false
-    description: The maximum number of concurrent spawned goroutines to complete jobs. Defaults to 4
+    description: The maximum number of concurrent spawned goroutines to complete jobs.
     example: "4"
+    default: "4"
     type: number
   - name: pollInterval
     required: false
-    description: Set the maximal interval between polling for new jobs. Defaults to 100 milliseconds
+    description: Set the maximal interval between polling for new jobs.
     example: "100ms"
     type: duration
+    default: "100ms"
   - name: pollThreshold
     required: false
-    description: Set the threshold of buffered activated jobs before polling for new jobs, i.e. threshold * maxJobsActive. Defaults to 0.3
+    description: Set the threshold of buffered activated jobs before polling for new jobs, i.e. threshold * maxJobsActive.
     example: "0.3"
     type: number
+    default: "0.3"
   - name: fetchVariables
     required: false
     description: A list of variables to fetch as the job variables; if empty, all visible variables at the time of activation for the scope of the job will be returned