Feature/s3 add tagging to metadata (#3799)

Signed-off-by: adam6878 <adamshamis.dev@gmail.com>
Co-authored-by: Yaron Schneider <schneider.yaron@live.com>

bindings/aws/s3/s3.go

@@ -49,6 +49,7 @@ const (
 	metadataFilePath     = "filePath"
 	metadataPresignTTL   = "presignTTL"
 	metadataStorageClass = "storageClass"
+	metadataTags         = "tags"
 
 	metatadataContentType = "Content-Type"
 	metadataKey           = "key"
@@ -191,6 +192,15 @@ func (s *AWSS3) create(ctx context.Context, req *bindings.InvokeRequest) (*bindi
 	if contentTypeStr != "" {
 		contentType = &contentTypeStr
 	}
+
+	var tagging *string
+	if rawTags, ok := req.Metadata[metadataTags]; ok {
+		tagging, err = s.parseS3Tags(rawTags)
+		if err != nil {
+			return nil, fmt.Errorf("s3 binding error: parsing tags falied error: %w", err)
+		}
+	}
+
 	var r io.Reader
 	if metadata.FilePath != "" {
 		r, err = os.Open(metadata.FilePath)
@@ -209,12 +219,14 @@ func (s *AWSS3) create(ctx context.Context, req *bindings.InvokeRequest) (*bindi
 	if metadata.StorageClass != "" {
 		storageClass = aws.String(metadata.StorageClass)
 	}
+
 	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,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("s3 binding error: uploading failed: %w", err)
@@ -418,6 +430,26 @@ func (s *AWSS3) parseMetadata(md bindings.Metadata) (*s3Metadata, error) {
 	return &m, nil
 }
 
+// Helper for parsing s3 tags metadata
+func (s *AWSS3) parseS3Tags(raw string) (*string, error) {
+	tagEntries := strings.Split(raw, ",")
+	pairs := make([]string, 0, len(tagEntries))
+	for _, tagEntry := range tagEntries {
+		kv := strings.SplitN(strings.TrimSpace(tagEntry), "=", 2)
+		isInvalidTag := len(kv) != 2 || strings.TrimSpace(kv[0]) == "" || strings.TrimSpace(kv[1]) == ""
+		if isInvalidTag {
+			return nil, fmt.Errorf("invalid tag format: '%s' (expected key=value)", tagEntry)
+		}
+		pairs = append(pairs, fmt.Sprintf("%s=%s", strings.TrimSpace(kv[0]), strings.TrimSpace(kv[1])))
+	}
+
+	if len(pairs) == 0 {
+		return nil, nil
+	}
+
+	return aws.String(strings.Join(pairs, "&")), nil
+}
+
 // Helper to merge config and request metadata.
 func (metadata s3Metadata) mergeWithRequestMetadata(req *bindings.InvokeRequest) (s3Metadata, error) {
 	merged := metadata

bindings/aws/s3/s3_test.go

@@ -53,6 +53,24 @@ func TestParseMetadata(t *testing.T) {
 	})
 }
 
+func TestParseS3Tags(t *testing.T) {
+	t.Run("Has parsed s3 tags", func(t *testing.T) {
+		request := bindings.InvokeRequest{}
+		request.Metadata = map[string]string{
+			"decodeBase64": "yes",
+			"encodeBase64": "false",
+			"filePath":     "/usr/vader.darth",
+			"storageClass": "STANDARD_IA",
+			"tags":         "project=myproject,year=2024",
+		}
+		s3 := AWSS3{}
+		parsedTags, err := s3.parseS3Tags(request.Metadata["tags"])
+
+		require.NoError(t, err)
+		assert.Equal(t, "project=myproject&year=2024", *parsedTags)
+	})
+}
+
 func TestMergeWithRequestMetadata(t *testing.T) {
 	t.Run("Has merged metadata", func(t *testing.T) {
 		m := bindings.Metadata{}