diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index a241c1640..cd54393ac 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -195,7 +195,7 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem return updateLocalEntry(option, message.NewParentPath, message.NewEntry, remoteEntry) } if filer_pb.IsDelete(resp) { - return processDeleteEvent(client, mountedDir, remoteStorageMountLocation, resp) + return processDeleteEvent(filerSource, client, mountedDir, remoteStorageMountLocation, resp) } if message.OldEntry != nil && message.NewEntry != nil { return processUpdateEvent(option, filerSource, *option.storageClass, client, mountedDir, remoteStorageMountLocation, resp) @@ -212,8 +212,11 @@ func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.Rem // event the inline delete already handled), and GCS reports that case as // ErrRemoteObjectNotFound where S3 and Azure answer an idempotent success. // Returning the error would pin the sync offset on an event that has nothing -// left to do. +// left to do. A file whose path a remote-only entry has taken since keeps its +// object: a rename copied that entry's content to this key, and it exists +// nowhere else. func processDeleteEvent( + filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, @@ -232,6 +235,16 @@ func processDeleteEvent( } glog.V(2).Infof("delete: %+v", resp) dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation) + if !message.OldEntry.IsDirectory { + current, err := currentEntry(filerSource, resp.Directory, message.OldEntry.Name) + if err != nil { + return err + } + if isRemoteOnly(current) { + glog.V(0).Infof("keep %s: a remote-only entry has taken its path since", remote_storage.FormatLocation(dest)) + return nil + } + } if message.OldEntry.IsDirectory { glog.V(0).Infof("rmdir %s", remote_storage.FormatLocation(dest)) return client.RemoveDirectory(dest) diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 8c2b1e1da..de9d92aed 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -999,7 +999,7 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { t.Run("absent object", func(t *testing.T) { remote := &recordingRemote{deleteErr: remote_storage.ErrRemoteObjectNotFound} - if err := processDeleteEvent(remote, mountedDir, mountLoc, resp); err != nil { + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, resp); err != nil { t.Fatalf("err = %v, want nil: deleting an absent object is complete", err) } if len(remote.deletes) != 1 || !proto.Equal(remote.deletes[0], wantDelete) { @@ -1011,7 +1011,7 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { marked := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse) marked.EventNotification.OldEntry.Extended = map[string][]byte{filer.ExtKeepRemoteObjectKey: []byte("true")} remote := &recordingRemote{} - if err := processDeleteEvent(remote, mountedDir, mountLoc, marked); err != nil { + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, marked); err != nil { t.Fatalf("err = %v", err) } if len(remote.deletes) != 0 { @@ -1019,9 +1019,41 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { } }) + t.Run("a remote-only entry that took the path keeps the object", func(t *testing.T) { + // deleted, then a remote-only file renamed onto the path: the rename + // copied its content to this key before the sync got to the delete + remote := &recordingRemote{} + filerClient := &stubFilerClient{entry: &filer_pb.Entry{ + Name: "state.db.tmp", + Attributes: &filer_pb.FuseAttributes{FileSize: 10}, + RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 10}, + }} + if err := processDeleteEvent(filerClient, remote, mountedDir, mountLoc, resp); err != nil { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 0 { + t.Errorf("deletes = %+v, want none", remote.deletes) + } + }) + + t.Run("an entry with local data at the path does not stop the delete", func(t *testing.T) { + remote := &recordingRemote{} + filerClient := &stubFilerClient{entry: &filer_pb.Entry{ + Name: "state.db.tmp", + Content: []byte("local"), + RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 10}, + }} + if err := processDeleteEvent(filerClient, remote, mountedDir, mountLoc, resp); err != nil { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 1 { + t.Errorf("deletes = %+v, want one", remote.deletes) + } + }) + t.Run("other failure still fails", func(t *testing.T) { remote := &recordingRemote{deleteErr: errors.New("AccessDenied: Access Denied")} - if err := processDeleteEvent(remote, mountedDir, mountLoc, resp); err == nil { + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, resp); err == nil { t.Fatal("err = nil, want the delete failure") } }) diff --git a/weed/filer/filer_lazy_remote.go b/weed/filer/filer_lazy_remote.go index 0cf4f6eff..bbfe63306 100644 --- a/weed/filer/filer_lazy_remote.go +++ b/weed/filer/filer_lazy_remote.go @@ -226,6 +226,13 @@ func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool, return true, nil } +// DeleteFromRemote deletes the object, or for a directory every object, the +// entry maps to on its mounted remote storage. +func (f *Filer) DeleteFromRemote(ctx context.Context, entry *Entry) error { + _, err := f.maybeDeleteFromRemote(ctx, entry) + return err +} + // CopyRemoteOnlyEntry copies the remote object of a renamed entry whose // content exists only on the remote to the key of newPath, and returns the // RemoteEntry of the copy. The rename deletes the old key along with the old diff --git a/weed/remote_storage/s3/s3_storage_client.go b/weed/remote_storage/s3/s3_storage_client.go index bcc9464cc..4166df8a4 100644 --- a/weed/remote_storage/s3/s3_storage_client.go +++ b/weed/remote_storage/s3/s3_storage_client.go @@ -690,11 +690,14 @@ func (s *s3RemoteStorageClient) CopyFile(src *remote_pb.RemoteStorageLocation, d } if size := aws.Int64Value(headOut.ContentLength); size <= s3CopyObjectSizeLimit { + // a copy takes the bucket's default encryption unless told otherwise if _, err = s.conn.CopyObject(&s3.CopyObjectInput{ - Bucket: aws.String(dst.Bucket), - Key: aws.String(dstKey), - CopySource: copySource, - StorageClass: storageClass, + Bucket: aws.String(dst.Bucket), + Key: aws.String(dstKey), + CopySource: copySource, + StorageClass: storageClass, + ServerSideEncryption: headOut.ServerSideEncryption, + SSEKMSKeyId: headOut.SSEKMSKeyId, }); err != nil { return nil, fmt.Errorf("copy s3 %s/%s to %s/%s: %w", src.Bucket, srcKey, dst.Bucket, dstKey, err) } @@ -744,6 +747,8 @@ func (s *s3RemoteStorageClient) multipartCopy(headOut *s3.HeadObjectOutput, src ContentDisposition: headOut.ContentDisposition, ContentLanguage: headOut.ContentLanguage, WebsiteRedirectLocation: headOut.WebsiteRedirectLocation, + ServerSideEncryption: headOut.ServerSideEncryption, + SSEKMSKeyId: headOut.SSEKMSKeyId, StorageClass: storageClass, } if headOut.Expires != nil { @@ -781,6 +786,9 @@ func (s *s3RemoteStorageClient) multipartCopy(headOut *s3.HeadObjectOutput, src PartNumber: aws.Int64(partNumber), CopySource: copySource, CopySourceRange: aws.String(fmt.Sprintf("bytes=%d-%d", offset, min(offset+partSize, size)-1)), + // every part from the object the HeadObject saw, not a mix of + // versions if it is overwritten mid-copy + CopySourceIfMatch: headOut.ETag, }) if partErr == nil && partOut.CopyPartResult == nil { partErr = errors.New("no part result") diff --git a/weed/remote_storage/s3/s3_storage_client_copy_test.go b/weed/remote_storage/s3/s3_storage_client_copy_test.go index f30af7bee..a20fbf57a 100644 --- a/weed/remote_storage/s3/s3_storage_client_copy_test.go +++ b/weed/remote_storage/s3/s3_storage_client_copy_test.go @@ -38,6 +38,8 @@ func (m *copyObjectMock) HeadObject(input *awss3.HeadObjectInput) (*awss3.HeadOb ContentType: aws.String("image/jpeg"), Expires: aws.String("Wed, 21 Oct 2026 07:28:00 GMT"), WebsiteRedirectLocation: aws.String("/elsewhere"), + ServerSideEncryption: aws.String("aws:kms"), + SSEKMSKeyId: aws.String("key-1"), ETag: aws.String(`"etag-` + aws.StringValue(input.Key) + `"`), }, nil } @@ -100,6 +102,8 @@ func TestS3CopyFileUsesCopyObjectUpToTheLimit(t *testing.T) { require.Len(t, mock.copyInputs, 1) require.Equal(t, "bucket/src/a%20b.jpg", aws.StringValue(mock.copyInputs[0].CopySource)) require.Equal(t, "dst/a b.jpg", aws.StringValue(mock.copyInputs[0].Key)) + require.Equal(t, "aws:kms", aws.StringValue(mock.copyInputs[0].ServerSideEncryption)) + require.Equal(t, "key-1", aws.StringValue(mock.copyInputs[0].SSEKMSKeyId)) require.Nil(t, mock.createInput) // the RemoteEntry describes the destination require.Equal(t, `"etag-dst/a b.jpg"`, remoteEntry.RemoteETag) @@ -117,6 +121,8 @@ func TestS3CopyFileCopiesLargeObjectsInParts(t *testing.T) { require.NotNil(t, mock.createInput) require.Equal(t, "image/jpeg", aws.StringValue(mock.createInput.ContentType)) require.Equal(t, "/elsewhere", aws.StringValue(mock.createInput.WebsiteRedirectLocation)) + require.Equal(t, "aws:kms", aws.StringValue(mock.createInput.ServerSideEncryption)) + require.Equal(t, "key-1", aws.StringValue(mock.createInput.SSEKMSKeyId)) require.Equal(t, int64(1792567680), aws.TimeValue(mock.createInput.Expires).Unix()) // tags only where the remote supports tagging require.Empty(t, mock.tagInputs) @@ -124,6 +130,9 @@ func TestS3CopyFileCopiesLargeObjectsInParts(t *testing.T) { require.Len(t, mock.partInputs, 11) require.Equal(t, "bytes=0-536870911", aws.StringValue(mock.partInputs[0].CopySourceRange)) require.Equal(t, "bytes=5368709120-5368709120", aws.StringValue(mock.partInputs[10].CopySourceRange)) + for _, part := range mock.partInputs { + require.Equal(t, `"etag-src/a b.jpg"`, aws.StringValue(part.CopySourceIfMatch)) + } require.Len(t, mock.completed.MultipartUpload.Parts, 11) require.Equal(t, int64(11), aws.Int64Value(mock.completed.MultipartUpload.Parts[10].PartNumber)) require.False(t, mock.aborted) diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index ae097ee41..8c67f7bba 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/weed/server/filer_grpc_server_rename.go @@ -150,9 +150,18 @@ type metadataEvent struct { oldEntry *filer.Entry newEntry *filer.Entry deleteChunks bool + // the old entry's remote object is deleted once the rename has committed: + // it, or an object under it, was copied to the new key, and a rename that + // fails must not lose it + deleteOldRemote bool } func (event metadataEvent) notify(f *filer.Filer, ctx context.Context, signatures []int32) { + if event.deleteOldRemote { + if err := f.DeleteFromRemote(ctx, event.oldEntry); err != nil { + glog.WarningfCtx(ctx, "delete remote object of renamed %s: %v", event.oldEntry.FullPath, err) + } + } f.NotifyUpdateEvent(ctx, event.oldEntry, event.newEntry, event.deleteChunks, false, signatures) } @@ -336,19 +345,33 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee }) } *metadataEvents = append(*metadataEvents, metadataEvent{ - oldEntry: sourceEntry, - newEntry: newEntry, + oldEntry: sourceEntry, + newEntry: newEntry, + deleteOldRemote: copiedRemote != nil, }) + selfEvent := len(*metadataEvents) - 1 if moveFolderSubEntries != nil { if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil { return moveChildrenErr } + // removing the old directory on the remote removes every object under + // it, so it waits for the commit too once a child was copied + for _, event := range (*metadataEvents)[selfEvent+1:] { + if event.deleteOldRemote { + (*metadataEvents)[selfEvent].deleteOldRemote = true + break + } + } } // delete old entry ctx = context.WithValue(ctx, "OP", "MV") - deleteErr := fs.filer.DeleteEntryMetaAndData(filer.WithSuppressedMetadataEvents(ctx), oldPath, false, false, false, false, signatures, 0) + deleteCtx := filer.WithSuppressedMetadataEvents(ctx) + if (*metadataEvents)[selfEvent].deleteOldRemote { + deleteCtx = filer.WithKeepRemoteObject(deleteCtx) + } + deleteErr := fs.filer.DeleteEntryMetaAndData(deleteCtx, oldPath, false, false, false, false, signatures, 0) if deleteErr != nil { return deleteErr } diff --git a/weed/server/filer_grpc_server_rename_remote_test.go b/weed/server/filer_grpc_server_rename_remote_test.go index 27eab9e0a..dc57de8c3 100644 --- a/weed/server/filer_grpc_server_rename_remote_test.go +++ b/weed/server/filer_grpc_server_rename_remote_test.go @@ -226,18 +226,26 @@ func TestRenameEmptyRemoteEntryDoesNotCopy(t *testing.T) { assert.Equal(t, []string{"delete origin/src/empty", "delete origin/m/empty"}, client.recorded()) } -func TestRenameRemoteOnlyEntryFailingAfterTheCopyKeepsTheContent(t *testing.T) { - client := copyingRenameRemoteClient{&renameRemoteClient{}} - server, store := newRenameRemoteTestServer(t, client) - store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) - store.commitErr = errors.New("commit failed") +func TestRenameRemoteOnlyEntryFailingAfterTheCopyKeepsTheSource(t *testing.T) { + for _, src := range []util.FullPath{"/buckets/b/src/a.jpg", "/buckets/b/src"} { + t.Run(string(src), func(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + store.commitErr = errors.New("commit failed") - err := renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg") - require.ErrorContains(t, err, "commit failed") + err := renameFile(server, src, "/buckets/b/moved") + require.ErrorContains(t, err, "commit failed") - // remote changes are not rolled back with the store: the old object is - // gone, but only after the copy at the new key was written - assert.Equal(t, []string{"copy origin/src/a.jpg origin/dst/a.jpg", "delete origin/src/a.jpg"}, client.recorded()) + // the old object, and the old directory holding it, are deleted + // only once the rename has committed + dst := "origin/moved" + if src == "/buckets/b/src" { + dst = "origin/moved/a.jpg" + } + assert.Equal(t, []string{"copy origin/src/a.jpg " + dst}, client.recorded()) + }) + } } func TestRenameEntryWithLocalDataDoesNotCopy(t *testing.T) { @@ -267,11 +275,13 @@ func TestRenameDirectoryCopiesRemoteOnlyChildren(t *testing.T) { require.NoError(t, renameFile(server, "/buckets/b/src", "/buckets/b/moved")) + // b.jpg's old object goes with its entry; the old directory and a.jpg's + // old object wait for the commit assert.Equal(t, []string{ "copy origin/src/a.jpg origin/moved/a.jpg", - "delete origin/src/a.jpg", "delete origin/src/b.jpg", "rmdir origin/src", + "delete origin/src/a.jpg", }, client.recorded()) moved, err := store.FindEntry(context.Background(), "/buckets/b/moved/a.jpg") require.NoError(t, err)