From 25ab4c3cac5292dae7cd3f8aee4c82d57106c9f6 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 15 Jul 2026 19:20:00 -0700 Subject: [PATCH] preserve Content-Encoding for remote-mounted objects (#10340) * 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. --- other/java/client/src/main/proto/filer.proto | 4 ++ weed/filer/filer_lazy_remote.go | 3 +- weed/filer/filer_lazy_remote_listing.go | 4 +- weed/filer/filer_lazy_remote_test.go | 14 +++-- weed/filer/read_remote.go | 20 ++++++ weed/filer/read_remote_test.go | 34 ++++++++++ weed/pb/filer.proto | 3 + weed/pb/filer_pb/filer.pb.go | 21 +++++-- weed/pb/filer_pb/filer_vtproto.pb.go | 44 +++++++++++++ .../azure/azure_storage_client.go | 38 +++++++++++- weed/remote_storage/gcs/gcs_storage_client.go | 58 ++++++++--------- weed/remote_storage/remote_storage.go | 9 +++ weed/remote_storage/s3/s3_storage_client.go | 62 +++++++++++++++++++ weed/shell/command_remote_cache.go | 6 +- weed/shell/command_remote_meta_sync.go | 39 ++++++++++-- weed/shell/command_remote_mount.go | 3 +- weed/shell/command_remote_mount_buckets.go | 2 +- 17 files changed, 315 insertions(+), 49 deletions(-) create mode 100644 weed/filer/read_remote_test.go diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index 070874e2e..db62bff65 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -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 { diff --git a/weed/filer/filer_lazy_remote.go b/weed/filer/filer_lazy_remote.go index 77599ca8c..801a3eac0 100644 --- a/weed/filer/filer_lazy_remote.go +++ b/weed/filer/filer_lazy_remote.go @@ -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) diff --git a/weed/filer/filer_lazy_remote_listing.go b/weed/filer/filer_lazy_remote_listing.go index 59b2d2ea3..4965589d7 100644 --- a/weed/filer/filer_lazy_remote_listing.go +++ b/weed/filer/filer_lazy_remote_listing.go @@ -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) diff --git a/weed/filer/filer_lazy_remote_test.go b/weed/filer/filer_lazy_remote_test.go index 6ad5e4449..33e0197f6 100644 --- a/weed/filer/filer_lazy_remote_test.go +++ b/weed/filer/filer_lazy_remote_test.go @@ -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") diff --git a/weed/filer/read_remote.go b/weed/filer/read_remote.go index b785014f3..0745aceaa 100644 --- a/weed/filer/read_remote.go +++ b/weed/filer/read_remote.go @@ -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, diff --git a/weed/filer/read_remote_test.go b/weed/filer/read_remote_test.go new file mode 100644 index 000000000..e7e34f76f --- /dev/null +++ b/weed/filer/read_remote_test.go @@ -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)) +} diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index ab9e0b285..db62bff65 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -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; diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 51d92a248..0013fb0ff 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -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), diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index ed0cd806d..39d59d9ed 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -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:]) diff --git a/weed/remote_storage/azure/azure_storage_client.go b/weed/remote_storage/azure/azure_storage_client.go index 2ed10cf3d..7f76fccb3 100644 --- a/weed/remote_storage/azure/azure_storage_client.go +++ b/weed/remote_storage/azure/azure_storage_client.go @@ -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 } diff --git a/weed/remote_storage/gcs/gcs_storage_client.go b/weed/remote_storage/gcs/gcs_storage_client.go index dc2c8f9ec..f776ae527 100644 --- a/weed/remote_storage/gcs/gcs_storage_client.go +++ b/weed/remote_storage/gcs/gcs_storage_client.go @@ -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) { diff --git a/weed/remote_storage/remote_storage.go b/weed/remote_storage/remote_storage.go index 117be244a..5e8454859 100644 --- a/weed/remote_storage/remote_storage.go +++ b/weed/remote_storage/remote_storage.go @@ -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 diff --git a/weed/remote_storage/s3/s3_storage_client.go b/weed/remote_storage/s3/s3_storage_client.go index e2a706c66..7505676cb 100644 --- a/weed/remote_storage/s3/s3_storage_client.go +++ b/weed/remote_storage/s3/s3_storage_client.go @@ -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{ diff --git a/weed/shell/command_remote_cache.go b/weed/shell/command_remote_cache.go index 575c9a771..ca78f2c2c 100644 --- a/weed/shell/command_remote_cache.go +++ b/weed/shell/command_remote_cache.go @@ -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 diff --git a/weed/shell/command_remote_meta_sync.go b/weed/shell/command_remote_meta_sync.go index b9e075ac2..2f270b435 100644 --- a/weed/shell/command_remote_meta_sync.go +++ b/weed/shell/command_remote_meta_sync.go @@ -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, }, }) diff --git a/weed/shell/command_remote_mount.go b/weed/shell/command_remote_mount.go index 108c1a011..60c1c93e0 100644 --- a/weed/shell/command_remote_mount.go +++ b/weed/shell/command_remote_mount.go @@ -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{ diff --git a/weed/shell/command_remote_mount_buckets.go b/weed/shell/command_remote_mount_buckets.go index 5f75c0feb..1fa464141 100644 --- a/weed/shell/command_remote_mount_buckets.go +++ b/weed/shell/command_remote_mount_buckets.go @@ -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) }