diff --git a/weed/shell/command_remote_copy_local.go b/weed/shell/command_remote_copy_local.go index 4d5495856..00a20d387 100644 --- a/weed/shell/command_remote_copy_local.go +++ b/weed/shell/command_remote_copy_local.go @@ -5,6 +5,7 @@ import ( "flag" "fmt" "io" + "sort" "strings" "sync" "sync/atomic" @@ -43,16 +44,25 @@ func (c *commandRemoteCopyLocal) Help() string { remote.copy.local -dir=/xxx -exclude=*.tmp # exclude temporary files remote.copy.local -dir=/xxx -dryRun=true # show what would be done without making changes remote.copy.local -dir=/xxx -forceUpdate=true # force update even if remote exists + remote.copy.local -dir=/xxx -delete # also delete remote files that do not exist locally This command will: 1. Find local files that don't exist on remote storage 2. Copy these files to remote storage 3. Update local metadata with remote information + 4. With -delete, remove remote files that do not exist locally (similar to rsync --delete) This is useful when: - You deleted filer logs and need to copy existing files - You have local files that were never copied to remote - You want to ensure all local files are backed up to remote + - You want scheduled one-shot backups that also propagate local deletions (with -delete) + + Notes on -delete: + - only remote files under -dir are considered; paths outside it are never deleted + - -include/-exclude patterns also limit which remote files are deleted + - size/age filters only apply to copying, not to deletion + - use -dryRun=true first to review what would be deleted ` } @@ -69,6 +79,7 @@ func (c *commandRemoteCopyLocal) Do(args []string, commandEnv *CommandEnv, write concurrency := remoteCopyLocalCommand.Int("concurrent", 16, "concurrent file operations") dryRun := remoteCopyLocalCommand.Bool("dryRun", false, "show what would be done without making changes") forceUpdate := remoteCopyLocalCommand.Bool("forceUpdate", false, "force update even if remote exists") + deleteExtraneous := remoteCopyLocalCommand.Bool("delete", false, "delete extraneous files from remote storage (files that do not exist locally), similar to rsync --delete") fileFilter := newFileFilter(remoteCopyLocalCommand) if err = remoteCopyLocalCommand.Parse(args); err != nil { @@ -86,10 +97,10 @@ func (c *commandRemoteCopyLocal) Do(args []string, commandEnv *CommandEnv, write } // perform local to remote copy - return c.doLocalToRemoteCopy(commandEnv, writer, util.FullPath(localMountedDir), remoteStorageMountedLocation, util.FullPath(*dir), remoteStorageConf, *concurrency, *dryRun, *forceUpdate, fileFilter) + return c.doLocalToRemoteCopy(commandEnv, writer, util.FullPath(localMountedDir), remoteStorageMountedLocation, util.FullPath(*dir), remoteStorageConf, *concurrency, *dryRun, *forceUpdate, *deleteExtraneous, fileFilter) } -func (c *commandRemoteCopyLocal) doLocalToRemoteCopy(commandEnv *CommandEnv, writer io.Writer, localMountedDir util.FullPath, remoteMountedLocation *remote_pb.RemoteStorageLocation, dirToCopy util.FullPath, remoteConf *remote_pb.RemoteConf, concurrency int, dryRun bool, forceUpdate bool, fileFilter *FileFilter) error { +func (c *commandRemoteCopyLocal) doLocalToRemoteCopy(commandEnv *CommandEnv, writer io.Writer, localMountedDir util.FullPath, remoteMountedLocation *remote_pb.RemoteStorageLocation, dirToCopy util.FullPath, remoteConf *remote_pb.RemoteConf, concurrency int, dryRun bool, forceUpdate bool, deleteExtraneous bool, fileFilter *FileFilter) error { // Get remote storage client remoteStorage, err := remote_storage.GetRemoteStorage(remoteConf) @@ -116,11 +127,11 @@ func (c *commandRemoteCopyLocal) doLocalToRemoteCopy(commandEnv *CommandEnv, wri fmt.Fprintf(writer, "Found %d files/directories in local storage\n", len(localFiles)) // Step 2: Check which files exist on remote storage - remoteFiles := make(map[string]bool) + remoteFiles := make(map[string]bool) // full path -> isDirectory err = remoteStorage.Traverse(remote, func(remoteDir, name string, isDirectory bool, remoteEntry *filer_pb.RemoteEntry) error { localDir := filer.MapRemoteStorageLocationPathToFullPath(localMountedDir, remoteMountedLocation, remoteDir) fullPath := string(localDir.Child(name)) - remoteFiles[fullPath] = true + remoteFiles[fullPath] = isDirectory return nil }) if err != nil { @@ -129,46 +140,26 @@ func (c *commandRemoteCopyLocal) doLocalToRemoteCopy(commandEnv *CommandEnv, wri fmt.Fprintf(writer, "Found %d files/directories in remote storage\n", len(remoteFiles)) - // Step 3: Determine files to copy - var filesToCopy []string - for localPath, localEntry := range localFiles { - // Skip directories - if localEntry.IsDirectory { - continue - } + // Step 3: Determine files to copy and files to delete + plan := planLocalToRemoteSync(localFiles, remoteFiles, dirToCopy, forceUpdate, deleteExtraneous, fileFilter) - // Apply file filter - if !fileFilter.matches(localEntry) { - continue - } - - // Check if file needs copying - needsCopy := false - if !remoteFiles[localPath] { - // File doesn't exist on remote - needsCopy = true - } else if forceUpdate { - // Force update requested and file exists on remote - needsCopy = true - } - - if needsCopy { - filesToCopy = append(filesToCopy, localPath) - } + fmt.Fprintf(writer, "Files to copy: %d\n", len(plan.filesToCopy)) + if deleteExtraneous { + fmt.Fprintf(writer, "Files to delete from remote: %d\n", len(plan.filesToDelete)) } - fmt.Fprintf(writer, "Files to copy: %d\n", len(filesToCopy)) - if dryRun { fmt.Fprintf(writer, "DRY RUN - showing what would be done:\n") - for _, path := range filesToCopy { + for _, path := range plan.filesToCopy { fmt.Fprintf(writer, "COPY: %s\n", path) } + for _, path := range plan.filesToDelete { + fmt.Fprintf(writer, "DELETE: %s\n", path) + } return nil } - // Step 4: Copy files to remote storage - if len(filesToCopy) == 0 { + if len(plan.filesToCopy) == 0 && len(plan.filesToDelete) == 0 { fmt.Fprintf(writer, "No files to copy\n") return nil } @@ -180,7 +171,8 @@ func (c *commandRemoteCopyLocal) doLocalToRemoteCopy(commandEnv *CommandEnv, wri var successCount atomic.Int64 var outputMu sync.Mutex - for _, pathToCopy := range filesToCopy { + // Step 4: Copy files to remote storage + for _, pathToCopy := range plan.filesToCopy { wg.Add(1) localPath := pathToCopy // Capture for closure limitedConcurrentExecutor.Execute(func() { @@ -222,13 +214,113 @@ func (c *commandRemoteCopyLocal) doLocalToRemoteCopy(commandEnv *CommandEnv, wri wg.Wait() if firstErr != nil { + // skip deletion when any copy failed, to stay on the safe side return firstErr } - fmt.Fprintf(writer, "Successfully copied %d files to remote storage\n", successCount.Load()) + if len(plan.filesToCopy) > 0 { + fmt.Fprintf(writer, "Successfully copied %d files to remote storage\n", successCount.Load()) + } + + // Step 5: Delete extraneous files from remote storage + if len(plan.filesToDelete) == 0 { + return nil + } + + var deleteErr error + var deleteErrOnce sync.Once + var deletedCount atomic.Int64 + + for _, pathToDelete := range plan.filesToDelete { + wg.Add(1) + localPath := pathToDelete // Capture for closure + limitedConcurrentExecutor.Execute(func() { + defer wg.Done() + + remoteLocation := filer.MapFullPathToRemoteStorageLocation(localMountedDir, remoteMountedLocation, util.FullPath(localPath)) + if err := remoteStorage.DeleteFile(remoteLocation); err != nil { + outputMu.Lock() + fmt.Fprintf(writer, "Deleting %s... failed: %v\n", localPath, err) + outputMu.Unlock() + deleteErrOnce.Do(func() { + deleteErr = err + }) + return + } + + deletedCount.Add(1) + outputMu.Lock() + fmt.Fprintf(writer, "Deleting %s... done\n", localPath) + outputMu.Unlock() + }) + } + wg.Wait() + + if deleteErr != nil { + return deleteErr + } + + fmt.Fprintf(writer, "Successfully deleted %d files from remote storage\n", deletedCount.Load()) return nil } +type localToRemoteSyncPlan struct { + filesToCopy []string + filesToDelete []string +} + +func planLocalToRemoteSync(localFiles map[string]*filer_pb.Entry, remoteFiles map[string]bool, dirToCopy util.FullPath, forceUpdate bool, deleteExtraneous bool, fileFilter *FileFilter) *localToRemoteSyncPlan { + plan := &localToRemoteSyncPlan{} + + for localPath, localEntry := range localFiles { + // Skip directories + if localEntry.IsDirectory { + continue + } + + // Apply file filter + if !fileFilter.matches(localEntry) { + continue + } + + // Copy if the file doesn't exist on remote, or if force update is requested + if _, foundOnRemote := remoteFiles[localPath]; !foundOnRemote || forceUpdate { + plan.filesToCopy = append(plan.filesToCopy, localPath) + } + } + sort.Strings(plan.filesToCopy) + + if !deleteExtraneous { + return plan + } + + for remotePath, isDirectory := range remoteFiles { + // Traverse lists by key prefix (no delimiter), so it can return entries + // outside dirToCopy that merely share its name prefix (e.g. a sibling + // "foobar" when dirToCopy maps to prefix "foo"). Never delete anything + // outside the requested subtree. + if !util.FullPath(remotePath).IsUnder(dirToCopy) { + continue + } + if _, foundLocally := localFiles[remotePath]; foundLocally { + continue + } + // object stores expose no real directories; only files are deleted + if isDirectory { + continue + } + // name filters also protect remote files from deletion; + // size/age filters need local attributes, so they only apply to copying + if !fileFilter.matchesName(util.FullPath(remotePath).Name()) { + continue + } + plan.filesToDelete = append(plan.filesToDelete, remotePath) + } + sort.Strings(plan.filesToDelete) + + return plan +} + func syncFileToRemote(commandEnv *CommandEnv, remoteStorage remote_storage.RemoteStorageClient, remoteConf *remote_pb.RemoteConf, remoteLocation *remote_pb.RemoteStorageLocation, dir util.FullPath, localEntry *filer_pb.Entry) error { // Upload to remote storage using the same approach as filer_remote_sync diff --git a/weed/shell/command_remote_copy_local_test.go b/weed/shell/command_remote_copy_local_test.go new file mode 100644 index 000000000..aec9637d1 --- /dev/null +++ b/weed/shell/command_remote_copy_local_test.go @@ -0,0 +1,267 @@ +package shell + +import ( + "reflect" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +func newTestFileFilter(include, exclude string) *FileFilter { + defaultInt64 := int64(-1) + return &FileFilter{ + include: &include, + exclude: &exclude, + minSize: &defaultInt64, + maxSize: &defaultInt64, + minAge: &defaultInt64, + maxAge: &defaultInt64, + minCacheAge: &defaultInt64, + now: time.Now().Unix(), + } +} + +func testFileEntry(name string) *filer_pb.Entry { + return &filer_pb.Entry{ + Name: name, + Attributes: &filer_pb.FuseAttributes{}, + } +} + +func testDirEntry(name string) *filer_pb.Entry { + return &filer_pb.Entry{ + Name: name, + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{}, + } +} + +func TestPlanLocalToRemoteSync(t *testing.T) { + tests := []struct { + name string + dirToCopy string + localFiles map[string]*filer_pb.Entry + remoteFiles map[string]bool // path -> isDirectory + forceUpdate bool + deleteExtraneous bool + fileFilter *FileFilter + wantFilesToCopy []string + wantFilesToDelete []string + }{ + { + name: "copy local-only files, no delete flag", + dirToCopy: "/mnt", + localFiles: map[string]*filer_pb.Entry{ + "/mnt/a.txt": testFileEntry("a.txt"), + "/mnt/b.txt": testFileEntry("b.txt"), + }, + remoteFiles: map[string]bool{ + "/mnt/b.txt": false, + "/mnt/orphan.txt": false, + }, + fileFilter: newTestFileFilter("", ""), + wantFilesToCopy: []string{"/mnt/a.txt"}, + }, + { + name: "delete flag removes remote-only files", + dirToCopy: "/mnt", + localFiles: map[string]*filer_pb.Entry{ + "/mnt/a.txt": testFileEntry("a.txt"), + }, + remoteFiles: map[string]bool{ + "/mnt/a.txt": false, + "/mnt/orphan.txt": false, + "/mnt/orphan2.txt": false, + }, + deleteExtraneous: true, + fileFilter: newTestFileFilter("", ""), + wantFilesToDelete: []string{"/mnt/orphan.txt", "/mnt/orphan2.txt"}, + }, + { + name: "delete with empty local storage removes all remote files", + dirToCopy: "/mnt", + localFiles: map[string]*filer_pb.Entry{}, + remoteFiles: map[string]bool{ + "/mnt/orphan.txt": false, + }, + deleteExtraneous: true, + fileFilter: newTestFileFilter("", ""), + wantFilesToDelete: []string{"/mnt/orphan.txt"}, + }, + { + name: "delete respects include pattern", + dirToCopy: "/mnt", + localFiles: map[string]*filer_pb.Entry{ + "/mnt/a.pdf": testFileEntry("a.pdf"), + }, + remoteFiles: map[string]bool{ + "/mnt/a.pdf": false, + "/mnt/orphan.pdf": false, + "/mnt/keep.txt": false, + }, + deleteExtraneous: true, + fileFilter: newTestFileFilter("*.pdf", ""), + wantFilesToDelete: []string{"/mnt/orphan.pdf"}, + }, + { + name: "delete respects exclude pattern", + dirToCopy: "/mnt", + localFiles: map[string]*filer_pb.Entry{ + "/mnt/a.txt": testFileEntry("a.txt"), + }, + remoteFiles: map[string]bool{ + "/mnt/a.txt": false, + "/mnt/orphan.txt": false, + "/mnt/keep.bak": false, + }, + deleteExtraneous: true, + fileFilter: newTestFileFilter("", "*.bak"), + wantFilesToDelete: []string{"/mnt/orphan.txt"}, + }, + { + name: "delete ignores size/age filters", + dirToCopy: "/mnt", + localFiles: map[string]*filer_pb.Entry{ + "/mnt/a.txt": testFileEntry("a.txt"), + }, + remoteFiles: map[string]bool{ + "/mnt/a.txt": false, + "/mnt/orphan.txt": false, + }, + deleteExtraneous: true, + fileFilter: func() *FileFilter { + ff := newTestFileFilter("", "") + minSize := int64(1000000) // would exclude orphan.txt from copying if it applied to deletion + ff.minSize = &minSize + return ff + }(), + wantFilesToDelete: []string{"/mnt/orphan.txt"}, + }, + { + name: "delete removes orphaned files, directory entries left alone", + dirToCopy: "/mnt", + localFiles: map[string]*filer_pb.Entry{ + "/mnt/a.txt": testFileEntry("a.txt"), + }, + remoteFiles: map[string]bool{ + "/mnt/a.txt": false, + "/mnt/old": true, + "/mnt/old/sub": true, + "/mnt/old/sub/orphan.txt": false, + "/mnt/old/sub2": true, + "/mnt/old/sub2/orphan.txt": false, + }, + deleteExtraneous: true, + fileFilter: newTestFileFilter("", ""), + wantFilesToDelete: []string{ + "/mnt/old/sub/orphan.txt", + "/mnt/old/sub2/orphan.txt", + }, + }, + { + name: "exclude pattern protects remote files under nested directory", + dirToCopy: "/mnt", + localFiles: map[string]*filer_pb.Entry{ + "/mnt/a.txt": testFileEntry("a.txt"), + }, + remoteFiles: map[string]bool{ + "/mnt/a.txt": false, + "/mnt/old": true, + "/mnt/old/orphan.txt": false, + "/mnt/old/keep.bak": false, + }, + deleteExtraneous: true, + fileFilter: newTestFileFilter("", "*.bak"), + wantFilesToDelete: []string{"/mnt/old/orphan.txt"}, + }, + { + name: "remote files outside -dir are never deleted", + dirToCopy: "/mnt/foo", + localFiles: map[string]*filer_pb.Entry{ + "/mnt/foo/a.txt": testFileEntry("a.txt"), + }, + remoteFiles: map[string]bool{ + "/mnt/foo/a.txt": false, + "/mnt/foo/orphan.txt": false, + "/mnt/foobar/b.txt": false, // sibling sharing the "foo" key prefix, must be left alone + }, + deleteExtraneous: true, + fileFilter: newTestFileFilter("", ""), + wantFilesToDelete: []string{"/mnt/foo/orphan.txt"}, + }, + { + name: "directory present locally is not deleted", + dirToCopy: "/mnt", + localFiles: map[string]*filer_pb.Entry{ + "/mnt/dir": testDirEntry("dir"), + }, + remoteFiles: map[string]bool{ + "/mnt/dir": true, + }, + deleteExtraneous: true, + fileFilter: newTestFileFilter("", ""), + }, + { + name: "forceUpdate copies files that exist on remote", + dirToCopy: "/mnt", + localFiles: map[string]*filer_pb.Entry{ + "/mnt/a.txt": testFileEntry("a.txt"), + }, + remoteFiles: map[string]bool{ + "/mnt/a.txt": false, + }, + forceUpdate: true, + fileFilter: newTestFileFilter("", ""), + wantFilesToCopy: []string{"/mnt/a.txt"}, + }, + { + name: "local directories are not copied", + dirToCopy: "/mnt", + localFiles: map[string]*filer_pb.Entry{ + "/mnt/dir": testDirEntry("dir"), + "/mnt/a.txt": testFileEntry("a.txt"), + }, + remoteFiles: map[string]bool{}, + fileFilter: newTestFileFilter("", ""), + wantFilesToCopy: []string{"/mnt/a.txt"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + plan := planLocalToRemoteSync(tt.localFiles, tt.remoteFiles, util.FullPath(tt.dirToCopy), tt.forceUpdate, tt.deleteExtraneous, tt.fileFilter) + if !reflect.DeepEqual(plan.filesToCopy, tt.wantFilesToCopy) { + t.Errorf("filesToCopy = %v, want %v", plan.filesToCopy, tt.wantFilesToCopy) + } + if !reflect.DeepEqual(plan.filesToDelete, tt.wantFilesToDelete) { + t.Errorf("filesToDelete = %v, want %v", plan.filesToDelete, tt.wantFilesToDelete) + } + }) + } +} + +func TestFileFilter_matchesName(t *testing.T) { + tests := []struct { + name string + include string + exclude string + fileName string + want bool + }{ + {"no filters", "", "", "a.txt", true}, + {"include match", "*.pdf", "", "a.pdf", true}, + {"include mismatch", "*.pdf", "", "a.txt", false}, + {"exclude match", "", "*.tmp", "a.tmp", false}, + {"exclude mismatch", "", "*.tmp", "a.txt", true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ff := newTestFileFilter(tt.include, tt.exclude) + if got := ff.matchesName(tt.fileName); got != tt.want { + t.Errorf("matchesName(%q) = %v, want %v", tt.fileName, got, tt.want) + } + }) + } +} diff --git a/weed/shell/command_remote_uncache.go b/weed/shell/command_remote_uncache.go index 48fd24c03..bd47b4c33 100644 --- a/weed/shell/command_remote_uncache.go +++ b/weed/shell/command_remote_uncache.go @@ -152,20 +152,29 @@ func newFileFilter(remoteMountCommand *flag.FlagSet) (ff *FileFilter) { return } -func (ff *FileFilter) matches(entry *filer_pb.Entry) bool { - if entry.Attributes == nil { - return false - } +// matchesName applies only the name-based include/exclude patterns, +// usable for remote entries where local attributes are not available. +func (ff *FileFilter) matchesName(name string) bool { if *ff.include != "" { - if ok, _ := filepath.Match(*ff.include, entry.Name); !ok { + if ok, _ := filepath.Match(*ff.include, name); !ok { return false } } if *ff.exclude != "" { - if ok, _ := filepath.Match(*ff.exclude, entry.Name); ok { + if ok, _ := filepath.Match(*ff.exclude, name); ok { return false } } + return true +} + +func (ff *FileFilter) matches(entry *filer_pb.Entry) bool { + if entry.Attributes == nil { + return false + } + if !ff.matchesName(entry.Name) { + return false + } if *ff.minSize != -1 { if int64(entry.Attributes.FileSize) < *ff.minSize { return false