cli: improved 'blob gc'

- do not remove blobs younger than 4h when performing blob GC,
  because they may have just been written
- parallelize deletes
- clean up console output

cli/command_blob_gc.go

@@ -2,10 +2,13 @@ package cli
 
 import (
 	"context"
-	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
 
+	"github.com/kopia/kopia/internal/units"
 	"github.com/kopia/kopia/repo"
 	"github.com/kopia/kopia/repo/blob"
 )
@@ -13,46 +16,81 @@ import (
 var (
 	blobGarbageCollectCommand       = blobCommands.Command("gc", "Garbage-collect unused blobs")
 	blobGarbageCollectCommandDelete = blobGarbageCollectCommand.Flag("delete", "Whether to delete unused blobs").String()
-	blobGarbageCollectParallel      = blobGarbageCollectCommand.Flag("parallel", "Number of parallel blob scans").Int()
+	blobGarbageCollectParallel      = blobGarbageCollectCommand.Flag("parallel", "Number of parallel blob scans").Default("16").Int()
+	blobGarbageCollectMinAge        = blobGarbageCollectCommand.Flag("min-age", "Garbage-collect blobs with minimum age").Default("24h").Duration()
 )
 
 func runBlobGarbageCollectCommand(ctx context.Context, rep *repo.Repository) error {
-	var mu sync.Mutex
+	const deleteQueueSize = 100
 
-	var unused []blob.Metadata
+	var totalUnreferencedSize, totalDeletedSize int64
+
+	var totalUnreferencedCount, totalDeletedCount int32
+
+	var eg errgroup.Group
+
+	unused := make(chan blob.Metadata, deleteQueueSize)
+
+	if *blobGarbageCollectCommandDelete == "yes" {
+		// start goroutines to delete blobs as they come.
+		for i := 0; i < *blobGarbageCollectParallel; i++ {
+			eg.Go(func() error {
+				for bm := range unused {
+					if err := rep.Blobs.DeleteBlob(ctx, bm.BlobID); err != nil {
+						return errors.Wrapf(err, "unable to delete blob %q", bm.BlobID)
+					}
+
+					del := atomic.AddInt64(&totalDeletedSize, bm.Length)
+					cnt := atomic.AddInt32(&totalDeletedCount, 1)
+					if cnt%100 == 0 {
+						printStderr("  deleted %v unreferenced blobs (%v)\n", cnt, units.BytesStringBase10(del))
+					}
+				}
+
+				return nil
+			})
+		}
+	}
+
+	// iterate unreferenced blobs and count them + optionally send to the channel to be deleted
+	printStderr("Looking for unreferenced blobs...\n")
 
 	if err := rep.Content.IterateUnreferencedBlobs(ctx, *blobGarbageCollectParallel, func(bm blob.Metadata) error {
-		mu.Lock()
-		unused = append(unused, bm)
-		mu.Unlock()
+		if age := time.Since(bm.Timestamp); age < *blobGarbageCollectMinAge {
+			printStderr("  preserving %v because it's too new (age: %v)\n", bm.BlobID, age)
+			return nil
+		}
+
+		atomic.AddInt64(&totalUnreferencedSize, bm.Length)
+		atomic.AddInt32(&totalUnreferencedCount, 1)
+
+		if *blobGarbageCollectCommandDelete == "yes" {
+			unused <- bm
+		}
+
 		return nil
 	}); err != nil {
 		return errors.Wrap(err, "error looking for unreferenced blobs")
 	}
 
-	if len(unused) == 0 {
-		printStderr("No unused blobs found.\n")
-		return nil
+	close(unused)
+
+	printStderr("Found %v blobs to delete (%v)\n", totalUnreferencedCount, units.BytesStringBase10(totalUnreferencedSize))
+
+	// wait for all delete workers to finish.
+	if err := eg.Wait(); err != nil {
+		return err
 	}
 
 	if *blobGarbageCollectCommandDelete != "yes" {
-		var totalBytes int64
-		for _, u := range unused {
-			totalBytes += u.Length
+		if totalUnreferencedCount > 0 {
+			printStderr("Pass --delete=yes to delete.\n")
 		}
 
-		printStderr("Would delete %v unused blobs (%v bytes), pass '--delete=yes' to actually delete.\n", len(unused), totalBytes)
-
 		return nil
 	}
 
-	for _, u := range unused {
-		printStderr("Deleting unused blob %q (%v bytes)...\n", u.BlobID, u.Length)
-
-		if err := rep.Blobs.DeleteBlob(ctx, u.BlobID); err != nil {
-			return errors.Wrapf(err, "unable to delete blob %q", u.BlobID)
-		}
-	}
+	printStderr("Deleted total %v unreferenced blobs (%v)\n", totalDeletedCount, units.BytesStringBase10(totalDeletedSize))
 
 	return nil
 }

repo/content/content_manager_iterate.go

@@ -226,7 +226,7 @@ func (bm *Manager) IterateContentInShortPacks(ctx context.Context, threshold int
 func (bm *Manager) IterateUnreferencedBlobs(ctx context.Context, parallellism int, callback func(blob.Metadata) error) error {
 	usedPacks := map[blob.ID]bool{}
 
-	log(ctx).Infof("determining blobs in use")
+	log(ctx).Debugf("determining blobs in use")
 	// find packs in use
 	if err := bm.IteratePacks(
 		ctx,
@@ -242,7 +242,7 @@ func (bm *Manager) IterateUnreferencedBlobs(ctx context.Context, parallellism in
 		return errors.Wrap(err, "error iterating packs")
 	}
 
-	log(ctx).Infof("found %v pack blobs in use", len(usedPacks))
+	log(ctx).Debugf("found %v pack blobs in use", len(usedPacks))
 
 	unusedCount := 0
 
@@ -272,7 +272,7 @@ func (bm *Manager) IterateUnreferencedBlobs(ctx context.Context, parallellism in
 		return errors.Wrap(err, "error iterating blobs")
 	}
 
-	log(ctx).Infof("found %v pack blobs not in use", unusedCount)
+	log(ctx).Debugf("found %v pack blobs not in use", unusedCount)
 
 	return nil
 }