Support for repository sync (#522)

* blob: added DisplayName() method to blob.Storage

* cli: added 'kopia repo sync-to <provider>' which replicates BLOBs

Usage demo: https://asciinema.org/a/352299

Fixes #509

* implemented suggestion by Ciantic to fail sync if the destination repository is not compatible with the source

* cli: added 'kopia repo sync --must-exist'

This ensures that target repository is not empty, otherwise syncing to
an accidentally unmounted filesystem directory might copy everything
again.

cli/command_repository_sync.go

@@ -0,0 +1,310 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
+
+	"github.com/kopia/kopia/internal/gather"
+	"github.com/kopia/kopia/internal/stats"
+	"github.com/kopia/kopia/internal/units"
+	"github.com/kopia/kopia/repo"
+	"github.com/kopia/kopia/repo/blob"
+)
+
+var (
+	repositorySyncCommand              = repositoryCommands.Command("sync-to", "Synchronizes contents of this repository to another location")
+	repositorySyncUpdate               = repositorySyncCommand.Flag("update", "Whether to update blobs present in destination and source if the source is newer.").Default("true").Bool()
+	repositorySyncDelete               = repositorySyncCommand.Flag("delete", "Whether to delete blobs present in destination but not source.").Bool()
+	repositorySyncDryRun               = repositorySyncCommand.Flag("dry-run", "Do not perform copying.").Short('n').Bool()
+	repositorySyncParallelism          = repositorySyncCommand.Flag("parallel", "Copy parallelism.").Default("1").Int()
+	repositorySyncDestinationMustExist = repositorySyncCommand.Flag("must-exist", "Fail if destination does not have repository format blob.").Bool()
+)
+
+const syncProgressInterval = 300 * time.Millisecond
+
+func runSyncWithStorage(ctx context.Context, src, dst blob.Storage) error {
+	var ()
+
+	noticeColor.Fprintf(os.Stderr, "Synchronizing repositories:\n\n  Source:      %v\n  Destination: %v\n\n", src.DisplayName(), dst.DisplayName()) //nolint:errcheck
+
+	if !*repositorySyncDelete {
+		noticeColor.Fprintf(os.Stderr, "NOTE: By default no BLOBs are deleted, pass --delete to allow it.\n\n") //nolint:errcheck
+	}
+
+	if err := ensureRepositoriesHaveSameFormatBlob(ctx, src, dst); err != nil {
+		return err
+	}
+
+	printStderr("Looking for BLOBs to synchronize...\n")
+
+	var (
+		inSyncBlobs int
+		inSyncBytes int64
+
+		blobsToCopy    []blob.Metadata
+		totalCopyBytes int64
+
+		blobsToDelete    []blob.Metadata
+		totalDeleteBytes int64
+
+		srcBlobs     int
+		totalSrcSize int64
+	)
+
+	dstMetadata, err := listDestinationBlobs(ctx, dst)
+	if err != nil {
+		return err
+	}
+
+	beginSyncProgress()
+
+	if err := src.ListBlobs(ctx, "", func(srcmd blob.Metadata) error {
+		totalSrcSize += srcmd.Length
+
+		dstmd, exists := dstMetadata[srcmd.BlobID]
+		delete(dstMetadata, srcmd.BlobID)
+
+		switch {
+		case !exists:
+			blobsToCopy = append(blobsToCopy, srcmd)
+			totalCopyBytes += srcmd.Length
+		case srcmd.Timestamp.After(dstmd.Timestamp) && *repositorySyncUpdate:
+			blobsToCopy = append(blobsToCopy, srcmd)
+			totalCopyBytes += srcmd.Length
+		default:
+			inSyncBlobs++
+			inSyncBytes += srcmd.Length
+		}
+
+		srcBlobs++
+		outputSyncProgress(fmt.Sprintf("  Found %v BLOBs (%v) in the source repository, %v (%v) to copy", srcBlobs, units.BytesStringBase10(totalSrcSize), len(blobsToCopy), units.BytesStringBase10(totalCopyBytes)))
+
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	finishSyncProcess()
+
+	if *repositorySyncDelete {
+		for _, dstmd := range dstMetadata {
+			// found in dst, not in src since we were deleting from dst as we found a match.
+			blobsToDelete = append(blobsToDelete, dstmd)
+			totalDeleteBytes += dstmd.Length
+		}
+	}
+
+	printStderr(
+		"  Found %v BLOBs to delete (%v), %v in sync (%v)\n",
+		len(blobsToDelete), units.BytesStringBase10(totalDeleteBytes),
+		inSyncBlobs, units.BytesStringBase10(inSyncBytes),
+	)
+
+	if *repositorySyncDryRun {
+		return nil
+	}
+
+	printStderr("Copying...\n")
+
+	beginSyncProgress()
+
+	finalErr := runSyncBlobs(ctx, src, dst, blobsToCopy, blobsToDelete, totalCopyBytes)
+
+	finishSyncProcess()
+
+	return finalErr
+}
+
+func listDestinationBlobs(ctx context.Context, dst blob.Storage) (map[blob.ID]blob.Metadata, error) {
+	dstTotalBytes := int64(0)
+	dstMetadata := map[blob.ID]blob.Metadata{}
+
+	beginSyncProgress()
+
+	if err := dst.ListBlobs(ctx, "", func(bm blob.Metadata) error {
+		dstMetadata[bm.BlobID] = bm
+		dstTotalBytes += bm.Length
+		outputSyncProgress(fmt.Sprintf("  Found %v BLOBs in the destination repository (%v)", len(dstMetadata), units.BytesStringBase10(dstTotalBytes)))
+		return nil
+	}); err != nil {
+		return nil, errors.Wrap(err, "error listing BLOBs in destination repository")
+	}
+
+	finishSyncProcess()
+
+	return dstMetadata, nil
+}
+
+var (
+	lastSyncProgress   string
+	syncProgressMutex  sync.Mutex
+	nextSyncOutputTime time.Time
+)
+
+func beginSyncProgress() {
+	lastSyncProgress = ""
+	nextSyncOutputTime = time.Now()
+}
+
+func outputSyncProgress(s string) {
+	syncProgressMutex.Lock()
+	defer syncProgressMutex.Unlock()
+
+	if len(s) < len(lastSyncProgress) {
+		s += strings.Repeat(" ", len(lastSyncProgress)-len(s))
+	}
+
+	if time.Now().After(nextSyncOutputTime) {
+		printStderr("\r%v", s)
+
+		nextSyncOutputTime = time.Now().Add(syncProgressInterval)
+	}
+
+	lastSyncProgress = s
+}
+
+func finishSyncProcess() {
+	printStderr("\r%v\n", lastSyncProgress)
+}
+
+func runSyncBlobs(ctx context.Context, src, dst blob.Storage, blobsToCopy, blobsToDelete []blob.Metadata, totalBytes int64) error {
+	eg, ctx := errgroup.WithContext(ctx)
+	copyCh := sliceToChannel(ctx, blobsToCopy)
+	deleteCh := sliceToChannel(ctx, blobsToDelete)
+
+	var progressMutex sync.Mutex
+
+	var totalCopied stats.CountSum
+
+	startTime := time.Now()
+
+	for i := 0; i < *repositorySyncParallelism; i++ {
+		workerID := i
+
+		eg.Go(func() error {
+			for m := range copyCh {
+				log(ctx).Debugf("[%v] Copying %v (%v bytes)...\n", workerID, m.BlobID, m.Length)
+				if err := syncCopyBlob(ctx, m, src, dst); err != nil {
+					return errors.Wrapf(err, "error copying %v", m.BlobID)
+				}
+
+				numBlobs, bytesCopied := totalCopied.Add(m.Length)
+				progressMutex.Lock()
+				percentage := float64(0)
+				eta := "unknown"
+				speed := "-"
+				elapsedTime := time.Since(startTime)
+				if totalBytes > 0 {
+					percentage = hundredPercent * float64(bytesCopied) / float64(totalBytes)
+					if percentage > 0 {
+						totalTimeSeconds := elapsedTime.Seconds() * (hundredPercent / percentage)
+						etaTime := startTime.Add(time.Duration(totalTimeSeconds) * time.Second)
+						eta = fmt.Sprintf("%v (%v)", time.Until(etaTime).Round(time.Second), formatTimestamp(etaTime))
+						bps := float64(bytesCopied) * 8 / elapsedTime.Seconds() //nolint:gomnd
+						speed = units.BitsPerSecondsString(bps)
+					}
+				}
+
+				outputSyncProgress(
+					fmt.Sprintf("  Copied %v blobs (%v), Speed: %v, ETA: %v",
+						numBlobs, units.BytesStringBase10(bytesCopied), speed, eta))
+				progressMutex.Unlock()
+			}
+
+			for m := range deleteCh {
+				log(ctx).Debugf("[%v] Deleting %v (%v bytes)...\n", workerID, m.BlobID, m.Length)
+				if err := syncDeleteBlob(ctx, m, dst); err != nil {
+					return errors.Wrapf(err, "error deleting %v", m.BlobID)
+				}
+			}
+			return nil
+		})
+	}
+
+	if err := eg.Wait(); err != nil {
+		return errors.Wrap(err, "error copying blobs")
+	}
+
+	return nil
+}
+
+func sliceToChannel(ctx context.Context, md []blob.Metadata) chan blob.Metadata {
+	ch := make(chan blob.Metadata)
+
+	go func() {
+		defer close(ch)
+
+		for _, it := range md {
+			select {
+			case ch <- it:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return ch
+}
+
+func syncCopyBlob(ctx context.Context, m blob.Metadata, src, dst blob.Storage) error {
+	data, err := src.GetBlob(ctx, m.BlobID, 0, -1)
+	if err != nil {
+		if errors.Is(err, blob.ErrBlobNotFound) {
+			log(ctx).Infof("ignoring BLOB not found: %v", m.BlobID)
+			return nil
+		}
+
+		return errors.Wrapf(err, "error reading blob '%v' from source", m.BlobID)
+	}
+
+	if err := dst.PutBlob(ctx, m.BlobID, gather.FromSlice(data)); err != nil {
+		return errors.Wrapf(err, "error writing blob '%v' to destination", m.BlobID)
+	}
+
+	return nil
+}
+
+func syncDeleteBlob(ctx context.Context, m blob.Metadata, dst blob.Storage) error {
+	err := dst.DeleteBlob(ctx, m.BlobID)
+
+	if errors.Is(err, blob.ErrBlobNotFound) {
+		return nil
+	}
+
+	return err
+}
+
+func ensureRepositoriesHaveSameFormatBlob(ctx context.Context, src, dst blob.Storage) error {
+	srcData, err := src.GetBlob(ctx, repo.FormatBlobID, 0, -1)
+	if err != nil {
+		return errors.Wrap(err, "error reading format blob")
+	}
+
+	dstData, err := dst.GetBlob(ctx, repo.FormatBlobID, 0, -1)
+	if err != nil {
+		// target does not have format blob, save it there first.
+		if errors.Is(err, blob.ErrBlobNotFound) {
+			if *repositorySyncDestinationMustExist {
+				return errors.Errorf("destination repository does not have a format blob")
+			}
+
+			return dst.PutBlob(ctx, repo.FormatBlobID, gather.FromSlice(srcData))
+		}
+
+		return errors.Wrap(err, "error reading destination repository format blob")
+	}
+
+	if bytes.Equal(srcData, dstData) {
+		return nil
+	}
+
+	return errors.Errorf("destination repository contains incompatible data")
+}

cli/storage_providers.go

@@ -6,6 +6,7 @@ import (
 	"github.com/pkg/errors"
 	kingpin "gopkg.in/alecthomas/kingpin.v2"
 
+	"github.com/kopia/kopia/repo"
 	"github.com/kopia/kopia/repo/blob"
 )
 
@@ -56,4 +57,27 @@ func RegisterStorageConnectFlags(
 
 		return runRepairCommandWithStorage(ctx, st)
 	})
+
+	// Set up 'sync-to' subcommand
+	cc = repositorySyncCommand.Command(name, "Synchronize repository data to another repository in "+description)
+	flags(cc)
+	cc.Action(func(_ *kingpin.ParseContext) error {
+		ctx := rootContext()
+		st, err := connect(ctx, false)
+		if err != nil {
+			return errors.Wrap(err, "can't connect to storage")
+		}
+
+		rep, err := openRepository(ctx, nil, true)
+		if err != nil {
+			return errors.Wrap(err, "open repository")
+		}
+
+		dr, ok := rep.(*repo.DirectRepository)
+		if !ok {
+			return errors.Errorf("sync only supports directly-connected repositories")
+		}
+
+		return runSyncWithStorage(ctx, dr.BlobStorage(), st)
+	})
 }

go.sum

@@ -740,6 +740,7 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e h1:vcxGaoTs7kV8m5Np9uUNQin4
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a h1:WXEvlFVvvGxCJLG6REjsT03iWnKLEWinaScsxF2Vm2o=
 golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208 h1:qwRHBd0NqMbJxfbotnDhm2ByMI1Shq4Y6oRJo21SGJA=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

internal/blobtesting/eventually_consistent.go

@@ -260,6 +260,10 @@ func (s *eventuallyConsistentStorage) ConnectionInfo() blob.ConnectionInfo {
 	return s.realStorage.ConnectionInfo()
 }
 
+func (s *eventuallyConsistentStorage) DisplayName() string {
+	return s.realStorage.DisplayName()
+}
+
 // NewEventuallyConsistentStorage returns an eventually-consistent storage wrapper on top
 // of provided storage.
 func NewEventuallyConsistentStorage(st blob.Storage, listSettleTime time.Duration, timeNow func() time.Time) blob.Storage {

internal/blobtesting/faulty.go

@@ -92,6 +92,11 @@ func (s *FaultyStorage) ConnectionInfo() blob.ConnectionInfo {
 	return s.Base.ConnectionInfo()
 }
 
+// DisplayName implements blob.Storage
+func (s *FaultyStorage) DisplayName() string {
+	return s.Base.DisplayName()
+}
+
 func (s *FaultyStorage) getNextFault(ctx context.Context, method string, args ...interface{}) error {
 	s.mu.Lock()
 

internal/blobtesting/map.go

@@ -152,6 +152,10 @@ func (s *mapStorage) ConnectionInfo() blob.ConnectionInfo {
 	return blob.ConnectionInfo{}
 }
 
+func (s *mapStorage) DisplayName() string {
+	return "Map"
+}
+
 // NewMapStorage returns an implementation of Storage backed by the contents of given map.
 // Used primarily for testing.
 func NewMapStorage(data DataMap, keyTime map[blob.ID]time.Time, timeNow func() time.Time) blob.Storage {

repo/blob/azure/azure_storage.go

@@ -212,6 +212,10 @@ func (az *azStorage) ConnectionInfo() blob.ConnectionInfo {
 	}
 }
 
+func (az *azStorage) DisplayName() string {
+	return fmt.Sprintf("Azure: %v", az.Options.Container)
+}
+
 func (az *azStorage) Close(ctx context.Context) error {
 	return az.bucket.Close()
 }

repo/blob/b2/b2_storage.go

@@ -241,6 +241,10 @@ func (s *b2Storage) ConnectionInfo() blob.ConnectionInfo {
 	}
 }
 
+func (s *b2Storage) DisplayName() string {
+	return fmt.Sprintf("B2: %v", s.BucketName)
+}
+
 func (s *b2Storage) Close(ctx context.Context) error {
 	return nil
 }

repo/blob/filesystem/filesystem_storage.go

@@ -257,6 +257,10 @@ func (fs *fsStorage) ConnectionInfo() blob.ConnectionInfo {
 	}
 }
 
+func (fs *fsStorage) DisplayName() string {
+	return fmt.Sprintf("Filesystem: %v", fs.RootPath)
+}
+
 func (fs *fsStorage) Close(ctx context.Context) error {
 	return nil
 }

repo/blob/gcs/gcs_storage.go

@@ -209,6 +209,10 @@ func (gcs *gcsStorage) ConnectionInfo() blob.ConnectionInfo {
 	}
 }
 
+func (gcs *gcsStorage) DisplayName() string {
+	return fmt.Sprintf("GCS: %v", gcs.BucketName)
+}
+
 func (gcs *gcsStorage) Close(ctx context.Context) error {
 	return gcs.storageClient.Close()
 }

repo/blob/logging/logging_storage.go

@@ -82,6 +82,10 @@ func (s *loggingStorage) ConnectionInfo() blob.ConnectionInfo {
 	return s.base.ConnectionInfo()
 }
 
+func (s *loggingStorage) DisplayName() string {
+	return s.base.DisplayName()
+}
+
 // NewWrapper returns a Storage wrapper that logs all storage commands.
 func NewWrapper(wrapped blob.Storage, printf func(msg string, args ...interface{}), prefix string) blob.Storage {
 	return &loggingStorage{base: wrapped, printf: printf, prefix: prefix}

repo/blob/s3/s3_storage.go

@@ -211,6 +211,10 @@ func (s *s3Storage) String() string {
 	return fmt.Sprintf("s3://%v/%v", s.BucketName, s.Prefix)
 }
 
+func (s *s3Storage) DisplayName() string {
+	return fmt.Sprintf("S3: %v %v", s.Endpoint, s.BucketName)
+}
+
 type progressReader struct {
 	cb           blob.ProgressFunc
 	blobID       string

repo/blob/sftp/sftp_storage.go

@@ -181,6 +181,11 @@ func (s *sftpStorage) ConnectionInfo() blob.ConnectionInfo {
 	}
 }
 
+func (s *sftpStorage) DisplayName() string {
+	o := s.Impl.(*sftpImpl).Options
+	return fmt.Sprintf("SFTP %v@%v", o.Username, o.Host)
+}
+
 func (s *sftpStorage) Close(ctx context.Context) error {
 	if err := s.Impl.(*sftpImpl).cli.Close(); err != nil {
 		return errors.Wrap(err, "closing SFTP client")

repo/blob/storage.go

@@ -56,6 +56,9 @@ type Storage interface {
 
 	// Close releases all resources associated with storage.
 	Close(ctx context.Context) error
+
+	// Name of the storage used for quick identification by humans.
+	DisplayName() string
 }
 
 // ID is a string that represents blob identifier.

repo/blob/webdav/webdav_storage.go

@@ -166,6 +166,11 @@ func (d *davStorage) ConnectionInfo() blob.ConnectionInfo {
 	}
 }
 
+func (d *davStorage) DisplayName() string {
+	o := d.Storage.Impl.(*davStorageImpl).Options
+	return fmt.Sprintf("WebDAV: %v", o.URL)
+}
+
 func (d *davStorage) Close(ctx context.Context) error {
 	return nil
 }

tests/end_to_end_test/repository_sync_test.go

@@ -0,0 +1,49 @@
+package endtoend_test
+
+import (
+	"testing"
+
+	"github.com/kopia/kopia/tests/testenv"
+)
+
+func TestRepositorySync(t *testing.T) {
+	t.Parallel()
+
+	e := testenv.NewCLITest(t)
+	defer e.Cleanup(t)
+	defer e.RunAndExpectSuccess(t, "repo", "disconnect")
+
+	e.RunAndExpectSuccess(t, "repo", "create", "filesystem", "--path", e.RepoDir)
+
+	e.RunAndExpectSuccess(t, "snapshot", "create", sharedTestDataDir1)
+	e.RunAndExpectSuccess(t, "snapshot", "create", sharedTestDataDir2)
+
+	sources := e.ListSnapshotsAndExpectSuccess(t)
+
+	// synchronize repository blobs to another directory
+	dir2 := makeScratchDir(t)
+	e.RunAndExpectSuccess(t, "repo", "sync-to", "filesystem", "--path", dir2)
+
+	// synchronizing to empty directory fails with --must-exist
+	dir3 := makeScratchDir(t)
+	e.RunAndExpectFailure(t, "repo", "sync-to", "filesystem", "--path", dir3, "--must-exist")
+
+	// now connect to the new repository in new location
+	e.RunAndExpectSuccess(t, "repo", "connect", "filesystem", "--path", dir2)
+
+	// snapshot list should be the same
+	sources2 := e.ListSnapshotsAndExpectSuccess(t)
+	if got, want := len(sources2), len(sources); got != want {
+		t.Errorf("unexpected number of sources: %v, want %v in %#v", got, want, sources2)
+	}
+
+	// now create a whole new repository
+	e2 := testenv.NewCLITest(t)
+	defer e2.Cleanup(t)
+	defer e2.RunAndExpectSuccess(t, "repo", "disconnect")
+
+	e2.RunAndExpectSuccess(t, "repo", "create", "filesystem", "--path", e2.RepoDir)
+
+	// syncing to the directory should fail because it contains incompatible format blob.
+	e2.RunAndExpectFailure(t, "repo", "sync-to", "filesystem", "--path", dir2)
+}