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) }