diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 1bf864f305..b8051e2cf9 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -43,14 +43,6 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour var lastLogTsNs = time.Now().UnixNano() processEventFnWithOffset := pb.AddOffsetFunc(func(resp *filer_pb.SubscribeMetadataResponse) error { - if resp.EventNotification.NewEntry != nil { - if *option.storageClass == "" { - delete(resp.EventNotification.NewEntry.Extended, s3_constants.AmzStorageClass) - } else { - resp.EventNotification.NewEntry.Extended[s3_constants.AmzStorageClass] = []byte(*option.storageClass) - } - } - processor.AddSyncJob(resp) return nil }, 3*time.Second, func(counter int64, lastTsNs int64) error { @@ -177,10 +169,10 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(parentPath, entryName), remoteStorageMountLocation) if message.NewEntry.IsDirectory { glog.V(0).Infof("mkdir %s", remote_storage.FormatLocation(dest)) - return client.WriteDirectory(dest, message.NewEntry) + return client.WriteDirectory(dest, remoteWriteEntry(message.NewEntry, *option.storageClass)) } glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest)) - remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, dest) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, remoteWriteEntry(message.NewEntry, *option.storageClass), dest) if errors.Is(writeErr, errSuperseded) { glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr) return nil @@ -215,7 +207,7 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem return client.DeleteFile(dest) } if message.OldEntry != nil && message.NewEntry != nil { - return processUpdateEvent(option, filerSource, client, mountedDir, remoteStorageMountLocation, resp) + return processUpdateEvent(option, filerSource, *option.storageClass, client, mountedDir, remoteStorageMountLocation, resp) } return nil @@ -226,6 +218,7 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem func processUpdateEvent( filerClient filer_pb.FilerClient, filerSource filer_pb.FilerClient, + storageClass string, client remote_storage.RemoteStorageClient, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, @@ -246,7 +239,7 @@ func processUpdateEvent( return nil } if message.NewEntry.IsDirectory { - return client.WriteDirectory(dest, message.NewEntry) + return client.WriteDirectory(dest, remoteWriteEntry(message.NewEntry, storageClass)) } if isMetadataOnlyUpdate(resp.Directory, message) { remoteEntry, err := liveRemoteEntry(filerClient, message.NewParentPath, message.NewEntry) @@ -259,7 +252,7 @@ func processUpdateEvent( } if remoteEntry != nil { glog.V(2).Infof("update meta: %+v", resp) - return client.UpdateFileMetadata(dest, message.OldEntry, message.NewEntry) + return client.UpdateFileMetadata(dest, message.OldEntry, remoteWriteEntry(message.NewEntry, storageClass)) } glog.V(0).Infof("never replicated, uploading %s", remote_storage.FormatLocation(dest)) } @@ -279,7 +272,7 @@ func processUpdateEvent( } } } - remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, message.NewEntry, dest) + remoteEntry, writeErr := retriedWriteFile(client, filerSource, message.NewParentPath, remoteWriteEntry(message.NewEntry, storageClass), dest) if errors.Is(writeErr, errSuperseded) { glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr) return nil @@ -419,6 +412,23 @@ func shouldSendToRemote(entry *filer_pb.Entry) bool { return false } +// remoteWriteEntry returns the entry as remote storage should see it: the +// storage class attribute is dropped, or overridden by -storageClass. The +// event entry is left untouched so updateLocalEntry still compares the entry +// the filer stored. +func remoteWriteEntry(entry *filer_pb.Entry, storageClass string) *filer_pb.Entry { + clone := proto.Clone(entry).(*filer_pb.Entry) + if storageClass == "" { + delete(clone.Extended, s3_constants.AmzStorageClass) + } else { + if clone.Extended == nil { + clone.Extended = map[string][]byte{} + } + clone.Extended[s3_constants.AmzStorageClass] = []byte(storageClass) + } + return clone +} + // updateLocalEntry stamps the entry an event described with its RemoteEntry. // The write carries IF_ENTRY_EQUAL over the event's entry: the filer deletes // every stored chunk absent from an updated entry, so a snapshot older than diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 5d45c04283..7141c2a2ca 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -686,7 +686,7 @@ func TestRenameWithInheritedRemoteEntryWritesNewKey(t *testing.T) { remote := &recordingRemote{} filerClient := &stubFilerClient{} - if err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp); err != nil { + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp); err != nil { t.Fatal(err) } @@ -733,7 +733,7 @@ func TestRenameRemoteOnlyEntrySkipsEmptyUpload(t *testing.T) { remote := &recordingRemote{} filerClient := &stubFilerClient{} - if err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp); err != nil { + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp); err != nil { t.Fatal(err) } if len(remote.writes) != 0 { @@ -766,7 +766,7 @@ func TestRenameDeleteOldKeyFailureReturnsError(t *testing.T) { deleteErr := errors.New("AccessDenied: Access Denied") remote := &recordingRemote{deleteErr: deleteErr} filerClient := &stubFilerClient{} - err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp) + err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp) if !errors.Is(err, deleteErr) { t.Errorf("err = %v, want the delete failure returned so the event is retried", err) } @@ -802,7 +802,7 @@ func TestRenameDeleteOldKeyNotFoundStillWrites(t *testing.T) { remote := &recordingRemote{deleteErr: remote_storage.ErrRemoteObjectNotFound} filerClient := &stubFilerClient{} - if err := processUpdateEvent(filerClient, filerClient, remote, mountedDir, mountLoc, resp); err != nil { + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, resp); err != nil { t.Fatalf("err = %v, want nil: an already-deleted old key must not block the write", err) } wantWrite := &remote_pb.RemoteStorageLocation{Name: "gcs", Bucket: "bucket", Path: "/b/dst/probe.bin"} @@ -829,6 +829,38 @@ func TestIfEntryEqualCarriesTheEventEntry(t *testing.T) { } } +// The remote-bound entry loses (or gains) the storage class attribute while +// the event entry keeps it, so IF_ENTRY_EQUAL still sees the entry the filer +// stored — otherwise every stamp on an S3-written object reads as stale. +func TestRemoteWriteEntryLeavesEventEntryUntouched(t *testing.T) { + entry := &filer_pb.Entry{ + Name: "f", + Attributes: &filer_pb.FuseAttributes{Mtime: 1}, + Extended: map[string][]byte{s3_constants.AmzStorageClass: []byte("STANDARD")}, + } + + stripped := remoteWriteEntry(entry, "") + if _, ok := stripped.Extended[s3_constants.AmzStorageClass]; ok { + t.Fatalf("remote entry still carries %s", s3_constants.AmzStorageClass) + } + if string(entry.Extended[s3_constants.AmzStorageClass]) != "STANDARD" { + t.Fatalf("event entry Extended mutated: %v", entry.Extended) + } + + overridden := remoteWriteEntry(entry, "GLACIER") + if got := string(overridden.Extended[s3_constants.AmzStorageClass]); got != "GLACIER" { + t.Fatalf("override = %q, want GLACIER", got) + } + if string(entry.Extended[s3_constants.AmzStorageClass]) != "STANDARD" { + t.Fatalf("event entry Extended mutated: %v", entry.Extended) + } + + bare := &filer_pb.Entry{Name: "g", Attributes: &filer_pb.FuseAttributes{Mtime: 1}} + if got := string(remoteWriteEntry(bare, "GLACIER").Extended[s3_constants.AmzStorageClass]); got != "GLACIER" { + t.Fatalf("override on nil Extended = %q, want GLACIER", got) + } +} + func TestIsFailedPrecondition(t *testing.T) { refused := status.Error(codes.FailedPrecondition, "precondition failed: /buckets/b/f") cases := []struct {