gian/seaweedfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

preserve Content-Encoding for remote-mounted objects (#10340)

Browse Source 
* remote storage: carry Content-Encoding into mounted entries

A RemoteEntry now records the remote object's Content-Encoding, and every
path that materializes a local entry from remote metadata (lazy fetch, lazy
listing, remote.mount, remote.meta.sync, remote.cache) stamps it into the
entry extended attributes, so HTTP and S3 HeadObject/GetObject return the
header. GCS and Azure populate it on listing and stat; S3 only exposes it
via HeadObject, so listings leave it empty.

* remote storage: set Content-Encoding when uploading to the remote

An entry carrying Content-Encoding in its extended attributes (a native S3
upload, or a value pulled from the remote) now keeps it when
filer.remote.sync or remote.copy.local writes the object to GCS, S3, or
Azure, instead of silently dropping it.

* gcs: read remote objects without decompressive transcoding

GCS transparently decompresses gzip-encoded objects on download, which
ignores range requests and returns byte counts that disagree with the
tracked RemoteSize. Request the stored bytes instead; chunked reads of
gzip-encoded objects then behave like any other object.

* remote storage: track Content-Encoding presence so removals propagate

A listing that does not report encodings (S3) leaves the field unset and
the local header untouched, while an authoritative report of no encoding
(GCS, Azure, any stat) now clears a previously stamped header instead of
leaving it stale. remote.cache also schedules a metadata update when only
the reported encoding changes.

* remote storage: propagate Content-Encoding on metadata-only updates

filer.remote.sync routes same-content changes through UpdateFileMetadata,
which only touched custom metadata (GCS, Azure) or tags (S3), so a
Content-Encoding change in the extended attributes never reached the
remote object's real header. GCS now patches contentEncoding alongside
the metadata, and Azure reissues the blob's HTTP headers with the new
value, carrying the others over since the call replaces the full set.
S3 stays tags-only: changing the header there means rewriting the
object, which the sync already does whenever content changes.

* remote.meta.sync: optional per-file stat for listing-omitted metadata

S3 listings carry no Content-Encoding, so entries synced from them never
learn it and the lazy-stat path never runs once an entry exists. With
-statFiles, each new or changed file whose listing left the encoding
unreported is stat-ed before reconciling, and the stat-derived value is
persisted so the next run only stats files that changed. Off by default:
it costs one remote request per file, and GCS and Azure listings already
carry the encoding.

* s3: apply metadata-only Content-Encoding changes with an in-place copy

Content-Encoding is S3 system metadata, so the tags-only metadata update
silently left the object's real header untouched. When the encoding
differs, reissue the object as a self-copy with replaced metadata,
carrying the content type and configured storage class like a fresh
write does. CopyObject caps at 5 GiB; beyond that the change is logged
and applies on the next content write.

* azure: skip the metadata call when user metadata is unchanged

An encoding-only change reissues the blob's HTTP headers; sending the
unchanged user metadata alongside it wastes a round trip and bumps the
blob's ETag once more than needed.

* s3: carry existing object metadata through the encoding copy

The replace directive drops everything not resent, and a mounted entry
usually has no local mime or user metadata, so the in-place copy wiped
the object's Content-Type, Cache-Control, user metadata, encryption
settings, and storage class. Read them back with a HeadObject first and
carry them over, overriding only what SeaweedFS manages: the encoding,
a locally set mime, and the configured storage class. S3 reports
Expires as a string while the copy input wants a time, so it is parsed
and skipped when malformed.
This commit is contained in:
Chris Lu 2026-07-15 19:20:00 -07:00 committed by GitHub
parent c015cc3939
commit 25ab4c3cac
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
17 changed files with 315 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
other/java/client/src/main/proto/filer.proto
View File

@ -143,6 +143,9 @@ message RemoteEntry {
string remote_e_tag = 3;
int64 remote_mtime = 4;
int64 remote_size = 5;
// unset when the remote listing does not report encodings (S3);
// empty when the remote object authoritatively has none
optional string remote_content_encoding = 6;
}
message Entry {
string name = 1;
@ -528,6 +531,7 @@ message AssignVolumeResponse {
string replication = 7;
string error = 8;
Location location = 9;
repeated Location replicas = 10;
}
message LookupVolumeRequest {

3
weed/filer/filer_lazy_remote.go
View File

@ -90,7 +90,8 @@ func (f *Filer) maybeLazyFetchFromRemote(ctx context.Context, p util.FullPath) (
Mode: 0644,
FileSize: uint64(remoteEntry.RemoteSize),
},
Remote: remoteEntry,
Extended: MergeRemoteContentEncoding(remoteEntry, nil),
Remote: remoteEntry,
}
persistBaseCtx, cancelPersist := context.WithTimeout(context.Background(), 30*time.Second)

4
weed/filer/filer_lazy_remote_listing.go
View File

@ -113,6 +113,7 @@ func (f *Filer) maybeLazyListFromRemote(ctx context.Context, p util.FullPath) {
existingEntry.Attr.Mtime = time.Unix(remoteEntry.RemoteMtime, 0)
}
existingEntry.Attr.FileSize = uint64(remoteEntry.RemoteSize)
existingEntry.Extended = MergeRemoteContentEncoding(remoteEntry, existingEntry.Extended)
}
if saveErr := f.Store.UpdateEntry(persistCtx, existingEntry); saveErr != nil {
glog.Warningf("maybeLazyListFromRemote: update %s: %v", childPath, saveErr)
@ -146,7 +147,8 @@ func (f *Filer) maybeLazyListFromRemote(ctx context.Context, p util.FullPath) {
Uid: OS_UID,
Gid: OS_GID,
},
Remote: remoteEntry,
Extended: MergeRemoteContentEncoding(remoteEntry, nil),
Remote: remoteEntry,
}
if remoteEntry != nil {
entry.Attr.FileSize = uint64(remoteEntry.RemoteSize)

14
weed/filer/filer_lazy_remote_test.go
View File

@ -24,6 +24,7 @@ import (
"github.com/stretchr/testify/require"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/protobuf/proto"
)
// --- minimal FilerStore stub ---
@ -304,7 +305,7 @@ func registerStubMaker(t *testing.T, storageType string, client remote_storage.R
func TestMaybeLazyFetchFromRemote_HitsRemoteAndPersists(t *testing.T) {
const storageType = "stub_lazy_hit"
stub := &stubRemoteClient{
statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 1234},
statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 1234, RemoteContentEncoding: proto.String("zstd")},
}
defer registerStubMaker(t, storageType, stub)()
@ -331,6 +332,7 @@ func TestMaybeLazyFetchFromRemote_HitsRemoteAndPersists(t *testing.T) {
stored, sErr := store.FindEntry(context.Background(), "/buckets/mybucket/file.txt")
require.NoError(t, sErr)
assert.Equal(t, int64(1234), stored.Remote.RemoteSize)
assert.Equal(t, []byte("zstd"), stored.Extended["Content-Encoding"])
}
func TestMaybeLazyFetchFromRemote_NotUnderMount(t *testing.T) {
@ -849,10 +851,11 @@ func TestMaybeLazyListFromRemote_PopulatesStoreFromRemote(t *testing.T) {
return err
}
if err := visitFn("/", "file.txt", false, &filer_pb.RemoteEntry{
RemoteMtime: 1700000000,
RemoteSize: 42,
RemoteETag: "abc",
StorageName: "myliststore",
RemoteMtime: 1700000000,
RemoteSize: 42,
RemoteETag: "abc",
StorageName: "myliststore",
RemoteContentEncoding: proto.String("gzip"),
}); err != nil {
return err
}
@ -882,6 +885,7 @@ func TestMaybeLazyListFromRemote_PopulatesStoreFromRemote(t *testing.T) {
require.NotNil(t, fileEntry, "file.txt should be persisted")
assert.Equal(t, uint64(42), fileEntry.FileSize)
assert.NotNil(t, fileEntry.Remote)
assert.Equal(t, []byte("gzip"), fileEntry.Extended["Content-Encoding"])
// Check that the subdirectory was persisted
dirEntry := store.getEntry("/buckets/mybucket/subdir")

20
weed/filer/read_remote.go
View File

@ -12,6 +12,26 @@ func (entry *Entry) IsInRemoteOnly() bool {
return len(entry.GetChunks()) == 0 && entry.Remote != nil && entry.Remote.RemoteSize > 0
}
// MergeRemoteContentEncoding stamps the remote object's Content-Encoding into the
// entry extended attributes so HTTP and S3 reads return the header. An unset
// remote value (a listing that does not report encodings) leaves the attribute
// alone; an authoritatively empty one removes it.
func MergeRemoteContentEncoding(remoteEntry *filer_pb.RemoteEntry, extended map[string][]byte) map[string][]byte {
if remoteEntry == nil || remoteEntry.RemoteContentEncoding == nil {
return extended
}
encoding := *remoteEntry.RemoteContentEncoding
if encoding == "" {
delete(extended, "Content-Encoding")
return extended
}
if extended == nil {
extended = make(map[string][]byte)
}
extended["Content-Encoding"] = []byte(encoding)
return extended
}
func MapFullPathToRemoteStorageLocation(localMountedDir util.FullPath, remoteMountedLocation *remote_pb.RemoteStorageLocation, fp util.FullPath) *remote_pb.RemoteStorageLocation {
remoteLocation := &remote_pb.RemoteStorageLocation{
Name: remoteMountedLocation.Name,

34
weed/filer/read_remote_test.go Normal file
View File

@ -0,0 +1,34 @@
package filer
import (
"testing"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/stretchr/testify/assert"
"google.golang.org/protobuf/proto"
)
func TestMergeRemoteContentEncoding(t *testing.T) {
merged := MergeRemoteContentEncoding(&filer_pb.RemoteEntry{RemoteContentEncoding: proto.String("zstd")}, nil)
assert.Equal(t, []byte("zstd"), merged["Content-Encoding"])
// existing keys are preserved, the encoding is overwritten from remote
merged = MergeRemoteContentEncoding(&filer_pb.RemoteEntry{RemoteContentEncoding: proto.String("gzip")}, map[string][]byte{
"Content-Encoding": []byte("zstd"),
"x-amz-meta-foo": []byte("bar"),
})
assert.Equal(t, []byte("gzip"), merged["Content-Encoding"])
assert.Equal(t, []byte("bar"), merged["x-amz-meta-foo"])
// an unreported encoding (S3 listings) leaves a locally set value alone
local := map[string][]byte{"Content-Encoding": []byte("zstd")}
merged = MergeRemoteContentEncoding(&filer_pb.RemoteEntry{}, local)
assert.Equal(t, []byte("zstd"), merged["Content-Encoding"])
// an authoritatively empty one removes it
merged = MergeRemoteContentEncoding(&filer_pb.RemoteEntry{RemoteContentEncoding: proto.String("")}, local)
assert.NotContains(t, merged, "Content-Encoding")
assert.Nil(t, MergeRemoteContentEncoding(nil, nil))
assert.Nil(t, MergeRemoteContentEncoding(&filer_pb.RemoteEntry{RemoteContentEncoding: proto.String("")}, nil))
}

3
weed/pb/filer.proto
View File

@ -143,6 +143,9 @@ message RemoteEntry {
string remote_e_tag = 3;
int64 remote_mtime = 4;
int64 remote_size = 5;
// unset when the remote listing does not report encodings (S3);
// empty when the remote object authoritatively has none
optional string remote_content_encoding = 6;
}
message Entry {
string name = 1;

21
weed/pb/filer_pb/filer.pb.go
View File

@ -552,8 +552,11 @@ type RemoteEntry struct {
RemoteETag string `protobuf:"bytes,3,opt,name=remote_e_tag,json=remoteETag,proto3" json:"remote_e_tag,omitempty"`
RemoteMtime int64 `protobuf:"varint,4,opt,name=remote_mtime,json=remoteMtime,proto3" json:"remote_mtime,omitempty"`
RemoteSize int64 `protobuf:"varint,5,opt,name=remote_size,json=remoteSize,proto3" json:"remote_size,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
// unset when the remote listing does not report encodings (S3);
// empty when the remote object authoritatively has none
RemoteContentEncoding *string `protobuf:"bytes,6,opt,name=remote_content_encoding,json=remoteContentEncoding,proto3,oneof" json:"remote_content_encoding,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *RemoteEntry) Reset() {
@ -621,6 +624,13 @@ func (x *RemoteEntry) GetRemoteSize() int64 {
return 0
}
func (x *RemoteEntry) GetRemoteContentEncoding() string {
if x != nil && x.RemoteContentEncoding != nil {
return *x.RemoteContentEncoding
}
return ""
}
type Entry struct {
state protoimpl.MessageState `protogen:"open.v1"`
Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
@ -6818,7 +6828,7 @@ const file_filer_proto_rawDesc = "" +
"\x0esnapshot_ts_ns\x18\x06 \x01(\x03R\fsnapshotTsNs\"b\n" +
"\x13ListEntriesResponse\x12%\n" +
"\x05entry\x18\x01 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x12$\n" +
"\x0esnapshot_ts_ns\x18\x02 \x01(\x03R\fsnapshotTsNs\"\xc8\x01\n" +
"\x0esnapshot_ts_ns\x18\x02 \x01(\x03R\fsnapshotTsNs\"\xa1\x02\n" +
"\vRemoteEntry\x12!\n" +
"\fstorage_name\x18\x01 \x01(\tR\vstorageName\x120\n" +
"\x15last_local_sync_ts_ns\x18\x02 \x01(\x03R\x11lastLocalSyncTsNs\x12 \n" +
@ -6826,7 +6836,9 @@ const file_filer_proto_rawDesc = "" +
"remoteETag\x12!\n" +
"\fremote_mtime\x18\x04 \x01(\x03R\vremoteMtime\x12\x1f\n" +
"\vremote_size\x18\x05 \x01(\x03R\n" +
"remoteSize\"\x89\x04\n" +
"remoteSize\x12;\n" +
"\x17remote_content_encoding\x18\x06 \x01(\tH\x00R\x15remoteContentEncoding\x88\x01\x01B\x1a\n" +
"\x18_remote_content_encoding\"\x89\x04\n" +
"\x05Entry\x12\x12\n" +
"\x04name\x18\x01 \x01(\tR\x04name\x12!\n" +
"\fis_directory\x18\x02 \x01(\bR\visDirectory\x12+\n" +
@ -7719,6 +7731,7 @@ func file_filer_proto_init() {
if File_filer_proto != nil {
return
}
file_filer_proto_msgTypes[4].OneofWrappers = []any{}
file_filer_proto_msgTypes[84].OneofWrappers = []any{
(*StreamMutateEntryRequest_CreateRequest)(nil),
(*StreamMutateEntryRequest_UpdateRequest)(nil),

44
weed/pb/filer_pb/filer_vtproto.pb.go
View File

@ -261,6 +261,13 @@ func (m *RemoteEntry) MarshalToSizedBufferVT(dAtA []byte) (int, error) {
i -= len(m.unknownFields)
copy(dAtA[i:], m.unknownFields)
}
if m.RemoteContentEncoding != nil {
i -= len(*m.RemoteContentEncoding)
copy(dAtA[i:], *m.RemoteContentEncoding)
i = protohelpers.EncodeVarint(dAtA, i, uint64(len(*m.RemoteContentEncoding)))
i--
dAtA[i] = 0x32
}
if m.RemoteSize != 0 {
i = protohelpers.EncodeVarint(dAtA, i, uint64(m.RemoteSize))
i--
@ -6247,6 +6254,10 @@ func (m *RemoteEntry) SizeVT() (n int) {
if m.RemoteSize != 0 {
n += 1 + protohelpers.SizeOfVarint(uint64(m.RemoteSize))
}
if m.RemoteContentEncoding != nil {
l = len(*m.RemoteContentEncoding)
n += 1 + l + protohelpers.SizeOfVarint(uint64(l))
}
n += len(m.unknownFields)
return n
}
@ -9260,6 +9271,39 @@ func (m *RemoteEntry) UnmarshalVT(dAtA []byte) error {
break
}
}
case 6:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field RemoteContentEncoding", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return protohelpers.ErrIntOverflow
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return protohelpers.ErrInvalidLength
}
postIndex := iNdEx + intStringLen
if postIndex < 0 {
return protohelpers.ErrInvalidLength
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
s := string(dAtA[iNdEx:postIndex])
m.RemoteContentEncoding = &s
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := protohelpers.Skip(dAtA[iNdEx:])

38
weed/remote_storage/azure/azure_storage_client.go
View File

@ -190,6 +190,7 @@ func (az *azureRemoteStorageClient) ListDirectory(ctx context.Context, loc *remo
if blobItem.Properties.ETag != nil {
remoteEntry.RemoteETag = string(*blobItem.Properties.ETag)
}
remoteEntry.RemoteContentEncoding = remoteContentEncoding(blobItem.Properties.ContentEncoding)
}
if err = visitFn(dir, name, false, remoteEntry); err != nil {
@ -224,9 +225,20 @@ func (az *azureRemoteStorageClient) StatFile(loc *remote_pb.RemoteStorageLocatio
if resp.ETag != nil {
remoteEntry.RemoteETag = string(*resp.ETag)
}
remoteEntry.RemoteContentEncoding = remoteContentEncoding(resp.ContentEncoding)
return remoteEntry, nil
}
// blob properties report no Content-Encoding as nil; the remote entry records
// that authoritatively as empty
func remoteContentEncoding(encoding *string) *string {
if encoding == nil {
empty := ""
return &empty
}
return encoding
}
func (az *azureRemoteStorageClient) Traverse(loc *remote_pb.RemoteStorageLocation, visitFn remote_storage.VisitFunc) (err error) {
pathKey := loc.Path[1:]
@ -263,6 +275,7 @@ func (az *azureRemoteStorageClient) Traverse(loc *remote_pb.RemoteStorageLocatio
if blobItem.Properties.ETag != nil {
remoteEntry.RemoteETag = string(*blobItem.Properties.ETag)
}
remoteEntry.RemoteContentEncoding = remoteContentEncoding(blobItem.Properties.ContentEncoding)
}
err = visitFn(dir, name, false, remoteEntry)
@ -379,6 +392,9 @@ func (az *azureRemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocati
if entry.Attributes != nil && entry.Attributes.Mime != "" {
httpHeaders.BlobContentType = &entry.Attributes.Mime
}
if contentEncoding := remote_storage.EntryContentEncoding(entry); contentEncoding != "" {
httpHeaders.BlobContentEncoding = &contentEncoding
}
_, err = blobClient.UploadStream(context.Background(), reader, &blockblob.UploadStreamOptions{
BlockSize: defaultBlockSize,
@ -424,7 +440,27 @@ func (az *azureRemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStor
key := loc.Path[1:]
blobClient := az.client.ServiceClient().NewContainerClient(loc.Bucket).NewBlobClient(key)
_, err = blobClient.SetMetadata(context.Background(), metadata, nil)
if !reflect.DeepEqual(toMetadata(oldEntry.Extended), metadata) {
if _, err = blobClient.SetMetadata(context.Background(), metadata, nil); err != nil {
return err
}
}
if encoding := remote_storage.EntryContentEncoding(newEntry); encoding != remote_storage.EntryContentEncoding(oldEntry) {
// SetHTTPHeaders replaces the whole header set, so carry the rest over
props, getErr := blobClient.GetProperties(context.Background(), nil)
if getErr != nil {
return fmt.Errorf("azure get properties %s%s: %w", loc.Bucket, loc.Path, getErr)
}
httpHeaders := blob.ParseHTTPHeaders(props)
httpHeaders.BlobContentEncoding = nil
if encoding != "" {
httpHeaders.BlobContentEncoding = &encoding
}
if _, err = blobClient.SetHTTPHeaders(context.Background(), httpHeaders, nil); err != nil {
return fmt.Errorf("azure set http headers %s%s: %w", loc.Bucket, loc.Path, err)
}
}
return
}

58
weed/remote_storage/gcs/gcs_storage_client.go
View File

@ -20,6 +20,7 @@ import (
"golang.org/x/oauth2/google"
"google.golang.org/api/iterator"
"google.golang.org/api/option"
"google.golang.org/protobuf/proto"
)
func init() {
@ -96,6 +97,16 @@ type gcsRemoteStorageClient struct {
var _ = remote_storage.RemoteStorageClient(&gcsRemoteStorageClient{})
func (gcs *gcsRemoteStorageClient) toRemoteEntry(attr *storage.ObjectAttrs) *filer_pb.RemoteEntry {
return &filer_pb.RemoteEntry{
StorageName: gcs.conf.Name,
RemoteMtime: attr.Updated.Unix(),
RemoteSize: attr.Size,
RemoteETag: attr.Etag,
RemoteContentEncoding: proto.String(attr.ContentEncoding),
}
}
func (gcs *gcsRemoteStorageClient) Traverse(loc *remote_pb.RemoteStorageLocation, visitFn remote_storage.VisitFunc) (err error) {
pathKey := loc.Path[1:]
@ -119,12 +130,7 @@ func (gcs *gcsRemoteStorageClient) Traverse(loc *remote_pb.RemoteStorageLocation
key := objectAttr.Name
key = "/" + key
dir, name := util.FullPath(key).DirAndName()
err = visitFn(dir, name, false, &filer_pb.RemoteEntry{
RemoteMtime: objectAttr.Updated.Unix(),
RemoteSize: objectAttr.Size,
RemoteETag: objectAttr.Etag,
StorageName: gcs.conf.Name,
})
err = visitFn(dir, name, false, gcs.toRemoteEntry(objectAttr))
}
return
}
@ -165,12 +171,7 @@ func (gcs *gcsRemoteStorageClient) ListDirectory(ctx context.Context, loc *remot
continue // skip directory markers
}
dir, name := util.FullPath(key).DirAndName()
if err = visitFn(dir, name, false, &filer_pb.RemoteEntry{
RemoteMtime: objectAttr.Updated.Unix(),
RemoteSize: objectAttr.Size,
RemoteETag: objectAttr.Etag,
StorageName: gcs.conf.Name,
}); err != nil {
if err = visitFn(dir, name, false, gcs.toRemoteEntry(objectAttr)); err != nil {
return err
}
}
@ -188,18 +189,15 @@ func (gcs *gcsRemoteStorageClient) StatFile(loc *remote_pb.RemoteStorageLocation
}
return nil, fmt.Errorf("stat gcs %s%s: %w", loc.Bucket, loc.Path, err)
}
return &filer_pb.RemoteEntry{
StorageName: gcs.conf.Name,
RemoteMtime: attr.Updated.Unix(),
RemoteSize: attr.Size,
RemoteETag: attr.Etag,
}, nil
return gcs.toRemoteEntry(attr), nil
}
func (gcs *gcsRemoteStorageClient) ReadFile(loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (data []byte, err error) {
key := loc.Path[1:]
rangeReader, readErr := gcs.client.Bucket(loc.Bucket).Object(key).NewRangeReader(context.Background(), offset, size)
// read the stored bytes: decompressive transcoding of gzip-encoded objects
// breaks range reads and returns sizes that disagree with RemoteSize
rangeReader, readErr := gcs.client.Bucket(loc.Bucket).Object(key).ReadCompressed(true).NewRangeReader(context.Background(), offset, size)
if readErr != nil {
return nil, readErr
}
@ -214,7 +212,7 @@ func (gcs *gcsRemoteStorageClient) ReadFile(loc *remote_pb.RemoteStorageLocation
func (gcs *gcsRemoteStorageClient) ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error) {
key := loc.Path[1:]
return gcs.client.Bucket(loc.Bucket).Object(key).NewRangeReader(ctx, offset, size)
return gcs.client.Bucket(loc.Bucket).Object(key).ReadCompressed(true).NewRangeReader(ctx, offset, size)
}
func (gcs *gcsRemoteStorageClient) WriteDirectory(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) (err error) {
@ -235,6 +233,7 @@ func (gcs *gcsRemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocatio
if entry.Attributes != nil && entry.Attributes.Mime != "" {
wc.ContentType = entry.Attributes.Mime
}
wc.ContentEncoding = remote_storage.EntryContentEncoding(entry)
if _, err = io.Copy(wc, reader); err != nil {
return nil, fmt.Errorf("upload to gcs %s/%s%s: %v", loc.Name, loc.Bucket, loc.Path, err)
}
@ -266,18 +265,21 @@ func (gcs *gcsRemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStora
if reflect.DeepEqual(oldEntry.Extended, newEntry.Extended) {
return nil
}
metadata := toMetadata(newEntry.Extended)
key := loc.Path[1:]
if len(metadata) > 0 {
_, err = gcs.client.Bucket(loc.Bucket).Object(key).Update(context.Background(), storage.ObjectAttrsToUpdate{
Metadata: metadata,
})
attrsToUpdate := storage.ObjectAttrsToUpdate{}
if metadata := toMetadata(newEntry.Extended); len(metadata) > 0 {
attrsToUpdate.Metadata = metadata
} else {
// no way to delete the metadata yet
}
if encoding := remote_storage.EntryContentEncoding(newEntry); encoding != remote_storage.EntryContentEncoding(oldEntry) {
attrsToUpdate.ContentEncoding = encoding // empty clears the header
}
if attrsToUpdate.Metadata == nil && attrsToUpdate.ContentEncoding == nil {
return nil
}
key := loc.Path[1:]
_, err = gcs.client.Bucket(loc.Bucket).Object(key).Update(context.Background(), attrsToUpdate)
return
}
func (gcs *gcsRemoteStorageClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) (err error) {

9
weed/remote_storage/remote_storage.go
View File

@ -66,6 +66,15 @@ func FormatLocation(loc *remote_pb.RemoteStorageLocation) string {
type VisitFunc func(dir string, name string, isDirectory bool, remoteEntry *filer_pb.RemoteEntry) error
// EntryContentEncoding returns the Content-Encoding stored in the entry
// extended attributes, for clients to set on uploaded remote objects.
func EntryContentEncoding(entry *filer_pb.Entry) string {
if entry == nil {
return ""
}
return string(entry.Extended["Content-Encoding"])
}
type Bucket struct {
Name string
CreatedAt time.Time

62
weed/remote_storage/s3/s3_storage_client.go
View File

@ -5,6 +5,7 @@ import (
"fmt"
"io"
"net/http"
"net/url"
"reflect"
"strings"
@ -18,6 +19,7 @@ import (
"github.com/aws/aws-sdk-go/service/s3/s3iface"
"github.com/aws/aws-sdk-go/service/s3/s3manager"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
"github.com/seaweedfs/seaweedfs/weed/remote_storage"
@ -224,6 +226,8 @@ func (s *s3RemoteStorageClient) StatFile(loc *remote_pb.RemoteStorageLocation) (
if resp.ETag != nil {
remoteEntry.RemoteETag = *resp.ETag
}
// a HeadObject response is authoritative: no header means no encoding
remoteEntry.RemoteContentEncoding = aws.String(aws.StringValue(resp.ContentEncoding))
return remoteEntry, nil
}
@ -323,6 +327,9 @@ func (s *s3RemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocation,
if entry.Attributes != nil && entry.Attributes.Mime != "" {
uploadInput.ContentType = aws.String(entry.Attributes.Mime)
}
if contentEncoding := remote_storage.EntryContentEncoding(entry); contentEncoding != "" {
uploadInput.ContentEncoding = aws.String(contentEncoding)
}
if s.conf.S3StorageClass != "" {
uploadInput.StorageClass = aws.String(s.conf.S3StorageClass)
}
@ -353,10 +360,65 @@ func (s *s3RemoteStorageClient) readFileRemoteEntry(loc *remote_pb.RemoteStorage
return s.StatFile(loc)
}
// the largest object a single CopyObject call accepts
const s3CopyObjectSizeLimit = 5 * 1024 * 1024 * 1024
func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageLocation, oldEntry *filer_pb.Entry, newEntry *filer_pb.Entry) (err error) {
if reflect.DeepEqual(oldEntry.Extended, newEntry.Extended) {
return nil
}
// Content-Encoding is S3 system metadata, changeable without a content
// rewrite only through an in-place copy
if encoding := remote_storage.EntryContentEncoding(newEntry); encoding != remote_storage.EntryContentEncoding(oldEntry) {
key := loc.Path[1:]
if fileSize := int64(filer.FileSize(newEntry)); fileSize > s3CopyObjectSizeLimit {
glog.Warningf("s3 %s/%s: applying the Content-Encoding change needs an object copy, but %d bytes exceeds the copy limit; it will apply on the next content write", loc.Bucket, key, fileSize)
} else {
// the replace directive drops everything not resent, so read the
// object's current metadata and carry it over
headOut, headErr := s.conn.HeadObject(&s3.HeadObjectInput{
Bucket: aws.String(loc.Bucket),
Key: aws.String(key),
})
if headErr != nil {
return fmt.Errorf("stat %s/%s before metadata copy: %w", loc.Bucket, key, headErr)
}
copyInput := &s3.CopyObjectInput{
Bucket: aws.String(loc.Bucket),
Key: aws.String(key),
CopySource: aws.String(url.PathEscape(loc.Bucket + "/" + key)),
MetadataDirective: aws.String(s3.MetadataDirectiveReplace),
Metadata: headOut.Metadata,
ContentType: headOut.ContentType,
CacheControl: headOut.CacheControl,
ContentDisposition: headOut.ContentDisposition,
ContentLanguage: headOut.ContentLanguage,
WebsiteRedirectLocation: headOut.WebsiteRedirectLocation,
ServerSideEncryption: headOut.ServerSideEncryption,
SSEKMSKeyId: headOut.SSEKMSKeyId,
StorageClass: headOut.StorageClass,
}
if headOut.Expires != nil {
if expires, parseErr := http.ParseTime(*headOut.Expires); parseErr == nil {
copyInput.Expires = aws.Time(expires)
}
}
if encoding != "" {
copyInput.ContentEncoding = aws.String(encoding)
}
if newEntry.Attributes != nil && newEntry.Attributes.Mime != "" {
copyInput.ContentType = aws.String(newEntry.Attributes.Mime)
}
if s.conf.S3StorageClass != "" {
copyInput.StorageClass = aws.String(s.conf.S3StorageClass)
}
if _, err = s.conn.CopyObject(copyInput); err != nil {
return fmt.Errorf("update content encoding of %s/%s: %w", loc.Bucket, key, err)
}
}
}
tagging := toTagging(newEntry.Extended)
if len(tagging.TagSet) > 0 {
_, err = s.conn.PutObjectTagging(&s3.PutObjectTaggingInput{

6
weed/shell/command_remote_cache.go
View File

@ -160,7 +160,9 @@ func (c *commandRemoteCache) doComprehensiveSync(commandEnv *CommandEnv, writer
// File exists locally, check if it needs updating
if localEntry.RemoteEntry == nil ||
localEntry.RemoteEntry.RemoteETag != remoteEntry.RemoteETag ||
localEntry.RemoteEntry.RemoteMtime < remoteEntry.RemoteMtime {
localEntry.RemoteEntry.RemoteMtime < remoteEntry.RemoteMtime ||
(remoteEntry.RemoteContentEncoding != nil &&
localEntry.RemoteEntry.GetRemoteContentEncoding() != remoteEntry.GetRemoteContentEncoding()) {
filesToUpdate = append(filesToUpdate, remotePath)
}
// Check if it needs caching
@ -275,6 +277,7 @@ func (c *commandRemoteCache) doComprehensiveSync(commandEnv *CommandEnv, writer
Mtime: remoteEntry.RemoteMtime,
FileMode: remoteEntryFileMode(isDirectory),
},
Extended: filer.MergeRemoteContentEncoding(remoteEntry, nil),
RemoteEntry: remoteEntry,
},
})
@ -295,6 +298,7 @@ func (c *commandRemoteCache) doComprehensiveSync(commandEnv *CommandEnv, writer
existingEntry.Attributes.FileSize = uint64(remoteEntry.RemoteSize)
existingEntry.Attributes.Mtime = remoteEntry.RemoteMtime
existingEntry.Attributes.Md5 = nil
existingEntry.Extended = filer.MergeRemoteContentEncoding(remoteEntry, existingEntry.Extended)
existingEntry.Chunks = nil
existingEntry.Content = nil

39
weed/shell/command_remote_meta_sync.go
View File

@ -43,6 +43,9 @@ func (c *commandRemoteMetaSync) Help() string {
Local metadata for files and directories removed from the remote is also
removed by default; pass -delete=false to keep it.
S3 listings do not report Content-Encoding; pass -statFiles to stat each
new or changed file so that metadata is synchronized too.
This is designed to run regularly. So you can add it to some cronjob.
If there are no other operations changing remote files, this operation is not needed.
@ -60,6 +63,7 @@ func (c *commandRemoteMetaSync) Do(args []string, commandEnv *CommandEnv, writer
dir := remoteMetaSyncCommand.String("dir", "", "a directory in filer")
deleteStale := remoteMetaSyncCommand.Bool("delete", true, "remove local metadata of files and directories deleted from remote")
statFiles := remoteMetaSyncCommand.Bool("statFiles", false, "stat each new or changed file when the listing does not report all metadata (S3 listings lack Content-Encoding); costs one extra remote request per file")
if err = remoteMetaSyncCommand.Parse(args); err != nil {
return nil
@ -72,7 +76,7 @@ func (c *commandRemoteMetaSync) Do(args []string, commandEnv *CommandEnv, writer
}
// pull metadata from remote
if err = pullMetadata(commandEnv, writer, util.FullPath(localMountedDir), remoteStorageMountedLocation, util.FullPath(*dir), remoteStorageConf, *deleteStale); err != nil {
if err = pullMetadata(commandEnv, writer, util.FullPath(localMountedDir), remoteStorageMountedLocation, util.FullPath(*dir), remoteStorageConf, *deleteStale, *statFiles); err != nil {
return fmt.Errorf("cache meta data: %w", err)
}
@ -148,7 +152,7 @@ type remoteChild struct {
remoteEntry *filer_pb.RemoteEntry
}
func pullMetadata(commandEnv *CommandEnv, writer io.Writer, localMountedDir util.FullPath, remoteMountedLocation *remote_pb.RemoteStorageLocation, dirToCache util.FullPath, remoteConf *remote_pb.RemoteConf, deleteStale bool) error {
func pullMetadata(commandEnv *CommandEnv, writer io.Writer, localMountedDir util.FullPath, remoteMountedLocation *remote_pb.RemoteStorageLocation, dirToCache util.FullPath, remoteConf *remote_pb.RemoteConf, deleteStale bool, statFiles bool) error {
remoteStorage, err := remote_storage.GetRemoteStorage(remoteConf)
if err != nil {
@ -158,7 +162,7 @@ func pullMetadata(commandEnv *CommandEnv, writer io.Writer, localMountedDir util
remote := filer.MapFullPathToRemoteStorageLocation(localMountedDir, remoteMountedLocation, dirToCache)
return commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
return pullMetadataDirectory(context.Background(), client, writer, remoteStorage, dirToCache, remote, deleteStale)
return pullMetadataDirectory(context.Background(), client, writer, remoteStorage, dirToCache, remote, deleteStale, statFiles)
})
}
@ -167,7 +171,7 @@ func pullMetadata(commandEnv *CommandEnv, writer io.Writer, localMountedDir util
// delimiter surfaces subdirectories, including empty ones, as their own
// entries, so directories are materialized locally even when they hold no
// files.
func pullMetadataDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, writer io.Writer, remoteStorage remote_storage.RemoteStorageClient, localDir util.FullPath, remoteLoc *remote_pb.RemoteStorageLocation, deleteStale bool) error {
func pullMetadataDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, writer io.Writer, remoteStorage remote_storage.RemoteStorageClient, localDir util.FullPath, remoteLoc *remote_pb.RemoteStorageLocation, deleteStale bool, statFiles bool) error {
remoteChildren := make(map[string]*remoteChild)
if err := remoteStorage.ListDirectory(ctx, remoteLoc, func(dir, name string, isDirectory bool, remoteEntry *filer_pb.RemoteEntry) error {
@ -218,6 +222,23 @@ func pullMetadataDirectory(ctx context.Context, client filer_pb.SeaweedFilerClie
existingEntry = nil
}
// enrich a listing that does not report encodings with a per-file stat,
// skipping files whose stored metadata is already present and unchanged
if statFiles && !child.isDirectory && child.remoteEntry.RemoteContentEncoding == nil {
needsStat := existingEntry == nil
if existingEntry != nil && existingEntry.RemoteEntry != nil {
needsStat = existingEntry.RemoteEntry.RemoteContentEncoding == nil ||
existingEntry.RemoteEntry.RemoteETag != child.remoteEntry.RemoteETag ||
existingEntry.RemoteEntry.RemoteMtime != child.remoteEntry.RemoteMtime ||
existingEntry.RemoteEntry.RemoteSize != child.remoteEntry.RemoteSize
}
if needsStat {
if statEntry, statErr := remoteStorage.StatFile(childRemoteLocation(remoteLoc, name)); statErr == nil && statEntry != nil {
child.remoteEntry = statEntry
}
}
}
if existingEntry == nil {
if err := createRemoteEntry(ctx, client, writer, localDir, name, child, remoteLoc.Name); err != nil {
return err
@ -228,7 +249,12 @@ func pullMetadataDirectory(ctx context.Context, client filer_pb.SeaweedFilerClie
fmt.Fprintf(writer, "%s (skip)\n", localPath)
} else if existingEntry.RemoteEntry.RemoteETag != child.remoteEntry.RemoteETag ||
existingEntry.RemoteEntry.RemoteMtime != child.remoteEntry.RemoteMtime ||
existingEntry.RemoteEntry.RemoteSize != child.remoteEntry.RemoteSize {
existingEntry.RemoteEntry.RemoteSize != child.remoteEntry.RemoteSize ||
(child.remoteEntry.RemoteContentEncoding != nil &&
existingEntry.RemoteEntry.GetRemoteContentEncoding() != child.remoteEntry.GetRemoteContentEncoding()) ||
// persist the stat-derived encoding so the next run skips the stat
(statFiles && existingEntry.RemoteEntry.RemoteContentEncoding == nil &&
child.remoteEntry.RemoteContentEncoding != nil) {
fmt.Fprintf(writer, "%s (update)\n", localPath)
if err := doSaveRemoteEntry(client, string(localDir), existingEntry, child.remoteEntry); err != nil {
return err
@ -239,7 +265,7 @@ func pullMetadataDirectory(ctx context.Context, client filer_pb.SeaweedFilerClie
}
if child.isDirectory {
if err := pullMetadataDirectory(ctx, client, writer, remoteStorage, localPath, childRemoteLocation(remoteLoc, name), deleteStale); err != nil {
if err := pullMetadataDirectory(ctx, client, writer, remoteStorage, localPath, childRemoteLocation(remoteLoc, name), deleteStale, statFiles); err != nil {
return err
}
}
@ -341,6 +367,7 @@ func createRemoteEntry(ctx context.Context, client filer_pb.SeaweedFilerClient,
Name: name,
IsDirectory: child.isDirectory,
Attributes: attributes,
Extended: filer.MergeRemoteContentEncoding(remoteEntry, nil),
RemoteEntry: remoteEntry,
},
})

3
weed/shell/command_remote_mount.go
View File

@ -102,7 +102,7 @@ func (c *commandRemoteMount) Do(args []string, commandEnv *CommandEnv, writer io
}
if strategy == MetadataCacheEager {
if err = pullMetadata(commandEnv, writer, util.FullPath(*dir), remoteStorageLocation, util.FullPath(*dir), remoteConf, false); err != nil {
if err = pullMetadata(commandEnv, writer, util.FullPath(*dir), remoteStorageLocation, util.FullPath(*dir), remoteConf, false, false); err != nil {
return fmt.Errorf("cache metadata: %w", err)
}
}
@ -203,6 +203,7 @@ func doSaveRemoteEntry(client filer_pb.SeaweedFilerClient, localDir string, exis
existingEntry.Attributes.Mtime = remoteEntry.RemoteMtime
existingEntry.Attributes.Md5 = nil
existingEntry.Attributes.TtlSec = 0 // Remote entries should not have TTL
existingEntry.Extended = filer.MergeRemoteContentEncoding(remoteEntry, existingEntry.Extended)
existingEntry.Chunks = nil
existingEntry.Content = nil
_, updateErr := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{

2
weed/shell/command_remote_mount_buckets.go
View File

@ -112,7 +112,7 @@ func (c *commandRemoteMountBuckets) Do(args []string, commandEnv *CommandEnv, wr
if err = ensureMountDirectory(commandEnv, string(dir), true, remoteConf); err != nil {
return fmt.Errorf("mount setup on %+v: %v", remoteStorageLocation, err)
}
if err = pullMetadata(commandEnv, writer, dir, remoteStorageLocation, dir, remoteConf, true); err != nil {
if err = pullMetadata(commandEnv, writer, dir, remoteStorageLocation, dir, remoteConf, true, false); err != nil {
return fmt.Errorf("cache metadata on %+v: %v", remoteStorageLocation, err)
}