remote_storage/s3: send only user tags to a remote object's tag set

WriteFile and UpdateFileMetadata dumped every extended attribute into the
S3 tagging header and PutObjectTagging. Seaweed-internal attributes such
as Seaweed-X-Amz-Acl carry values that are not legal S3 tags, so a
tag-aware remote now rejects the upload with InvalidTag. Mirror the
s3sink rule: only X-Amz-Tagging-* attributes are tags, prefix stripped,
and skip the tagging call when the tag set did not change.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/remote_storage/s3/s3_storage_client.go

@@ -9,6 +9,7 @@ import (
 	"net/url"
 	"os"
 	"reflect"
+	"sort"
 	"strings"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -25,6 +26,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
 	"github.com/seaweedfs/seaweedfs/weed/remote_storage"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/util/version"
 )
@@ -567,18 +569,12 @@ func (s *s3RemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocation,
 		u.Concurrency = s.uploadConcurrency()
 	})
 
-	// process tagging
-	tags := ""
 	var awsTags *string
 	// openstack swift doesn't support s3 object tagging
 	if s.conf.S3SupportTagging {
-		for k, v := range entry.Extended {
-			if len(tags) > 0 {
-				tags = tags + "&"
-			}
-			tags = tags + k + "=" + string(v)
+		if tagging := toTagging(entry.Extended); len(tagging.TagSet) > 0 {
+			awsTags = aws.String(s3TaggingHeader(tagging.TagSet))
 		}
-		awsTags = aws.String(tags)
 	}
 
 	// Upload the file to S3.
@@ -609,14 +605,23 @@ func (s *s3RemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocation,
 
 }
 
+// toTagging maps the entry's extended attributes to the remote object's tag
+// set. Only X-Amz-Tagging-* attributes are S3 tags; the rest of Extended is
+// Seaweed-internal metadata whose values are not legal tag values.
 func toTagging(attributes map[string][]byte) *s3.Tagging {
 	tagging := &s3.Tagging{}
 	for k, v := range attributes {
+		if !strings.HasPrefix(k, s3_constants.AmzObjectTaggingPrefix) {
+			continue
+		}
 		tagging.TagSet = append(tagging.TagSet, &s3.Tag{
-			Key:   aws.String(k),
+			Key:   aws.String(k[len(s3_constants.AmzObjectTaggingPrefix):]),
 			Value: aws.String(string(v)),
 		})
 	}
+	sort.Slice(tagging.TagSet, func(i, j int) bool {
+		return aws.StringValue(tagging.TagSet[i].Key) < aws.StringValue(tagging.TagSet[j].Key)
+	})
 	return tagging
 }
 
@@ -707,6 +712,9 @@ func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageL
 	}
 
 	tagging := toTagging(newEntry.Extended)
+	if reflect.DeepEqual(tagging.TagSet, toTagging(oldEntry.Extended).TagSet) {
+		return
+	}
 	if len(tagging.TagSet) > 0 {
 		_, err = s.conn.PutObjectTagging(&s3.PutObjectTaggingInput{
 			Bucket:  aws.String(loc.Bucket),

weed/remote_storage/s3/s3_storage_client_test.go

@@ -364,7 +364,7 @@ func TestS3UpdateFileMetadataSkipsTaggingWhenUnsupported(t *testing.T) {
 	client, rt := newRecordingS3Client(t, false)
 	loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/dir/file.bin"}
 	plain := &filer_pb.Entry{}
-	tagged := &filer_pb.Entry{Extended: map[string][]byte{"k1": []byte("v1")}}
+	tagged := &filer_pb.Entry{Extended: map[string][]byte{"X-Amz-Tagging-k1": []byte("v1")}}
 
 	_, err := client.UpdateFileMetadata(loc, plain, tagged)
 	require.NoError(t, err)
@@ -375,11 +375,22 @@ func TestS3UpdateFileMetadataSkipsTaggingWhenUnsupported(t *testing.T) {
 	require.Empty(t, rt.requests, "removing attributes must not send DeleteObjectTagging")
 }
 
+func TestS3UpdateFileMetadataIgnoresNonTagAttributes(t *testing.T) {
+	client, rt := newRecordingS3Client(t, true)
+	loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/dir/file.bin"}
+	old := &filer_pb.Entry{Extended: map[string][]byte{"k1": []byte("v1")}}
+	updated := &filer_pb.Entry{Extended: map[string][]byte{"k1": []byte("v2")}}
+
+	_, err := client.UpdateFileMetadata(loc, old, updated)
+	require.NoError(t, err)
+	require.Empty(t, rt.requests, "non-tag attribute changes must not send tagging calls")
+}
+
 func TestS3UpdateFileMetadataSendsTaggingWhenSupported(t *testing.T) {
 	client, rt := newRecordingS3Client(t, true)
 	loc := &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/dir/file.bin"}
 	plain := &filer_pb.Entry{}
-	tagged := &filer_pb.Entry{Extended: map[string][]byte{"k1": []byte("v1")}}
+	tagged := &filer_pb.Entry{Extended: map[string][]byte{"X-Amz-Tagging-k1": []byte("v1")}}
 
 	_, err := client.UpdateFileMetadata(loc, plain, tagged)
 	require.NoError(t, err)
@@ -395,6 +406,47 @@ func TestS3UpdateFileMetadataSendsTaggingWhenSupported(t *testing.T) {
 	require.Contains(t, rt.requests[0].URL.RawQuery, "tagging")
 }
 
+func TestS3WriteFileSendsOnlyTagAttributes(t *testing.T) {
+	rt := &captureRoundTripper{}
+	client := newS3ClientWithTransport(t, rt, true)
+	loc := &remote_pb.RemoteStorageLocation{
+		Name:   "test",
+		Bucket: "bucket",
+		Path:   "/dir/test.bin",
+	}
+	entry := &filer_pb.Entry{
+		Extended: map[string][]byte{
+			"X-Amz-Tagging-a b":  []byte("x=y"),
+			"Seaweed-X-Amz-Acl":  []byte(`[{"Permission":"FULL_CONTROL"}]`),
+			"Seaweed-X-Amz-ETag": []byte("etag-1"),
+			"Cache-Control":      []byte("no-cache"),
+		},
+	}
+
+	_, _ = client.WriteFile(loc, entry, bytes.NewReader([]byte("data")))
+
+	require.NotNil(t, rt.uploadReq, "uploader should have issued a PUT")
+	require.Equal(t, "a%20b=x%3Dy", rt.uploadReq.Header.Get("X-Amz-Tagging"))
+}
+
+func TestS3WriteFileOmitsTaggingWhenNoTagAttributes(t *testing.T) {
+	rt := &captureRoundTripper{}
+	client := newS3ClientWithTransport(t, rt, true)
+	loc := &remote_pb.RemoteStorageLocation{
+		Name:   "test",
+		Bucket: "bucket",
+		Path:   "/dir/test.bin",
+	}
+	entry := &filer_pb.Entry{
+		Extended: map[string][]byte{"Seaweed-X-Amz-Acl": []byte(`[{"Permission":"FULL_CONTROL"}]`)},
+	}
+
+	_, _ = client.WriteFile(loc, entry, bytes.NewReader([]byte("data")))
+
+	require.NotNil(t, rt.uploadReq, "uploader should have issued a PUT")
+	require.Empty(t, rt.uploadReq.Header.Get("X-Amz-Tagging"))
+}
+
 func TestS3WriteFileOmitsContentTypeWhenMimeMissing(t *testing.T) {
 	client, rt := newCapturingS3Client(t)
 	loc := &remote_pb.RemoteStorageLocation{