diff --git a/weed/replication/sink/s3sink/s3_sink.go b/weed/replication/sink/s3sink/s3_sink.go index 8f365a6e5a..0ffc2aaa2a 100644 --- a/weed/replication/sink/s3sink/s3_sink.go +++ b/weed/replication/sink/s3sink/s3_sink.go @@ -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 +} diff --git a/weed/replication/sink/s3sink/s3_sink_test.go b/weed/replication/sink/s3sink/s3_sink_test.go index 63bf8373ab..5d62e08ac8 100644 --- a/weed/replication/sink/s3sink/s3_sink_test.go +++ b/weed/replication/sink/s3sink/s3_sink_test.go @@ -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"]) + } +}