restore: added support for incremental restore and ignoring copy errors (#794)

* restore: added support for incremental restore and ignoring copy errors

Co-authored-by: Julio López <julio+gh@kasten.io>

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)
 		},
 	})

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)

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)

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)}

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}

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 {

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")
 }