diff --git a/cli/command_restore.go b/cli/command_restore.go index c8c5209b1..629a09412 100644 --- a/cli/command_restore.go +++ b/cli/command_restore.go @@ -77,6 +77,8 @@ var ( restoreSkipTimes = false restoreSkipOwners = false restoreSkipPermissions = false + restoreIncremental = false + restoreIgnoreErrors = false ) const ( @@ -101,6 +103,8 @@ func addRestoreFlags(cmd *kingpin.CmdClause) { cmd.Flag("skip-permissions", "Skip permissions during restore").BoolVar(&restoreSkipPermissions) cmd.Flag("skip-times", "Skip times during restore").BoolVar(&restoreSkipTimes) cmd.Flag("ignore-permission-errors", "Ignore permission errors").BoolVar(&restoreIgnorePermissionErrors) + cmd.Flag("ignore-errors", "Ignore all errors").BoolVar(&restoreIgnoreErrors) + cmd.Flag("skip-existing", "Skip files and symlinks that exist in the output").BoolVar(&restoreIncremental) } func restoreOutput(ctx context.Context) (restore.Output, error) { @@ -182,7 +186,22 @@ func detectRestoreMode(ctx context.Context, m string) string { } func printRestoreStats(ctx context.Context, st restore.Stats) { - log(ctx).Infof("Restored %v files, %v directories and %v symbolic links (%v)\n", st.RestoredFileCount, st.RestoredDirCount, st.RestoredSymlinkCount, units.BytesStringBase10(st.RestoredTotalFileSize)) + var maybeSkipped, maybeErrors string + + if st.SkippedCount > 0 { + maybeSkipped = fmt.Sprintf(", skipped %v (%v)", st.SkippedCount, units.BytesStringBase10(st.SkippedTotalFileSize)) + } + + if st.IgnoredErrorCount > 0 { + maybeErrors = fmt.Sprintf(", ignored %v errors", st.IgnoredErrorCount) + } + + log(ctx).Infof("Restored %v files, %v directories and %v symbolic links (%v)%v%v.\n", + st.RestoredFileCount, + st.RestoredDirCount, + st.RestoredSymlinkCount, + units.BytesStringBase10(st.RestoredTotalFileSize), + maybeSkipped, maybeErrors) } func runRestoreCommand(ctx context.Context, rep repo.Repository) error { @@ -199,16 +218,18 @@ func runRestoreCommand(ctx context.Context, rep repo.Repository) error { t0 := clock.Now() st, err := restore.Entry(ctx, rep, output, rootEntry, restore.Options{ - Parallel: restoreParallel, + Parallel: restoreParallel, + Incremental: restoreIncremental, + IgnoreErrors: restoreIgnoreErrors, ProgressCallback: func(ctx context.Context, stats restore.Stats) { - restoredCount := stats.RestoredFileCount + stats.RestoredDirCount + stats.RestoredSymlinkCount + restoredCount := stats.RestoredFileCount + stats.RestoredDirCount + stats.RestoredSymlinkCount + stats.SkippedCount enqueuedCount := stats.EnqueuedFileCount + stats.EnqueuedDirCount + stats.EnqueuedSymlinkCount if restoredCount == 0 { return } - var maybeRemaining string + var maybeRemaining, maybeSkipped, maybeErrors string if stats.EnqueuedTotalFileSize > 0 { progress := float64(stats.RestoredTotalFileSize) / float64(stats.EnqueuedTotalFileSize) @@ -223,9 +244,19 @@ func runRestoreCommand(ctx context.Context, rep repo.Repository) error { } } - log(ctx).Infof("Processed %v (%v) of %v (%v)%v.", + if stats.SkippedCount > 0 { + maybeSkipped = fmt.Sprintf(", skipped %v (%v)", stats.SkippedCount, units.BytesStringBase10(stats.SkippedTotalFileSize)) + } + + if stats.IgnoredErrorCount > 0 { + maybeErrors = fmt.Sprintf(", ignored %v errors", stats.IgnoredErrorCount) + } + + log(ctx).Infof("Processed %v (%v) of %v (%v)%v%v.", restoredCount, units.BytesStringBase10(stats.RestoredTotalFileSize), enqueuedCount, units.BytesStringBase10(stats.EnqueuedTotalFileSize), + maybeSkipped, + maybeErrors, maybeRemaining) }, }) diff --git a/snapshot/restore/local_fs_output.go b/snapshot/restore/local_fs_output.go index 0e3020b41..cb32b1de9 100644 --- a/snapshot/restore/local_fs_output.go +++ b/snapshot/restore/local_fs_output.go @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "runtime" + "time" "github.com/natefinch/atomic" "github.com/pkg/errors" @@ -17,6 +18,8 @@ import ( const modBits = os.ModePerm | os.ModeSetgid | os.ModeSetuid | os.ModeSticky +const maxTimeDeltaToConsiderFileTheSame = 2 * time.Second + // FilesystemOutput contains the options for outputting a file system tree. type FilesystemOutput struct { // TargetPath for restore. @@ -95,6 +98,31 @@ func (o *FilesystemOutput) WriteFile(ctx context.Context, relativePath string, f return nil } +// FileExists implements restore.Output interface. +func (o *FilesystemOutput) FileExists(ctx context.Context, relativePath string, e fs.File) bool { + st, err := os.Lstat(filepath.Join(o.TargetPath, relativePath)) + if err != nil { + return false + } + + if (st.Mode() & os.ModeType) != 0 { + // not a file + return false + } + + if st.Size() != e.Size() { + // wrong size + return false + } + + timeDelta := st.ModTime().Sub(e.ModTime()) + if timeDelta < 0 { + timeDelta = -timeDelta + } + + return timeDelta < maxTimeDeltaToConsiderFileTheSame +} + // CreateSymlink implements restore.Output interface. func (o *FilesystemOutput) CreateSymlink(ctx context.Context, relativePath string, e fs.Symlink) error { targetPath, err := e.Readlink(ctx) @@ -139,6 +167,16 @@ func fileIsSymlink(stat os.FileInfo) bool { return stat.Mode()&os.ModeSymlink != 0 } +// SymlinkExists implements restore.Output interface. +func (o *FilesystemOutput) SymlinkExists(ctx context.Context, relativePath string, e fs.Symlink) bool { + st, err := os.Lstat(filepath.Join(o.TargetPath, relativePath)) + if err != nil { + return false + } + + return (st.Mode() & os.ModeType) == os.ModeSymlink +} + // set permission, modification time and user/group ids on targetPath. func (o *FilesystemOutput) setAttributes(targetPath string, e fs.Entry) error { le, err := localfs.NewEntry(targetPath) diff --git a/snapshot/restore/restore.go b/snapshot/restore/restore.go index 66f06f359..cb61e270d 100644 --- a/snapshot/restore/restore.go +++ b/snapshot/restore/restore.go @@ -22,7 +22,9 @@ type Output interface { BeginDirectory(ctx context.Context, relativePath string, e fs.Directory) error FinishDirectory(ctx context.Context, relativePath string, e fs.Directory) error WriteFile(ctx context.Context, relativePath string, e fs.File) error + FileExists(ctx context.Context, relativePath string, e fs.File) bool CreateSymlink(ctx context.Context, relativePath string, e fs.Symlink) error + SymlinkExists(ctx context.Context, relativePath string, e fs.Symlink) bool Close(ctx context.Context) error } @@ -30,6 +32,7 @@ type Output interface { type Stats struct { RestoredTotalFileSize int64 EnqueuedTotalFileSize int64 + SkippedTotalFileSize int64 RestoredFileCount int32 RestoredDirCount int32 @@ -37,18 +40,24 @@ type Stats struct { EnqueuedFileCount int32 EnqueuedDirCount int32 EnqueuedSymlinkCount int32 + SkippedCount int32 + IgnoredErrorCount int32 } func (s *Stats) clone() Stats { return Stats{ RestoredTotalFileSize: atomic.LoadInt64(&s.RestoredTotalFileSize), EnqueuedTotalFileSize: atomic.LoadInt64(&s.EnqueuedTotalFileSize), - RestoredFileCount: atomic.LoadInt32(&s.RestoredFileCount), - RestoredDirCount: atomic.LoadInt32(&s.RestoredDirCount), - RestoredSymlinkCount: atomic.LoadInt32(&s.RestoredSymlinkCount), - EnqueuedFileCount: atomic.LoadInt32(&s.EnqueuedFileCount), - EnqueuedDirCount: atomic.LoadInt32(&s.EnqueuedDirCount), - EnqueuedSymlinkCount: atomic.LoadInt32(&s.EnqueuedSymlinkCount), + SkippedTotalFileSize: atomic.LoadInt64(&s.SkippedTotalFileSize), + + RestoredFileCount: atomic.LoadInt32(&s.RestoredFileCount), + RestoredDirCount: atomic.LoadInt32(&s.RestoredDirCount), + RestoredSymlinkCount: atomic.LoadInt32(&s.RestoredSymlinkCount), + EnqueuedFileCount: atomic.LoadInt32(&s.EnqueuedFileCount), + EnqueuedDirCount: atomic.LoadInt32(&s.EnqueuedDirCount), + EnqueuedSymlinkCount: atomic.LoadInt32(&s.EnqueuedSymlinkCount), + SkippedCount: atomic.LoadInt32(&s.SkippedCount), + IgnoredErrorCount: atomic.LoadInt32(&s.IgnoredErrorCount), } } @@ -56,11 +65,18 @@ func (s *Stats) clone() Stats { type Options struct { Parallel int ProgressCallback func(ctx context.Context, s Stats) + Incremental bool + IgnoreErrors bool } // Entry walks a snapshot root with given root entry and restores it to the provided output. func Entry(ctx context.Context, rep repo.Repository, output Output, rootEntry fs.Entry, options Options) (Stats, error) { - c := copier{output: output, q: parallelwork.NewQueue()} + c := copier{ + output: output, + q: parallelwork.NewQueue(), + incremental: options.Incremental, + ignoreErrors: options.IgnoreErrors, + } c.q.ProgressCallback = func(ctx context.Context, enqueued, active, completed int64) { options.ProgressCallback(ctx, c.stats.clone()) @@ -91,12 +107,52 @@ func Entry(ctx context.Context, rep repo.Repository, output Output, rootEntry fs } type copier struct { - stats Stats - output Output - q *parallelwork.Queue + stats Stats + output Output + q *parallelwork.Queue + incremental bool + ignoreErrors bool } func (c *copier) copyEntry(ctx context.Context, e fs.Entry, targetPath string, onCompletion func() error) error { + if c.incremental { + // in incremental mode, do not copy if the output already exists + switch e := e.(type) { + case fs.File: + if c.output.FileExists(ctx, targetPath, e) { + log(ctx).Debugf("skipping file %v because it already exists and metadata matches", targetPath) + atomic.AddInt32(&c.stats.SkippedCount, 1) + atomic.AddInt64(&c.stats.SkippedTotalFileSize, e.Size()) + + return onCompletion() + } + + case fs.Symlink: + if c.output.SymlinkExists(ctx, targetPath, e) { + atomic.AddInt32(&c.stats.SkippedCount, 1) + log(ctx).Debugf("skipping symlink %v because it already exists", targetPath) + + return onCompletion() + } + } + } + + err := c.copyEntryInternal(ctx, e, targetPath, onCompletion) + if err == nil { + return nil + } + + if c.ignoreErrors { + atomic.AddInt32(&c.stats.IgnoredErrorCount, 1) + log(ctx).Warningf("ignored error %v on %v", err, targetPath) + + return nil + } + + return err +} + +func (c *copier) copyEntryInternal(ctx context.Context, e fs.Entry, targetPath string, onCompletion func() error) error { switch e := e.(type) { case fs.Directory: log(ctx).Debugf("dir: '%v'", targetPath) diff --git a/snapshot/restore/tar_output.go b/snapshot/restore/tar_output.go index db6fb68a9..03d21dafb 100644 --- a/snapshot/restore/tar_output.go +++ b/snapshot/restore/tar_output.go @@ -86,6 +86,11 @@ func (o *TarOutput) WriteFile(ctx context.Context, relativePath string, f fs.Fil return nil } +// FileExists implements restore.Output interface. +func (o *TarOutput) FileExists(ctx context.Context, relativePath string, f fs.File) bool { + return false +} + // CreateSymlink implements restore.Output interface. func (o *TarOutput) CreateSymlink(ctx context.Context, relativePath string, l fs.Symlink) error { target, err := l.Readlink(ctx) @@ -110,6 +115,11 @@ func (o *TarOutput) CreateSymlink(ctx context.Context, relativePath string, l fs return nil } +// SymlinkExists implements restore.Output interface. +func (o *TarOutput) SymlinkExists(ctx context.Context, relativePath string, l fs.Symlink) bool { + return false +} + // NewTarOutput creates new tar writer output. func NewTarOutput(w io.WriteCloser) *TarOutput { return &TarOutput{w, tar.NewWriter(w)} diff --git a/snapshot/restore/zip_output.go b/snapshot/restore/zip_output.go index 8f2daf0fa..caf0da49a 100644 --- a/snapshot/restore/zip_output.go +++ b/snapshot/restore/zip_output.go @@ -69,12 +69,22 @@ func (o *ZipOutput) WriteFile(ctx context.Context, relativePath string, f fs.Fil return nil } +// FileExists implements restore.Output interface. +func (o *ZipOutput) FileExists(ctx context.Context, relativePath string, l fs.File) bool { + return false +} + // CreateSymlink implements restore.Output interface. func (o *ZipOutput) CreateSymlink(ctx context.Context, relativePath string, e fs.Symlink) error { log(ctx).Debugf("create symlink not implemented yet") return nil } +// SymlinkExists implements restore.Output interface. +func (o *ZipOutput) SymlinkExists(ctx context.Context, relativePath string, l fs.Symlink) bool { + return false +} + // NewZipOutput creates new zip writer output. func NewZipOutput(w io.WriteCloser, method uint16) *ZipOutput { return &ZipOutput{w, zip.NewWriter(w), method} diff --git a/tests/end_to_end_test/restore_fail_test.go b/tests/end_to_end_test/restore_fail_test.go index 0f8739948..9d022195d 100644 --- a/tests/end_to_end_test/restore_fail_test.go +++ b/tests/end_to_end_test/restore_fail_test.go @@ -64,6 +64,9 @@ func TestRestoreFail(t *testing.T) { // Expect a subsequent restore to fail e.RunAndExpectFailure(t, "snapshot", "restore", snapID, targetDir) + + // --ignore-errors allows the snapshot to succeed despite missing blob. + e.RunAndExpectSuccess(t, "snapshot", "restore", "--ignore-errors", snapID, targetDir) } func findPackBlob(blobIDs []string) string { diff --git a/tests/end_to_end_test/restore_test.go b/tests/end_to_end_test/restore_test.go index f026be5c6..37d4defad 100644 --- a/tests/end_to_end_test/restore_test.go +++ b/tests/end_to_end_test/restore_test.go @@ -9,7 +9,9 @@ import ( "io" "os" "path/filepath" + "regexp" "runtime" + "strconv" "testing" "time" @@ -88,6 +90,33 @@ func TestRestoreCommand(t *testing.T) { // Attempt to restore into a target directory that already exists e.RunAndExpectFailure(t, "restore", rootID, restoreDir, "--no-overwrite-directories") + // Very quick incremental restore where all files already exist. + // Look for status output that indicates files were skipped. + re := regexp.MustCompile(`Restored (\d+) files.*skipped (\d+) `) + foundStatus := false + lastFileCount := 0 + _, stderr := e.RunAndExpectSuccessWithErrOut(t, "restore", rootID, restoreDir, "--skip-existing") + + for _, l := range stderr { + if m := re.FindStringSubmatch(l); m != nil { + fileCount, _ := strconv.Atoi(m[1]) + skippedCount, _ := strconv.Atoi(m[2]) + lastFileCount = fileCount + + if fileCount == 0 && skippedCount > 0 { + foundStatus = true + } + } + } + + if !foundStatus { + t.Fatalf("expected status line indicating files were skipped, none found: %v", stderr) + } + + if lastFileCount != 0 { + t.Fatalf("not all files were skipped: %v", stderr) + } + // Attempt to restore into a target directory that already exists e.RunAndExpectFailure(t, "restore", rootID, restoreDir, "--no-overwrite-files") }