diff --git a/cli/command_repository_sync.go b/cli/command_repository_sync.go new file mode 100644 index 000000000..9084011f2 --- /dev/null +++ b/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") +} diff --git a/cli/storage_providers.go b/cli/storage_providers.go index 3deae853e..5ba5ccf27 100644 --- a/cli/storage_providers.go +++ b/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) + }) } diff --git a/go.sum b/go.sum index 71a4cacdd..0029acf61 100644 --- a/go.sum +++ b/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= diff --git a/internal/blobtesting/eventually_consistent.go b/internal/blobtesting/eventually_consistent.go index c3ce774f6..f028b752f 100644 --- a/internal/blobtesting/eventually_consistent.go +++ b/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 { diff --git a/internal/blobtesting/faulty.go b/internal/blobtesting/faulty.go index 9b8bdfad3..ab994af7c 100644 --- a/internal/blobtesting/faulty.go +++ b/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() diff --git a/internal/blobtesting/map.go b/internal/blobtesting/map.go index 3395468ee..9b34df0e6 100644 --- a/internal/blobtesting/map.go +++ b/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 { diff --git a/repo/blob/azure/azure_storage.go b/repo/blob/azure/azure_storage.go index 1a0402f96..1ed674b29 100644 --- a/repo/blob/azure/azure_storage.go +++ b/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() } diff --git a/repo/blob/b2/b2_storage.go b/repo/blob/b2/b2_storage.go index c096a9ae8..1da5eae48 100644 --- a/repo/blob/b2/b2_storage.go +++ b/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 } diff --git a/repo/blob/filesystem/filesystem_storage.go b/repo/blob/filesystem/filesystem_storage.go index 1acc814f2..5ac072221 100644 --- a/repo/blob/filesystem/filesystem_storage.go +++ b/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 } diff --git a/repo/blob/gcs/gcs_storage.go b/repo/blob/gcs/gcs_storage.go index 8ba876fac..f81295152 100644 --- a/repo/blob/gcs/gcs_storage.go +++ b/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() } diff --git a/repo/blob/logging/logging_storage.go b/repo/blob/logging/logging_storage.go index 8ee334d1f..22ab98ba0 100644 --- a/repo/blob/logging/logging_storage.go +++ b/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} diff --git a/repo/blob/s3/s3_storage.go b/repo/blob/s3/s3_storage.go index c97bee92f..c1f2abfac 100644 --- a/repo/blob/s3/s3_storage.go +++ b/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 diff --git a/repo/blob/sftp/sftp_storage.go b/repo/blob/sftp/sftp_storage.go index 4bbb35aed..4f0d1c496 100644 --- a/repo/blob/sftp/sftp_storage.go +++ b/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") diff --git a/repo/blob/storage.go b/repo/blob/storage.go index 43a51e193..9aad37915 100644 --- a/repo/blob/storage.go +++ b/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. diff --git a/repo/blob/webdav/webdav_storage.go b/repo/blob/webdav/webdav_storage.go index 37f8f0db6..90e8a69e8 100644 --- a/repo/blob/webdav/webdav_storage.go +++ b/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 } diff --git a/tests/end_to_end_test/repository_sync_test.go b/tests/end_to_end_test/repository_sync_test.go new file mode 100644 index 000000000..3c12e7588 --- /dev/null +++ b/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) +}