diff --git a/cli/command_snapshot_gc.go b/cli/command_snapshot_gc.go new file mode 100644 index 000000000..0e1839f68 --- /dev/null +++ b/cli/command_snapshot_gc.go @@ -0,0 +1,130 @@ +package cli + +import ( + "context" + "sync" + "sync/atomic" + "time" + + "github.com/pkg/errors" + + "github.com/kopia/kopia/internal/units" + "github.com/kopia/kopia/repo" + "github.com/kopia/kopia/repo/content" + "github.com/kopia/kopia/repo/manifest" + "github.com/kopia/kopia/repo/object" + "github.com/kopia/kopia/snapshot" + "github.com/kopia/kopia/snapshot/snapshotfs" +) + +var ( + snapshotGCCommand = snapshotCommands.Command("gc", "Remove contents not used by any snapshot") + snapshotGCMinContentAge = snapshotGCCommand.Flag("min-age", "Minimum content age to allow deletion").Default("24h").Duration() + snapshotGCDelete = snapshotGCCommand.Flag("delete", "Delete unreferenced contents").Bool() +) + +func findInUseContentIDs(ctx context.Context, rep *repo.Repository, used *sync.Map) error { + w := snapshotfs.NewTreeWalker() + + ids, err := snapshot.ListSnapshotManifests(ctx, rep, nil) + if err != nil { + return errors.Wrap(err, "unable to list snapshot manifest IDs") + } + + manifests, err := snapshot.LoadSnapshots(ctx, rep, ids) + if err != nil { + return errors.Wrap(err, "unable to load manifest IDs") + } + + for _, m := range manifests { + root, err := snapshotfs.SnapshotRoot(rep, m) + if err != nil { + return errors.Wrap(err, "unable to get snapshot root") + } + w.RootEntries = append(w.RootEntries, root) + } + + w.ObjectCallback = func(oid object.ID) error { + _, contentIDs, err := rep.Objects.VerifyObject(ctx, oid) + if err != nil { + return errors.Wrapf(err, "error verifying %v", oid) + } + for _, cid := range contentIDs { + used.Store(cid, nil) + } + return nil + } + + log.Infof("looking for active contents") + if err := w.Run(ctx); err != nil { + return errors.Wrap(err, "error walking snapshot tree") + } + + return nil +} + +func runSnapshotGCCommand(ctx context.Context, rep *repo.Repository) error { + var used sync.Map + if err := findInUseContentIDs(ctx, rep, &used); err != nil { + return errors.Wrap(err, "unable to find in-use content ID") + } + + var unusedCount, inUseCount, systemCount, tooRecentCount int32 + var totalUnusedBytes, totalInUseBytes, totalSystemBytes, totalTooRecentBytes int64 + + log.Infof("looking for unreferenced contents") + if err := rep.Content.IterateContents(content.IterateOptions{}, func(ci content.Info) error { + if manifest.ContentPrefix == ci.ID.Prefix() { + atomic.AddInt32(&systemCount, 1) + atomic.AddInt64(&totalSystemBytes, int64(ci.Length)) + return nil + } + + if _, ok := used.Load(ci.ID); !ok { + if time.Since(ci.Timestamp()) < *snapshotGCMinContentAge { + log.Debugf("recent unreferenced content %v (%v bytes, modified %v)", ci.ID, ci.Length, ci.Timestamp()) + atomic.AddInt32(&tooRecentCount, 1) + atomic.AddInt64(&totalTooRecentBytes, int64(ci.Length)) + return nil + } + log.Debugf("unreferenced %v (%v bytes, modified %v)", ci.ID, ci.Length, ci.Timestamp()) + cnt := atomic.AddInt32(&unusedCount, 1) + totalSize := atomic.AddInt64(&totalUnusedBytes, int64(ci.Length)) + if *snapshotGCDelete { + if err := rep.Content.DeleteContent(ci.ID); err != nil { + return errors.Wrap(err, "error deleting content") + } + } + + if cnt%100000 == 0 { + log.Infof("... found %v unused contents so far (%v bytes)", cnt, units.BytesStringBase2(totalSize)) + if *snapshotGCDelete { + if err := rep.Flush(ctx); err != nil { + return errors.Wrap(err, "flush error") + } + } + } + } else { + atomic.AddInt32(&inUseCount, 1) + atomic.AddInt64(&totalInUseBytes, int64(ci.Length)) + } + return nil + }); err != nil { + return errors.Wrap(err, "error iterating contents") + } + + log.Infof("found %v unused contents (%v bytes)", unusedCount, units.BytesStringBase2(totalUnusedBytes)) + log.Infof("found %v unused contents that are too recent to delete (%v bytes)", tooRecentCount, units.BytesStringBase2(totalTooRecentBytes)) + log.Infof("found %v in-use contents (%v bytes)", inUseCount, units.BytesStringBase2(totalInUseBytes)) + log.Infof("found %v in-use system-contents (%v bytes)", systemCount, units.BytesStringBase2(totalSystemBytes)) + + if unusedCount > 0 && !*snapshotGCDelete { + return errors.Errorf("Not deleting because '--delete' flag was not set.") + } + + return nil +} + +func init() { + snapshotGCCommand.Action(repositoryAction(runSnapshotGCCommand)) +} diff --git a/cli/command_object_verify.go b/cli/command_snapshot_verify.go similarity index 92% rename from cli/command_object_verify.go rename to cli/command_snapshot_verify.go index ec096d7d8..a0e6386f2 100644 --- a/cli/command_object_verify.go +++ b/cli/command_snapshot_verify.go @@ -21,7 +21,7 @@ import ( ) var ( - verifyCommand = app.Command("verify", "Verify the contents of stored object") + verifyCommand = snapshotCommands.Command("verify", "Verify the contents of stored snapshot") verifyCommandErrorThreshold = verifyCommand.Flag("max-errors", "Maximum number of errors before stopping").Default("0").Int() verifyCommandDirObjectIDs = verifyCommand.Flag("directory-id", "Directory object IDs to verify").Strings() verifyCommandFileObjectIDs = verifyCommand.Flag("file-id", "File object IDs to verify").Strings() @@ -96,8 +96,8 @@ func (v *verifier) enqueueVerifyDirectory(ctx context.Context, oid object.ID, pa if !v.shouldEnqueue(oid) { return } - v.workQueue.EnqueueFront(func() { - v.doVerifyDirectory(ctx, oid, path) + v.workQueue.EnqueueFront(func() error { + return v.doVerifyDirectory(ctx, oid, path) }) } @@ -106,19 +106,19 @@ func (v *verifier) enqueueVerifyObject(ctx context.Context, oid object.ID, path if !v.shouldEnqueue(oid) { return } - v.workQueue.EnqueueBack(func() { - v.doVerifyObject(ctx, oid, path, expectedLength) + v.workQueue.EnqueueBack(func() error { + return v.doVerifyObject(ctx, oid, path, expectedLength) }) } -func (v *verifier) doVerifyDirectory(ctx context.Context, oid object.ID, path string) { +func (v *verifier) doVerifyDirectory(ctx context.Context, oid object.ID, path string) error { log.Debugf("verifying directory %q (%v)", path, oid) d := snapshotfs.DirectoryEntry(v.rep, oid, nil) entries, err := d.Readdir(ctx) if err != nil { v.reportError(path, errors.Wrapf(err, "error reading %v", oid)) - return + return nil } for _, e := range entries { @@ -134,9 +134,11 @@ func (v *verifier) doVerifyDirectory(ctx context.Context, oid object.ID, path st v.enqueueVerifyObject(ctx, objectID, childPath, e.Size()) } } + + return nil } -func (v *verifier) doVerifyObject(ctx context.Context, oid object.ID, path string, expectedLength int64) { +func (v *verifier) doVerifyObject(ctx context.Context, oid object.ID, path string, expectedLength int64) error { if expectedLength < 0 { log.Debugf("verifying object %v", oid) } else { @@ -160,6 +162,8 @@ func (v *verifier) doVerifyObject(ctx context.Context, oid object.ID, path strin v.reportError(path, errors.Wrapf(err, "error reading object %v", oid)) } } + + return nil } func (v *verifier) readEntireObject(ctx context.Context, oid object.ID, path string) error { @@ -191,7 +195,9 @@ func runVerifyCommand(ctx context.Context, rep *repo.Repository) error { } v.workQueue.ProgressCallback = v.progressCallback - v.workQueue.Process(*verifyCommandParallel) + if err := v.workQueue.Process(*verifyCommandParallel); err != nil { + return errors.Wrap(err, "error processing work queue") + } if len(v.errors) == 0 { return nil diff --git a/internal/parallelwork/parallel_work_queue.go b/internal/parallelwork/parallel_work_queue.go index 98426c45f..cd6fc8873 100644 --- a/internal/parallelwork/parallel_work_queue.go +++ b/internal/parallelwork/parallel_work_queue.go @@ -20,17 +20,20 @@ type Queue struct { ProgressCallback func(enqueued, active, completed int64) } +// CallbackFunc is a callback function +type CallbackFunc func() error + // EnqueueFront adds the work to the front of the queue. -func (v *Queue) EnqueueFront(callback func()) { +func (v *Queue) EnqueueFront(callback CallbackFunc) { v.enqueue(true, callback) } // EnqueueBack adds the work to the back of the queue. -func (v *Queue) EnqueueBack(callback func()) { +func (v *Queue) EnqueueBack(callback CallbackFunc) { v.enqueue(false, callback) } -func (v *Queue) enqueue(front bool, callback func()) { +func (v *Queue) enqueue(front bool, callback CallbackFunc) { v.monitor.L.Lock() defer v.monitor.L.Unlock() @@ -48,8 +51,9 @@ func (v *Queue) enqueue(front bool, callback func()) { // Process starts N workers, which will be processing elements in the queue until the queue // is empty and all workers are idle. -func (v *Queue) Process(workers int) { +func (v *Queue) Process(workers int) error { var wg sync.WaitGroup + errors := make(chan error, workers) for i := 0; i < workers; i++ { wg.Add(1) @@ -63,16 +67,23 @@ func (v *Queue) Process(workers int) { if callback == nil { break } - callback() + if err := callback(); err != nil { + errors <- err + break + } v.completed() } }(i) } wg.Wait() + close(errors) + + // return first error or nil + return <-errors } -func (v *Queue) dequeue() func() { +func (v *Queue) dequeue() CallbackFunc { v.monitor.L.Lock() defer v.monitor.L.Unlock() @@ -91,7 +102,7 @@ func (v *Queue) dequeue() func() { front := v.queueItems.Front() v.queueItems.Remove(front) - return front.Value.(func()) + return front.Value.(CallbackFunc) } func (v *Queue) completed() { diff --git a/repo/manifest/manifest_manager.go b/repo/manifest/manifest_manager.go index 83dc2c49f..a30df80a4 100644 --- a/repo/manifest/manifest_manager.go +++ b/repo/manifest/manifest_manager.go @@ -23,7 +23,7 @@ var log = repologging.Logger("kopia/manifest") // ErrNotFound is returned when the metadata item is not found. var ErrNotFound = errors.New("not found") -const manifestContentPrefix = "m" +const ContentPrefix = "m" const autoCompactionContentCount = 16 type contentManager interface { @@ -228,7 +228,7 @@ func (m *Manager) flushPendingEntriesLocked(ctx context.Context) (content.ID, er mustSucceed(gz.Flush()) mustSucceed(gz.Close()) - contentID, err := m.b.WriteContent(ctx, buf.Bytes(), manifestContentPrefix) + contentID, err := m.b.WriteContent(ctx, buf.Bytes(), ContentPrefix) if err != nil { return "", err } @@ -287,7 +287,7 @@ func (m *Manager) loadCommittedContentsLocked(ctx context.Context) error { manifests = map[content.ID]manifest{} err := m.b.IterateContents(content.IterateOptions{ - Prefix: manifestContentPrefix, + Prefix: ContentPrefix, Parallel: 8, }, func(ci content.Info) error { man, err := m.loadManifestContent(ctx, ci.ID) diff --git a/repo/manifest/manifest_manager_test.go b/repo/manifest/manifest_manager_test.go index dd0c15b1e..0e4915bb2 100644 --- a/repo/manifest/manifest_manager_test.go +++ b/repo/manifest/manifest_manager_test.go @@ -99,7 +99,7 @@ func TestManifest(t *testing.T) { foundContents := 0 if err := mgr.b.IterateContents( - content.IterateOptions{Prefix: manifestContentPrefix}, + content.IterateOptions{Prefix: ContentPrefix}, func(ci content.Info) error { foundContents++ return nil diff --git a/snapshot/snapshotfs/snapshot_tree_walker.go b/snapshot/snapshotfs/snapshot_tree_walker.go new file mode 100644 index 000000000..38518eaa4 --- /dev/null +++ b/snapshot/snapshotfs/snapshot_tree_walker.go @@ -0,0 +1,74 @@ +package snapshotfs + +import ( + "context" + "runtime" + "sync" + + "github.com/pkg/errors" + + "github.com/kopia/kopia/fs" + "github.com/kopia/kopia/internal/parallelwork" + "github.com/kopia/kopia/repo/object" +) + +type TreeWalker struct { + Parallelism int + RootEntries []fs.Entry + ObjectCallback func(oid object.ID) error + + enqueued sync.Map + queue *parallelwork.Queue +} + +func oidOf(entry fs.Entry) object.ID { + return entry.(object.HasObjectID).ObjectID() +} + +func (w *TreeWalker) enqueueEntry(ctx context.Context, entry fs.Entry) { + oid := oidOf(entry) + if _, existing := w.enqueued.LoadOrStore(oid, w); existing { + return + } + + w.queue.EnqueueBack(func() error { return w.processEntry(ctx, entry) }) +} + +func (w *TreeWalker) processEntry(ctx context.Context, entry fs.Entry) error { + err := w.ObjectCallback(oidOf(entry)) + if err != nil { + return err + } + + if dir, ok := entry.(fs.Directory); ok { + entries, err := dir.Readdir(ctx) + if err != nil { + return errors.Wrap(err, "error reading directory") + } + + for _, ent := range entries { + w.enqueueEntry(ctx, ent) + } + } + + return nil +} + +func (w *TreeWalker) Run(ctx context.Context) error { + for _, root := range w.RootEntries { + w.enqueueEntry(ctx, root) + } + w.queue.ProgressCallback = func(enqueued, active, completed int64) { + log.Infof("processed(%v/%v) active %v", completed, enqueued, active) + + } + return w.queue.Process(w.Parallelism) +} + +// NewTreeWalker creates new tree walker. +func NewTreeWalker() *TreeWalker { + return &TreeWalker{ + Parallelism: 4 * runtime.NumCPU(), + queue: parallelwork.NewQueue(), + } +} diff --git a/tests/end_to_end_test/end_to_end_test.go b/tests/end_to_end_test/end_to_end_test.go index 18f4e05bc..0eb91690d 100644 --- a/tests/end_to_end_test/end_to_end_test.go +++ b/tests/end_to_end_test/end_to_end_test.go @@ -222,6 +222,65 @@ func TestEndToEnd(t *testing.T) { }) } +func TestSnapshotGC(t *testing.T) { + e := newTestEnv(t) + defer e.cleanup(t) + defer e.runAndExpectSuccess(t, "repo", "disconnect") + + e.runAndExpectSuccess(t, "repo", "create", "filesystem", "--path", e.repoDir) + + expectedContentCount := len(e.runAndExpectSuccess(t, "content", "list")) + + dataDir := filepath.Join(e.dataDir, "dir1") + assertNoError(t, os.MkdirAll(dataDir, 0777)) + assertNoError(t, ioutil.WriteFile(filepath.Join(dataDir, "some-file1"), []byte(` +hello world +how are you +`), 0600)) + + // take a snapshot of a directory with 1 file + e.runAndExpectSuccess(t, "snap", "create", dataDir) + + // data block + directory block + manifest block + expectedContentCount += 3 + e.runAndVerifyOutputLineCount(t, expectedContentCount, "content", "list") + + // now delete all manifests, making the content unreachable + for _, line := range e.runAndExpectSuccess(t, "snap", "list", "-m") { + p := strings.Index(line, "manifest:") + if p >= 0 { + manifestID := strings.TrimPrefix(strings.Split(line[p:], " ")[0], "manifest:") + t.Logf("manifestID: %v", manifestID) + e.runAndExpectSuccess(t, "manifest", "rm", manifestID) + } + } + + // deletion of manifests creates a new manifest + expectedContentCount++ + + // run verification + e.runAndExpectSuccess(t, "snapshot", "verify", "--all-sources") + + // garbage-collect in dry run mode + e.runAndExpectSuccess(t, "snapshot", "gc") + + // data block + directory block + manifest block + manifest block from manifest deletion + e.runAndVerifyOutputLineCount(t, expectedContentCount, "content", "list") + + // garbage-collect for real, but contents are too recent so won't be deleted + e.runAndExpectSuccess(t, "snapshot", "gc", "--delete") + + // data block + directory block + manifest block + manifest block from manifest deletion + e.runAndVerifyOutputLineCount(t, expectedContentCount, "content", "list") + + // garbage-collect for real, this time without age limit + e.runAndExpectSuccess(t, "snapshot", "gc", "--delete", "--min-age", "0s") + + // two contents are deleted + expectedContentCount -= 2 + e.runAndVerifyOutputLineCount(t, expectedContentCount, "content", "list") +} + func TestDiff(t *testing.T) { e := newTestEnv(t) defer e.cleanup(t)