mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-01 09:50:39 +01:00
s3 sink: forward user metadata to the destination object (#12286)
* s3 sink: forward user metadata to the destination object filer.backup and filer.sync to an S3 sink copied the object body and tags but dropped the X-Amz-Meta- attributes the S3 gateway stores with each object, so the destination object came back without its user metadata. Send them as the upload's metadata, keyed by the lowercased suffix. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3 sink: send mtime as metadata without writing it into the entry The sink stored X-Amz-Meta-Mtime into the replicated entry's Extended map, modifying the caller's entry, and since tags were limited to X-Amz-Tagging- keys it was never sent anywhere. Add it to the upload metadata instead, keeping an mtime the source object already carries. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3 sink: RFC 2047 encode metadata values that are not printable ASCII The S3 gateway keeps user metadata values as raw bytes. Forwarded as is, a UTF-8 value fails SigV4 on AWS and a control character makes the HTTP client reject the request, and filer.backup retries such an object forever. Encode those values as RFC 2047 encoded words, which S3 decodes on write. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3 sink: upload without user metadata past the S3 2 KB limit The S3 gateway does not cap user metadata, but S3 rejects more than 2 KB of it, and filer.backup would retry that object forever. Upload such an object without its metadata and log a warning, as the sink did before metadata was forwarded. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3 sink: skip metadata keys that are not valid header names Filer gRPC clients can store any Extended key. An X-Amz-Meta- key with a space or control character makes the HTTP client reject the whole upload, which filer.backup retries forever. Leave such keys out with a warning. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3 sink: pick the same value for metadata keys that differ only in case An entry holding both X-Amz-Meta-Project and the legacy x-amz-meta-project maps both to project, and map order decided which value was uploaded. Keep the lowest key, so the canonical form wins over the legacy lowercase one. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3 sink: use a test value codespell accepts Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chris LuandClaude Opus 5.5 authored and GitHub committed
1 parent
1c84f98cdf
commit
fdbaba52f9
2 files changed
+154
-9
No files matched your search
@@ -3,6 +3,7 @@ package S3Sink
|
||||
import (
|
||||
"encoding/base64"
|
||||
"fmt"
|
||||
"mime"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -22,8 +23,12 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/replication/sink"
|
||||
"github.com/seaweedfs/seaweedfs/weed/replication/source"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
"golang.org/x/net/http/httpguts"
|
||||
)
|
||||
|
||||
// maxMetadataSize is the S3 limit on user metadata, counted over keys and values.
|
||||
const maxMetadataSize = 2 * 1024
|
||||
|
||||
type S3Sink struct {
|
||||
conn s3iface.S3API
|
||||
filerSource *source.FilerSource
|
||||
@@ -192,15 +197,6 @@ func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures
|
||||
}
|
||||
}
|
||||
|
||||
doSaveMtime := true
|
||||
if entry.Extended == nil {
|
||||
entry.Extended = make(map[string][]byte)
|
||||
} else if _, ok := entry.Extended[s3_constants.AmzUserMetaMtime]; ok {
|
||||
doSaveMtime = false
|
||||
}
|
||||
if doSaveMtime {
|
||||
entry.Extended[s3_constants.AmzUserMetaMtime] = []byte(strconv.FormatInt(entry.Attributes.Mtime, 10))
|
||||
}
|
||||
// process tagging
|
||||
tags := buildTaggingString(entry.Extended)
|
||||
|
||||
@@ -213,6 +209,15 @@ func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures
|
||||
if tags != "" {
|
||||
uploadInput.Tagging = aws.String(tags)
|
||||
}
|
||||
metadata := toMetadata(entry.Extended)
|
||||
if _, found := metadata["mtime"]; !found {
|
||||
metadata["mtime"] = aws.String(strconv.FormatInt(entry.Attributes.Mtime, 10))
|
||||
}
|
||||
if size := metadataSize(metadata); size <= maxMetadataSize {
|
||||
uploadInput.Metadata = metadata
|
||||
} else {
|
||||
glog.Warningf("[%s] %s: %d bytes of user metadata exceed the S3 limit of %d, uploading without it", s3sink.bucket, key, size, maxMetadataSize)
|
||||
}
|
||||
if entry.Attributes.Mime != "" {
|
||||
uploadInput.ContentType = aws.String(entry.Attributes.Mime)
|
||||
}
|
||||
@@ -256,3 +261,37 @@ func buildTaggingString(extended map[string][]byte) string {
|
||||
}
|
||||
return tagValues.Encode()
|
||||
}
|
||||
|
||||
// toMetadata maps the X-Amz-Meta- attributes, matched case-insensitively for
|
||||
// entries written before the S3 gateway canonicalized them, to S3 user metadata.
|
||||
// Of keys that differ only in case the lowest wins, so X-Amz-Meta-Foo beats the
|
||||
// legacy x-amz-meta-foo. Values that are not printable ASCII are RFC 2047
|
||||
// encoded, which S3 decodes.
|
||||
func toMetadata(extended map[string][]byte) map[string]*string {
|
||||
prefix := s3_constants.AmzUserMetaPrefix
|
||||
metadata := make(map[string]*string)
|
||||
sourceKeys := make(map[string]string)
|
||||
for k, v := range extended {
|
||||
if len(k) <= len(prefix) || !strings.EqualFold(k[:len(prefix)], prefix) {
|
||||
continue
|
||||
}
|
||||
if !httpguts.ValidHeaderFieldName(k) {
|
||||
glog.Warningf("skip metadata %q: not a valid header name", k)
|
||||
continue
|
||||
}
|
||||
name := strings.ToLower(k[len(prefix):])
|
||||
if seen, found := sourceKeys[name]; found && seen < k {
|
||||
continue
|
||||
}
|
||||
sourceKeys[name] = k
|
||||
metadata[name] = aws.String(mime.QEncoding.Encode("utf-8", string(v)))
|
||||
}
|
||||
return metadata
|
||||
}
|
||||
|
||||
func metadataSize(metadata map[string]*string) (size int) {
|
||||
for k, v := range metadata {
|
||||
size += len(k) + len(*v)
|
||||
}
|
||||
return size
|
||||
}
|
||||
@@ -131,3 +131,109 @@ func TestBuildTaggingString_EmptyWhenNoTags(t *testing.T) {
|
||||
t.Errorf("expected empty tagging when no tag keys present, got %q", tagging)
|
||||
}
|
||||
}
|
||||
|
||||
func TestS3SinkCreateEntryForwardsUserMetadata(t *testing.T) {
|
||||
sink, rt := newCapturingS3Sink(t)
|
||||
entry := &filer_pb.Entry{
|
||||
Attributes: &filer_pb.FuseAttributes{Mtime: 1700000000},
|
||||
Content: []byte("data"),
|
||||
Extended: map[string][]byte{
|
||||
"X-Amz-Meta-Project": []byte("alpha"),
|
||||
"x-amz-meta-legacy-Key": []byte("old"),
|
||||
s3_constants.AmzObjectTaggingPrefix + "env": []byte("prod"),
|
||||
s3_constants.AmzStorageClass: []byte("STANDARD_IA"),
|
||||
"Content-Encoding": []byte("gzip"),
|
||||
},
|
||||
}
|
||||
require.NoError(t, sink.CreateEntry("/dir/test.bin", entry, nil))
|
||||
|
||||
require.NotNil(t, rt.uploadReq, "uploader should have issued a PUT")
|
||||
var metadata []string
|
||||
for k := range rt.uploadReq.Header {
|
||||
if strings.HasPrefix(k, s3_constants.AmzUserMetaPrefix) {
|
||||
metadata = append(metadata, k)
|
||||
}
|
||||
}
|
||||
require.ElementsMatch(t, []string{"X-Amz-Meta-Project", "X-Amz-Meta-Legacy-Key", "X-Amz-Meta-Mtime"}, metadata)
|
||||
require.Equal(t, "alpha", rt.uploadReq.Header.Get("X-Amz-Meta-Project"))
|
||||
require.Equal(t, "old", rt.uploadReq.Header.Get("X-Amz-Meta-Legacy-Key"))
|
||||
require.Equal(t, "1700000000", rt.uploadReq.Header.Get("X-Amz-Meta-Mtime"))
|
||||
require.Equal(t, "env=prod", rt.uploadReq.Header.Get("X-Amz-Tagging"))
|
||||
}
|
||||
|
||||
func TestS3SinkCreateEntryLeavesEntryExtendedUnchanged(t *testing.T) {
|
||||
sink, rt := newCapturingS3Sink(t)
|
||||
entry := &filer_pb.Entry{
|
||||
Attributes: &filer_pb.FuseAttributes{Mtime: 1700000000},
|
||||
Content: []byte("data"),
|
||||
}
|
||||
require.NoError(t, sink.CreateEntry("/dir/a.bin", entry, nil))
|
||||
require.Nil(t, entry.Extended)
|
||||
require.Equal(t, "1700000000", rt.uploadReq.Header.Get("X-Amz-Meta-Mtime"))
|
||||
|
||||
entry.Extended = map[string][]byte{"X-Amz-Meta-Mtime": []byte("1600000000")}
|
||||
require.NoError(t, sink.CreateEntry("/dir/b.bin", entry, nil))
|
||||
require.Equal(t, map[string][]byte{"X-Amz-Meta-Mtime": []byte("1600000000")}, entry.Extended)
|
||||
require.Equal(t, "1600000000", rt.uploadReq.Header.Get("X-Amz-Meta-Mtime"))
|
||||
}
|
||||
|
||||
func TestS3SinkCreateEntryEncodesNonASCIIMetadata(t *testing.T) {
|
||||
sink, rt := newCapturingS3Sink(t)
|
||||
entry := &filer_pb.Entry{
|
||||
Attributes: &filer_pb.FuseAttributes{},
|
||||
Content: []byte("data"),
|
||||
Extended: map[string][]byte{
|
||||
"X-Amz-Meta-City": []byte("Zürich"),
|
||||
"X-Amz-Meta-Note": []byte("line1\nline2"),
|
||||
"X-Amz-Meta-Plain": []byte("=?utf-8?q?Gen=C3=A8ve?="),
|
||||
},
|
||||
}
|
||||
require.NoError(t, sink.CreateEntry("/dir/test.bin", entry, nil))
|
||||
|
||||
require.Equal(t, "=?utf-8?q?Z=C3=BCrich?=", rt.uploadReq.Header.Get("X-Amz-Meta-City"))
|
||||
require.Equal(t, "=?utf-8?q?line1=0Aline2?=", rt.uploadReq.Header.Get("X-Amz-Meta-Note"))
|
||||
require.Equal(t, "=?utf-8?q?Gen=C3=A8ve?=", rt.uploadReq.Header.Get("X-Amz-Meta-Plain"))
|
||||
}
|
||||
|
||||
func TestS3SinkCreateEntrySkipsOversizedMetadata(t *testing.T) {
|
||||
sink, rt := newCapturingS3Sink(t)
|
||||
entry := &filer_pb.Entry{
|
||||
Attributes: &filer_pb.FuseAttributes{Mtime: 1700000000},
|
||||
Content: []byte("data"),
|
||||
Extended: map[string][]byte{
|
||||
"X-Amz-Meta-Big": []byte(strings.Repeat("x", maxMetadataSize)),
|
||||
s3_constants.AmzObjectTaggingPrefix + "env": []byte("prod"),
|
||||
},
|
||||
}
|
||||
require.NoError(t, sink.CreateEntry("/dir/big.bin", entry, nil))
|
||||
require.Equal(t, "", rt.uploadReq.Header.Get("X-Amz-Meta-Big"))
|
||||
require.Equal(t, "", rt.uploadReq.Header.Get("X-Amz-Meta-Mtime"))
|
||||
require.Equal(t, "env=prod", rt.uploadReq.Header.Get("X-Amz-Tagging"))
|
||||
|
||||
entry.Extended["X-Amz-Meta-Big"] = []byte(strings.Repeat("x", maxMetadataSize-len("big")-len("mtime")-len("1700000000")))
|
||||
require.NoError(t, sink.CreateEntry("/dir/fits.bin", entry, nil))
|
||||
require.Equal(t, string(entry.Extended["X-Amz-Meta-Big"]), rt.uploadReq.Header.Get("X-Amz-Meta-Big"))
|
||||
require.Equal(t, "1700000000", rt.uploadReq.Header.Get("X-Amz-Meta-Mtime"))
|
||||
}
|
||||
|
||||
func TestToMetadataSkipsInvalidHeaderNames(t *testing.T) {
|
||||
metadata := toMetadata(map[string][]byte{
|
||||
"X-Amz-Meta-Good": []byte("yes"),
|
||||
"X-Amz-Meta-bad key": []byte("space"),
|
||||
"X-Amz-Meta-bad\nkey": []byte("newline"),
|
||||
})
|
||||
require.Len(t, metadata, 1)
|
||||
require.Equal(t, "yes", *metadata["good"])
|
||||
}
|
||||
|
||||
func TestToMetadataPrefersCanonicalKeyOnCaseCollision(t *testing.T) {
|
||||
for i := 0; i < 20; i++ {
|
||||
metadata := toMetadata(map[string][]byte{
|
||||
"x-amz-meta-project": []byte("legacy"),
|
||||
"X-Amz-Meta-Project": []byte("canonical"),
|
||||
"X-Amz-Meta-project": []byte("mixed"),
|
||||
})
|
||||
require.Len(t, metadata, 1)
|
||||
require.Equal(t, "canonical", *metadata["project"])
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user