diff --git a/weed/filer/filer_lazy_remote.go b/weed/filer/filer_lazy_remote.go index 7c642cd6a..0cf4f6eff 100644 --- a/weed/filer/filer_lazy_remote.go +++ b/weed/filer/filer_lazy_remote.go @@ -225,3 +225,48 @@ func (f *Filer) maybeDeleteFromRemote(ctx context.Context, entry *Entry) (bool, glog.V(3).InfofCtx(ctx, "maybeDeleteFromRemote: deleted %s from remote", entry.FullPath) return true, nil } + +// 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 +// entry, and with no local data nothing would write the new key. It returns +// nil when no copy is needed: the entry has local data, which the remote sync +// uploads to the new key, it is empty, which the sync writes as an empty +// object, or it is not under a remote mount. A rename that +// moves such an entry out of its mount, or onto a remote that cannot copy, is +// refused, since it would delete the only copy of the content. +func (f *Filer) CopyRemoteOnlyEntry(ctx context.Context, entry *Entry, oldPath, newPath util.FullPath) (*filer_pb.RemoteEntry, error) { + if f.RemoteStorage == nil || entry.IsDirectory() || !entry.IsInRemoteOnly() || len(entry.Content) > 0 { + return nil, nil + } + + mountDir, remoteLoc := f.RemoteStorage.FindMountDirectory(oldPath) + if remoteLoc == nil { + return nil, nil + } + if newMountDir, _ := f.RemoteStorage.FindMountDirectory(newPath); newMountDir != mountDir { + return nil, fmt.Errorf("%s exists only on the remote of mount %s and cannot be renamed out of it; cache it first", oldPath, mountDir) + } + + remoteConf, found := f.RemoteStorage.GetRemoteStorageConf(remoteLoc.Name) + if !found { + return nil, fmt.Errorf("resolve remote storage client for %s: not found", oldPath) + } + client, clientErr := f.buildRemoteStorageClient(ctx, remoteConf) + if clientErr != nil { + return nil, fmt.Errorf("resolve remote storage client for %s: %w", oldPath, clientErr) + } + copier, ok := client.(remote_storage.RemoteStorageObjectCopier) + if !ok { + return nil, fmt.Errorf("%s exists only on the remote and remote storage type %s cannot copy it to the new name; cache it first", oldPath, remoteConf.Type) + } + + src := MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, oldPath) + dst := MapFullPathToRemoteStorageLocation(mountDir, remoteLoc, newPath) + remoteEntry, err := copier.CopyFile(src, dst) + if err != nil { + return nil, fmt.Errorf("copy remote %s to %s: %w", remote_storage.FormatLocation(src), remote_storage.FormatLocation(dst), err) + } + glog.V(3).InfofCtx(ctx, "CopyRemoteOnlyEntry: copied %s to %s", remote_storage.FormatLocation(src), remote_storage.FormatLocation(dst)) + return remoteEntry, nil +} diff --git a/weed/remote_storage/remote_storage.go b/weed/remote_storage/remote_storage.go index f1845e1b0..0cefe362a 100644 --- a/weed/remote_storage/remote_storage.go +++ b/weed/remote_storage/remote_storage.go @@ -110,6 +110,14 @@ type RemoteStorageStreamReader interface { ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (reader io.ReadCloser, err error) } +// RemoteStorageObjectCopier is an optional interface for remote storage clients +// that copy an object to another key within the storage, without the content +// passing through the client. A rename of an entry whose content exists only +// on the remote needs it. +type RemoteStorageObjectCopier interface { + CopyFile(src *remote_pb.RemoteStorageLocation, dst *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) +} + // CacheWaitTimeout is how long a read of an uncached remote-only object waits // for the local cache before serving another way: small files wait longer since // their cache completes quickly, large files fail fast for better TTFB. A mount diff --git a/weed/remote_storage/s3/s3_storage_client.go b/weed/remote_storage/s3/s3_storage_client.go index a57a063cf..bcc9464cc 100644 --- a/weed/remote_storage/s3/s3_storage_client.go +++ b/weed/remote_storage/s3/s3_storage_client.go @@ -610,7 +610,7 @@ func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageL copyInput := &s3.CopyObjectInput{ Bucket: aws.String(loc.Bucket), Key: aws.String(key), - CopySource: aws.String(url.PathEscape(loc.Bucket + "/" + key)), + CopySource: aws.String(s3CopySource(loc.Bucket, key)), MetadataDirective: aws.String(s3.MetadataDirectiveReplace), Metadata: headOut.Metadata, ContentType: headOut.ContentType, @@ -663,6 +663,160 @@ func (s *s3RemoteStorageClient) UpdateFileMetadata(loc *remote_pb.RemoteStorageL } return } + +// s3CopyPartSize is the smallest part a multipart copy uses; parts grow so +// the largest object stays within the 10000-part limit +const s3CopyPartSize = 512 * 1024 * 1024 + +// CopyFile copies the object at src to dst on the server side, with a single +// CopyObject up to the 5 GB limit and a multipart copy above it. The copy +// keeps the source's metadata. +func (s *s3RemoteStorageClient) CopyFile(src *remote_pb.RemoteStorageLocation, dst *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) { + srcKey, dstKey := src.Path[1:], dst.Path[1:] + headOut, err := s.conn.HeadObject(&s3.HeadObjectInput{ + Bucket: aws.String(src.Bucket), + Key: aws.String(srcKey), + }) + if err != nil { + if reqErr, ok := err.(awserr.RequestFailure); ok && reqErr.StatusCode() == http.StatusNotFound { + return nil, remote_storage.ErrRemoteObjectNotFound + } + return nil, fmt.Errorf("stat s3 %s/%s before copy: %w", src.Bucket, srcKey, err) + } + copySource := aws.String(s3CopySource(src.Bucket, srcKey)) + var storageClass *string + if s.conf.S3StorageClass != "" { + storageClass = aws.String(s.conf.S3StorageClass) + } + + if size := aws.Int64Value(headOut.ContentLength); size <= s3CopyObjectSizeLimit { + if _, err = s.conn.CopyObject(&s3.CopyObjectInput{ + Bucket: aws.String(dst.Bucket), + Key: aws.String(dstKey), + CopySource: copySource, + StorageClass: storageClass, + }); err != nil { + return nil, fmt.Errorf("copy s3 %s/%s to %s/%s: %w", src.Bucket, srcKey, dst.Bucket, dstKey, err) + } + } else if err = s.multipartCopy(headOut, src, copySource, size, dst.Bucket, dstKey, storageClass); err != nil { + return nil, fmt.Errorf("copy s3 %s/%s to %s/%s: %w", src.Bucket, srcKey, dst.Bucket, dstKey, err) + } + + return s.readFileRemoteEntry(dst) +} + +// s3CopySource escapes each segment of the copy source and keeps the slashes, +// which every S3 implementation reads back +func s3CopySource(bucket, key string) string { + segments := strings.Split(bucket+"/"+key, "/") + for i, segment := range segments { + segments[i] = url.PathEscape(segment) + } + return strings.Join(segments, "/") +} + +// s3TaggingHeader encodes tags as the query string x-amz-tagging takes, with +// spaces as %20: a + is read back as a space by some implementations and +// literally by others +func s3TaggingHeader(tagSet []*s3.Tag) string { + escape := func(s string) string { + return strings.ReplaceAll(url.QueryEscape(s), "+", "%20") + } + pairs := make([]string, 0, len(tagSet)) + for _, tag := range tagSet { + pairs = append(pairs, escape(aws.StringValue(tag.Key))+"="+escape(aws.StringValue(tag.Value))) + } + return strings.Join(pairs, "&") +} + +// multipartCopy copies an object above the CopyObject limit part by part. A +// multipart upload does not inherit the source's metadata or tags, so they are +// carried over like CopyObject would: metadata from the source's HeadObject, +// tags when the remote supports tagging. +func (s *s3RemoteStorageClient) multipartCopy(headOut *s3.HeadObjectOutput, src *remote_pb.RemoteStorageLocation, copySource *string, size int64, bucket, key string, storageClass *string) error { + createInput := &s3.CreateMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + Metadata: headOut.Metadata, + ContentType: headOut.ContentType, + ContentEncoding: headOut.ContentEncoding, + CacheControl: headOut.CacheControl, + ContentDisposition: headOut.ContentDisposition, + ContentLanguage: headOut.ContentLanguage, + WebsiteRedirectLocation: headOut.WebsiteRedirectLocation, + StorageClass: storageClass, + } + if headOut.Expires != nil { + if expires, parseErr := http.ParseTime(*headOut.Expires); parseErr == nil { + createInput.Expires = aws.Time(expires) + } + } + if s.conf.S3SupportTagging { + tagOut, tagErr := s.conn.GetObjectTagging(&s3.GetObjectTaggingInput{ + Bucket: aws.String(src.Bucket), + Key: aws.String(src.Path[1:]), + }) + if tagErr != nil { + return fmt.Errorf("read tags: %w", tagErr) + } + if len(tagOut.TagSet) > 0 { + createInput.Tagging = aws.String(s3TaggingHeader(tagOut.TagSet)) + } + } + createOut, err := s.conn.CreateMultipartUpload(createInput) + if err != nil { + return err + } + + partSize := int64(s3CopyPartSize) + for partSize*10000 < size { + partSize *= 2 + } + var parts []*s3.CompletedPart + for offset, partNumber := int64(0), int64(1); offset < size; offset, partNumber = offset+partSize, partNumber+1 { + partOut, partErr := s.conn.UploadPartCopy(&s3.UploadPartCopyInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + UploadId: createOut.UploadId, + PartNumber: aws.Int64(partNumber), + CopySource: copySource, + CopySourceRange: aws.String(fmt.Sprintf("bytes=%d-%d", offset, min(offset+partSize, size)-1)), + }) + if partErr == nil && partOut.CopyPartResult == nil { + partErr = errors.New("no part result") + } + if partErr != nil { + s.abortMultipartUpload(bucket, key, createOut.UploadId) + return fmt.Errorf("copy part %d: %w", partNumber, partErr) + } + parts = append(parts, &s3.CompletedPart{ + ETag: partOut.CopyPartResult.ETag, + PartNumber: aws.Int64(partNumber), + }) + } + + if _, err = s.conn.CompleteMultipartUpload(&s3.CompleteMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + UploadId: createOut.UploadId, + MultipartUpload: &s3.CompletedMultipartUpload{Parts: parts}, + }); err != nil { + s.abortMultipartUpload(bucket, key, createOut.UploadId) + return err + } + return nil +} + +func (s *s3RemoteStorageClient) abortMultipartUpload(bucket, key string, uploadId *string) { + if _, err := s.conn.AbortMultipartUpload(&s3.AbortMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + UploadId: uploadId, + }); err != nil { + glog.Warningf("abort multipart copy to s3 %s/%s: %v", bucket, key, err) + } +} + func (s *s3RemoteStorageClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) (err error) { _, err = s.conn.DeleteObject(&s3.DeleteObjectInput{ Bucket: aws.String(loc.Bucket), diff --git a/weed/remote_storage/s3/s3_storage_client_copy_test.go b/weed/remote_storage/s3/s3_storage_client_copy_test.go new file mode 100644 index 000000000..f30af7bee --- /dev/null +++ b/weed/remote_storage/s3/s3_storage_client_copy_test.go @@ -0,0 +1,160 @@ +package s3 + +import ( + "errors" + "net/http" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/awserr" + awss3 "github.com/aws/aws-sdk-go/service/s3" + "github.com/aws/aws-sdk-go/service/s3/s3iface" + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" + "github.com/seaweedfs/seaweedfs/weed/remote_storage" + "github.com/stretchr/testify/require" +) + +// copyObjectMock serves HeadObject for a source of size bytes and records the +// copy calls. +type copyObjectMock struct { + s3iface.S3API + size int64 + missing bool + partErr error + copyInputs []*awss3.CopyObjectInput + partInputs []*awss3.UploadPartCopyInput + createInput *awss3.CreateMultipartUploadInput + tagInputs []*awss3.GetObjectTaggingInput + completed *awss3.CompleteMultipartUploadInput + aborted bool +} + +func (m *copyObjectMock) HeadObject(input *awss3.HeadObjectInput) (*awss3.HeadObjectOutput, error) { + if m.missing { + return nil, awserr.NewRequestFailure(awserr.New("NotFound", "not found", nil), http.StatusNotFound, "") + } + return &awss3.HeadObjectOutput{ + ContentLength: aws.Int64(m.size), + ContentType: aws.String("image/jpeg"), + Expires: aws.String("Wed, 21 Oct 2026 07:28:00 GMT"), + WebsiteRedirectLocation: aws.String("/elsewhere"), + ETag: aws.String(`"etag-` + aws.StringValue(input.Key) + `"`), + }, nil +} + +func (m *copyObjectMock) GetObjectTagging(input *awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) { + m.tagInputs = append(m.tagInputs, input) + return &awss3.GetObjectTaggingOutput{TagSet: []*awss3.Tag{ + {Key: aws.String("album"), Value: aws.String("2026 & co")}, + {Key: aws.String("a+b"), Value: aws.String("x=y")}, + }}, nil +} + +func (m *copyObjectMock) CopyObject(input *awss3.CopyObjectInput) (*awss3.CopyObjectOutput, error) { + m.copyInputs = append(m.copyInputs, input) + return &awss3.CopyObjectOutput{}, nil +} + +func (m *copyObjectMock) CreateMultipartUpload(input *awss3.CreateMultipartUploadInput) (*awss3.CreateMultipartUploadOutput, error) { + m.createInput = input + return &awss3.CreateMultipartUploadOutput{UploadId: aws.String("upload-1")}, nil +} + +func (m *copyObjectMock) UploadPartCopy(input *awss3.UploadPartCopyInput) (*awss3.UploadPartCopyOutput, error) { + m.partInputs = append(m.partInputs, input) + if m.partErr != nil && len(m.partInputs) == 2 { + return nil, m.partErr + } + return &awss3.UploadPartCopyOutput{CopyPartResult: &awss3.CopyPartResult{ETag: aws.String(`"part"`)}}, nil +} + +func (m *copyObjectMock) CompleteMultipartUpload(input *awss3.CompleteMultipartUploadInput) (*awss3.CompleteMultipartUploadOutput, error) { + m.completed = input + return &awss3.CompleteMultipartUploadOutput{}, nil +} + +func (m *copyObjectMock) AbortMultipartUpload(*awss3.AbortMultipartUploadInput) (*awss3.AbortMultipartUploadOutput, error) { + m.aborted = true + return &awss3.AbortMultipartUploadOutput{}, nil +} + +func newCopyTestClient(mock *copyObjectMock) *s3RemoteStorageClient { + return &s3RemoteStorageClient{conf: &remote_pb.RemoteConf{Name: "test"}, conn: mock} +} + +var ( + copySrc = &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/src/a b.jpg"} + copyDst = &remote_pb.RemoteStorageLocation{Name: "test", Bucket: "bucket", Path: "/dst/a b.jpg"} +) + +func TestS3ClientImplementsObjectCopier(t *testing.T) { + var _ remote_storage.RemoteStorageObjectCopier = &s3RemoteStorageClient{} +} + +func TestS3CopyFileUsesCopyObjectUpToTheLimit(t *testing.T) { + mock := ©ObjectMock{size: s3CopyObjectSizeLimit} + + remoteEntry, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst) + require.NoError(t, err) + + 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.Nil(t, mock.createInput) + // the RemoteEntry describes the destination + require.Equal(t, `"etag-dst/a b.jpg"`, remoteEntry.RemoteETag) + require.Equal(t, int64(s3CopyObjectSizeLimit), remoteEntry.RemoteSize) +} + +func TestS3CopyFileCopiesLargeObjectsInParts(t *testing.T) { + size := int64(s3CopyObjectSizeLimit) + 1 + mock := ©ObjectMock{size: size} + + _, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst) + require.NoError(t, err) + + require.Empty(t, mock.copyInputs) + 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, int64(1792567680), aws.TimeValue(mock.createInput.Expires).Unix()) + // tags only where the remote supports tagging + require.Empty(t, mock.tagInputs) + require.Nil(t, mock.createInput.Tagging) + 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)) + 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) +} + +func TestS3CopyFileCarriesTagsIntoAMultipartCopy(t *testing.T) { + mock := ©ObjectMock{size: s3CopyObjectSizeLimit + 1} + client := newCopyTestClient(mock) + client.conf.S3SupportTagging = true + + _, err := client.CopyFile(copySrc, copyDst) + require.NoError(t, err) + + require.Len(t, mock.tagInputs, 1) + require.Equal(t, "src/a b.jpg", aws.StringValue(mock.tagInputs[0].Key)) + require.Equal(t, "album=2026%20%26%20co&a%2Bb=x%3Dy", aws.StringValue(mock.createInput.Tagging)) +} + +func TestS3CopyFileAbortsAFailedMultipartCopy(t *testing.T) { + mock := ©ObjectMock{size: 2 * s3CopyObjectSizeLimit, partErr: errors.New("part failed")} + + _, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst) + require.ErrorContains(t, err, "part failed") + require.True(t, mock.aborted) + require.Nil(t, mock.completed) +} + +func TestS3CopyFileMissingSource(t *testing.T) { + mock := ©ObjectMock{missing: true} + + _, err := newCopyTestClient(mock).CopyFile(copySrc, copyDst) + require.ErrorIs(t, err, remote_storage.ErrRemoteObjectNotFound) + require.Empty(t, mock.copyInputs) +} diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index 020a59c71..ae097ee41 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/weed/server/filer_grpc_server_rename.go @@ -3,6 +3,7 @@ package weed_server import ( "context" "fmt" + "maps" "path/filepath" "time" @@ -235,8 +236,32 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee case !existingTarget.IsDirectory() && entry.IsDirectory(): return fmt.Errorf("%s: %w", existingTarget.FullPath, filer_pb.ErrExistingIsFile) } + } + + // a remote-only entry is copied on the remote before anything changes: + // deleting the old entry deletes the old object + copiedRemote, copyErr := fs.filer.CopyRemoteOnlyEntry(ctx, entry, oldPath, newPath) + if copyErr != nil { + return copyErr + } + remoteEntry := entry.Remote + if copiedRemote != nil { + remoteEntry = copiedRemote + } + + if existingTarget != nil { + targetCtx := filer.WithSuppressedMetadataEvents(ctx) + if copiedRemote != nil { + // the copy has replaced the target's object; keep it + targetCtx = filer.WithKeepRemoteObject(targetCtx) + existingTarget.Extended = maps.Clone(existingTarget.Extended) + if existingTarget.Extended == nil { + existingTarget.Extended = map[string][]byte{} + } + existingTarget.Extended[filer.ExtKeepRemoteObjectKey] = []byte("true") + } if deleteErr := fs.filer.DeleteEntryMetaAndData( - filer.WithSuppressedMetadataEvents(ctx), + targetCtx, newPath, false, false, @@ -258,7 +283,7 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee Content: entry.Content, HardLinkCounter: entry.HardLinkCounter, HardLinkId: entry.HardLinkId, - Remote: entry.Remote, + Remote: remoteEntry, Quota: entry.Quota, } if skipTargetLookup { diff --git a/weed/server/filer_grpc_server_rename_remote_test.go b/weed/server/filer_grpc_server_rename_remote_test.go new file mode 100644 index 000000000..27eab9e0a --- /dev/null +++ b/weed/server/filer_grpc_server_rename_remote_test.go @@ -0,0 +1,282 @@ +package weed_server + +import ( + "context" + "errors" + "sync" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" + "github.com/seaweedfs/seaweedfs/weed/remote_storage" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +// renameRemoteClient records the remote calls a rename makes. It cannot copy; +// copyingRenameRemoteClient adds CopyFile. +type renameRemoteClient struct { + remote_storage.RemoteStorageClient + mu sync.Mutex + calls []string + copyErr error +} + +func (c *renameRemoteClient) record(call string) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls = append(c.calls, call) +} + +func (c *renameRemoteClient) recorded() []string { + c.mu.Lock() + defer c.mu.Unlock() + return append([]string(nil), c.calls...) +} + +func (c *renameRemoteClient) StatFile(*remote_pb.RemoteStorageLocation) (*filer_pb.RemoteEntry, error) { + return nil, remote_storage.ErrRemoteObjectNotFound +} + +func (c *renameRemoteClient) ListDirectory(context.Context, *remote_pb.RemoteStorageLocation, remote_storage.VisitFunc) error { + return nil +} + +func (c *renameRemoteClient) DeleteFile(loc *remote_pb.RemoteStorageLocation) error { + c.record("delete " + loc.Bucket + loc.Path) + return nil +} + +func (c *renameRemoteClient) RemoveDirectory(loc *remote_pb.RemoteStorageLocation) error { + c.record("rmdir " + loc.Bucket + loc.Path) + return nil +} + +type copyingRenameRemoteClient struct { + *renameRemoteClient +} + +func (c copyingRenameRemoteClient) CopyFile(src, dst *remote_pb.RemoteStorageLocation) (*filer_pb.RemoteEntry, error) { + if c.copyErr != nil { + return nil, c.copyErr + } + c.record("copy " + src.Bucket + src.Path + " " + dst.Bucket + dst.Path) + return &filer_pb.RemoteEntry{StorageName: "r1", RemoteSize: 10, RemoteMtime: 1800000000, RemoteETag: `"copy"`}, nil +} + +// newRenameRemoteTestServer mounts /buckets/b, /buckets/c and /data/m on r1, +// through client. +func newRenameRemoteTestServer(t *testing.T, client remote_storage.RemoteStorageClient) (*FilerServer, *renameTestStore) { + t.Helper() + store := newRenameTestStore() + f := newRenameTestFiler(t, store) + f.BuildGuardedRemoteClient = func(context.Context, *remote_pb.RemoteConf, bool) (remote_storage.RemoteStorageClient, error) { + return client, nil + } + mapping, err := proto.Marshal(&remote_pb.RemoteStorageMapping{Mappings: map[string]*remote_pb.RemoteStorageLocation{ + "/buckets/b": {Name: "r1", Bucket: "origin", Path: "/"}, + "/buckets/c": {Name: "r1", Bucket: "other", Path: "/"}, + "/data/m": {Name: "r1", Bucket: "origin", Path: "/m"}, + }}) + require.NoError(t, err) + conf, err := proto.Marshal(&remote_pb.RemoteConf{Name: "r1", Type: "s3"}) + require.NoError(t, err) + for name, content := range map[string][]byte{filer.REMOTE_STORAGE_MOUNT_FILE: mapping, "r1" + filer.REMOTE_STORAGE_CONF_SUFFIX: conf} { + entry := newFileEntry(filer.DirectoryEtcRemote+"/"+name, 1) + entry.Content = content + store.entries[string(entry.FullPath)] = entry + } + require.NoError(t, f.RemoteStorage.LoadRemoteStorageConfigurationsAndMapping(f)) + for _, dir := range []string{"/buckets", "/buckets/b", "/buckets/b/src", "/buckets/b/dst", "/buckets/c", "/data", "/data/m"} { + store.entries[dir] = newDirectoryEntry(dir, 2) + } + return &FilerServer{filer: f, option: &FilerOption{}, entryLockTable: util.NewLockTable[util.FullPath]()}, store +} + +func remoteOnlyRenameEntry(path string, inode uint64) *filer.Entry { + entry := newFileEntry(path, inode) + entry.FileSize = 10 + entry.Remote = &filer_pb.RemoteEntry{StorageName: "r1", RemoteSize: 10, RemoteMtime: 1700000000, RemoteETag: `"orig"`} + return entry +} + +func renameFile(server *FilerServer, oldPath, newPath util.FullPath) error { + oldDir, oldName := oldPath.DirAndName() + newDir, newName := newPath.DirAndName() + _, err := server.AtomicRenameEntry(context.Background(), &filer_pb.AtomicRenameEntryRequest{ + OldDirectory: oldDir, + OldName: oldName, + NewDirectory: newDir, + NewName: newName, + }) + return err +} + +func TestRenameRemoteOnlyEntryCopiesBeforeDeletingTheOldObject(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) + + require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg")) + + assert.Equal(t, []string{"copy origin/src/a.jpg origin/dst/a.jpg", "delete origin/src/a.jpg"}, client.recorded()) + _, err := store.FindEntry(context.Background(), "/buckets/b/src/a.jpg") + assert.ErrorIs(t, err, filer_pb.ErrNotFound) + dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"copy"`, dst.Remote.RemoteETag) + assert.Equal(t, uint64(101), dst.Attr.Inode) +} + +func TestRenameRemoteOnlyEntryOverTargetKeepsTheCopy(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.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202) + queue := &captureQueue{} + swapNotificationQueue(t, queue) + + require.NoError(t, renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg")) + + // the target's delete must not take the copy that replaced its object + assert.Equal(t, []string{"copy origin/src/a.jpg origin/dst/a.jpg", "delete origin/src/a.jpg"}, client.recorded()) + events := queue.snapshot() + require.Len(t, events, 2) + assert.True(t, filer.IsMetadataOnlyDelete(events[0].notification.OldEntry), "the target's delete event must leave the remote object alone") + assert.False(t, filer.IsMetadataOnlyDelete(events[1].notification.OldEntry)) + dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + require.NoError(t, err) + assert.Equal(t, uint64(101), dst.Attr.Inode) + assert.Equal(t, `"copy"`, dst.Remote.RemoteETag) +} + +func TestRenameRemoteOnlyEntryCopyFailureChangesNothing(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{copyErr: errors.New("copy refused")}} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + store.entries["/buckets/b/dst/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/dst/a.jpg", 202) + + err := renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg") + require.ErrorContains(t, err, "copy refused") + + assert.Empty(t, client.recorded()) + src, err := store.FindEntry(context.Background(), "/buckets/b/src/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"orig"`, src.Remote.RemoteETag) + dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + require.NoError(t, err) + assert.Equal(t, uint64(202), dst.Attr.Inode) +} + +func TestRenameRemoteOnlyEntryWithoutCopierIsRefused(t *testing.T) { + client := &renameRemoteClient{} + server, store := newRenameRemoteTestServer(t, client) + store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101) + + err := renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg") + require.ErrorContains(t, err, "cannot copy") + + assert.Empty(t, client.recorded()) + _, err = store.FindEntry(context.Background(), "/buckets/b/src/a.jpg") + require.NoError(t, err) + _, err = store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") + assert.ErrorIs(t, err, filer_pb.ErrNotFound) +} + +func TestRenameRemoteOnlyEntryOutOfItsMountIsRefused(t *testing.T) { + for _, paths := range [][2]util.FullPath{ + {"/buckets/b/src/a.jpg", "/buckets/c/a.jpg"}, + {"/data/m/a.jpg", "/data/a.jpg"}, + } { + t.Run(string(paths[1]), func(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + store.entries[string(paths[0])] = remoteOnlyRenameEntry(string(paths[0]), 101) + + err := renameFile(server, paths[0], paths[1]) + require.ErrorContains(t, err, "cannot be renamed out of it") + + assert.Empty(t, client.recorded()) + _, err = store.FindEntry(context.Background(), paths[0]) + require.NoError(t, err) + _, err = store.FindEntry(context.Background(), paths[1]) + assert.ErrorIs(t, err, filer_pb.ErrNotFound) + }) + } +} + +func TestRenameEmptyRemoteEntryDoesNotCopy(t *testing.T) { + // nothing to lose: no copy, and no refusal even out of the mount or + // without a copier; the sync writes an empty object at a new key + client := &renameRemoteClient{} + server, store := newRenameRemoteTestServer(t, client) + for _, path := range []string{"/buckets/b/src/empty", "/data/m/empty"} { + entry := remoteOnlyRenameEntry(path, 101) + entry.FileSize = 0 + entry.Remote.RemoteSize = 0 + store.entries[path] = entry + } + + require.NoError(t, renameFile(server, "/buckets/b/src/empty", "/buckets/b/dst/empty")) + require.NoError(t, renameFile(server, "/data/m/empty", "/data/empty")) + + 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") + + err := renameFile(server, "/buckets/b/src/a.jpg", "/buckets/b/dst/a.jpg") + 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()) +} + +func TestRenameEntryWithLocalDataDoesNotCopy(t *testing.T) { + client := copyingRenameRemoteClient{&renameRemoteClient{}} + server, store := newRenameRemoteTestServer(t, client) + entry := remoteOnlyRenameEntry("/data/m/a.jpg", 101) + entry.Chunks = []*filer_pb.FileChunk{{FileId: "1,0123456789", Size: 10}} + store.entries["/data/m/a.jpg"] = entry + + // even out of the mount: the old object goes, the content stays local + require.NoError(t, renameFile(server, "/data/m/a.jpg", "/data/a.jpg")) + + assert.Equal(t, []string{"delete origin/m/a.jpg"}, client.recorded()) + dst, err := store.FindEntry(context.Background(), "/data/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"orig"`, dst.Remote.RemoteETag) + assert.Len(t, dst.GetChunks(), 1) +} + +func TestRenameDirectoryCopiesRemoteOnlyChildren(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) + local := remoteOnlyRenameEntry("/buckets/b/src/b.jpg", 102) + local.Content = []byte("0123456789") + store.entries["/buckets/b/src/b.jpg"] = local + + require.NoError(t, renameFile(server, "/buckets/b/src", "/buckets/b/moved")) + + 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", + }, client.recorded()) + moved, err := store.FindEntry(context.Background(), "/buckets/b/moved/a.jpg") + require.NoError(t, err) + assert.Equal(t, `"copy"`, moved.Remote.RemoteETag) + moved, err = store.FindEntry(context.Background(), "/buckets/b/moved/b.jpg") + require.NoError(t, err) + assert.Equal(t, `"orig"`, moved.Remote.RemoteETag) +}