seaweedfs/weed/server/filer_grpc_server_rename.go

232 lines
7.1 KiB
Go
Raw Normal View History

package weed_server
import (
"context"
"fmt"
2020-03-23 07:01:34 +00:00
"path/filepath"
2021-10-17 11:22:42 +00:00
"time"
2020-03-23 07:01:34 +00:00
"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/util"
)
func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.AtomicRenameEntryRequest) (*filer_pb.AtomicRenameEntryResponse, error) {
2019-05-03 21:12:51 +00:00
glog.V(1).Infof("AtomicRenameEntry %v", req)
2020-12-11 07:50:32 +00:00
oldParent := util.FullPath(filepath.ToSlash(req.OldDirectory))
newParent := util.FullPath(filepath.ToSlash(req.NewDirectory))
if err := fs.filer.CanRename(oldParent, newParent, req.OldName); err != nil {
2020-12-11 07:50:32 +00:00
return nil, err
}
ctx, err := fs.filer.BeginTransaction(ctx)
if err != nil {
return nil, err
}
oldEntry, err := fs.filer.FindEntry(ctx, oldParent.Child(req.OldName))
if err != nil {
fs.filer.RollbackTransaction(ctx)
return nil, fmt.Errorf("%s/%s not found: %v", req.OldDirectory, req.OldName, err)
}
2021-10-17 11:22:42 +00:00
moveErr := fs.moveEntry(ctx, nil, oldParent, oldEntry, newParent, req.NewName, req.Signatures)
if moveErr != nil {
fs.filer.RollbackTransaction(ctx)
2020-06-10 17:59:25 +00:00
return nil, fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr)
} else {
if commitError := fs.filer.CommitTransaction(ctx); commitError != nil {
fs.filer.RollbackTransaction(ctx)
2020-06-10 17:59:25 +00:00
return nil, fmt.Errorf("%s/%s move commit error: %v", req.OldDirectory, req.OldName, commitError)
}
}
return &filer_pb.AtomicRenameEntryResponse{}, nil
}
2021-10-17 11:22:42 +00:00
func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, stream filer_pb.SeaweedFiler_StreamRenameEntryServer) (err error) {
2021-10-17 11:22:42 +00:00
glog.V(1).Infof("StreamRenameEntry %v", req)
oldParent := util.FullPath(filepath.ToSlash(req.OldDirectory))
newParent := util.FullPath(filepath.ToSlash(req.NewDirectory))
if err := fs.filer.CanRename(oldParent, newParent, req.OldName); err != nil {
2021-10-17 11:22:42 +00:00
return err
}
ctx := context.Background()
ctx, err = fs.filer.BeginTransaction(ctx)
if err != nil {
return err
}
oldEntry, err := fs.filer.FindEntry(ctx, oldParent.Child(req.OldName))
if err != nil {
fs.filer.RollbackTransaction(ctx)
return fmt.Errorf("%s/%s not found: %v", req.OldDirectory, req.OldName, err)
}
2022-01-11 11:23:03 +00:00
if oldEntry.IsDirectory() {
2022-01-13 10:01:53 +00:00
// follow https://pubs.opengroup.org/onlinepubs/000095399/functions/rename.html
2022-01-11 11:23:03 +00:00
targetDir := newParent.Child(req.NewName)
newEntry, err := fs.filer.FindEntry(ctx, targetDir)
if err == nil {
if !newEntry.IsDirectory() {
fs.filer.RollbackTransaction(ctx)
return fmt.Errorf("%s is not directory", targetDir)
}
if entries, _, _ := fs.filer.ListDirectoryEntries(context.Background(), targetDir, "", false, 1, "", "", ""); len(entries) > 0 {
return fmt.Errorf("%s is not empty", targetDir)
}
}
}
2021-10-17 11:22:42 +00:00
moveErr := fs.moveEntry(ctx, stream, oldParent, oldEntry, newParent, req.NewName, req.Signatures)
if moveErr != nil {
fs.filer.RollbackTransaction(ctx)
return fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr)
} else {
if commitError := fs.filer.CommitTransaction(ctx); commitError != nil {
fs.filer.RollbackTransaction(ctx)
return fmt.Errorf("%s/%s move commit error: %v", req.OldDirectory, req.OldName, commitError)
}
}
return nil
}
func (fs *FilerServer) moveEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32) error {
if err := fs.moveSelfEntry(ctx, stream, oldParent, entry, newParent, newName, func() error {
if entry.IsDirectory() {
2021-10-17 11:22:42 +00:00
if err := fs.moveFolderSubEntries(ctx, stream, oldParent, entry, newParent, newName, signatures); err != nil {
return err
}
}
return nil
2021-07-01 08:19:28 +00:00
}, signatures); err != nil {
return fmt.Errorf("fail to move %s => %s: %v", oldParent.Child(entry.Name()), newParent.Child(newName), err)
}
return nil
}
2021-10-17 11:22:42 +00:00
func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32) error {
currentDirPath := oldParent.Child(entry.Name())
newDirPath := newParent.Child(newName)
glog.V(1).Infof("moving folder %s => %s", currentDirPath, newDirPath)
lastFileName := ""
includeLastFile := false
for {
entries, hasMore, err := fs.filer.ListDirectoryEntries(ctx, currentDirPath, lastFileName, includeLastFile, 1024, "", "", "")
if err != nil {
return err
}
2019-09-03 16:22:56 +00:00
// println("found", len(entries), "entries under", currentDirPath)
for _, item := range entries {
lastFileName = item.Name()
2019-09-03 16:22:56 +00:00
// println("processing", lastFileName)
2021-10-17 11:22:42 +00:00
err := fs.moveEntry(ctx, stream, currentDirPath, item, newDirPath, item.Name(), signatures)
if err != nil {
return err
}
}
if !hasMore {
break
}
}
return nil
}
2021-10-17 11:22:42 +00:00
func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error, signatures []int32) error {
oldPath, newPath := oldParent.Child(entry.Name()), newParent.Child(newName)
glog.V(1).Infof("moving entry %s => %s", oldPath, newPath)
if oldPath == newPath {
glog.V(1).Infof("skip moving entry %s => %s", oldPath, newPath)
return nil
}
// add to new directory
2020-09-01 07:21:19 +00:00
newEntry := &filer.Entry{
2022-02-25 10:53:37 +00:00
FullPath: newPath,
Attr: entry.Attr,
Chunks: entry.GetChunks(),
2022-02-25 10:53:37 +00:00
Extended: entry.Extended,
Content: entry.Content,
HardLinkCounter: entry.HardLinkCounter,
HardLinkId: entry.HardLinkId,
Remote: entry.Remote,
Quota: entry.Quota,
}
if createErr := fs.filer.CreateEntry(ctx, newEntry, false, false, signatures, false, fs.filer.MaxFilenameLength); createErr != nil {
return createErr
}
2021-10-17 11:22:42 +00:00
if stream != nil {
if err := stream.Send(&filer_pb.StreamRenameEntryResponse{
Directory: string(oldParent),
2021-10-17 11:22:42 +00:00
EventNotification: &filer_pb.EventNotification{
2021-11-03 06:39:16 +00:00
OldEntry: &filer_pb.Entry{
2021-10-17 11:22:42 +00:00
Name: entry.Name(),
},
NewEntry: newEntry.ToProtoEntry(),
DeleteChunks: false,
NewParentPath: string(newParent),
IsFromOtherCluster: false,
Signatures: nil,
},
TsNs: time.Now().UnixNano(),
}); err != nil {
return err
}
}
2020-04-13 04:00:55 +00:00
if moveFolderSubEntries != nil {
if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil {
return moveChildrenErr
}
}
// delete old entry
ctx = context.WithValue(ctx, "OP", "MV")
2021-07-01 08:19:28 +00:00
deleteErr := fs.filer.DeleteEntryMetaAndData(ctx, oldPath, false, false, false, false, signatures)
if deleteErr != nil {
return deleteErr
}
2021-10-17 11:22:42 +00:00
if stream != nil {
if err := stream.Send(&filer_pb.StreamRenameEntryResponse{
Directory: string(oldParent),
EventNotification: &filer_pb.EventNotification{
OldEntry: &filer_pb.Entry{
Name: entry.Name(),
},
NewEntry: nil,
DeleteChunks: false,
NewParentPath: "",
IsFromOtherCluster: false,
Signatures: nil,
},
TsNs: time.Now().UnixNano(),
}); err != nil {
return err
}
}
return nil
}