Merge branch 'master' into naman/export-backup

ee/backup/run.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/dgraph-io/badger/v3/options"
+	"golang.org/x/sync/errgroup"
 
 	"google.golang.org/grpc/credentials"
 
@@ -377,7 +378,7 @@ func runExportBackup() error {
 		return err
 	}
 	// Export the data from the p directories produced by the last step.
-	ch := make(chan error, len(files))
+	eg, _ := errgroup.WithContext(context.Background())
 	for _, f := range files {
 		if !f.IsDir() {
 			continue
@@ -386,22 +387,23 @@ func runExportBackup() error {
 		dir := filepath.Join(filepath.Join(tmpDir, f.Name()))
 		gid, err := strconv.ParseUint(strings.TrimPrefix(f.Name(), "p"), 32, 10)
 		if err != nil {
-			ch <- errors.Wrapf(err, "cannot export data inside DB at %s", dir)
+			fmt.Printf("WARNING WARNING WARNING: unable to get group id from directory "+
+				"inside DB at %s: %v", dir, err)
+			continue
 		}
-		go worker.StoreExport(&pb.ExportRequest{
-			GroupId:     uint32(gid),
-			ReadTs:      restore.Version,
-			UnixTs:      time.Now().Unix(),
-			Format:      opt.format,
-			Destination: exportDir,
-		}, dir, opt.key, ch)
+		eg.Go(func() error {
+			return worker.StoreExport(&pb.ExportRequest{
+				GroupId:     uint32(gid),
+				ReadTs:      restore.Version,
+				UnixTs:      time.Now().Unix(),
+				Format:      opt.format,
+				Destination: exportDir,
+			}, dir, opt.key)
+		})
 	}
 
-	for i := 0; i < len(files); i++ {
-		err := <-ch
-		if err != nil {
-			return err
-		}
+	if err := eg.Wait(); err != nil {
+		return errors.Wrapf(err, "error while exporting data")
 	}
 
 	// Clean up temporary directory.

worker/backup_common.go

@@ -101,24 +101,20 @@ func GetCredentialsFromRequest(req *pb.BackupRequest) *x.MinioCredentials {
 	}
 }
 
-func StoreExport(request *pb.ExportRequest, dir string,
-	key x.SensitiveByteSlice, ch chan error) {
-
+func StoreExport(request *pb.ExportRequest, dir string, key x.SensitiveByteSlice) error {
 	db, err := badger.OpenManaged(badger.DefaultOptions(dir).
 		WithSyncWrites(false).
 		WithValueThreshold(1 << 10).
 		WithNumVersionsToKeep(math.MaxInt32).
 		WithEncryptionKey(key))
 
 	if err != nil {
-		ch <- errors.Wrapf(err, "cannot open DB at %s", dir)
-		return
+		return err
 	}
 
 	_, err = exportInternal(context.Background(), request, db, true)
 	// It is important to close the db before sending err to ch. Else, we will see a memory
 	// leak.
 	db.Close()
-	ch <- errors.Wrapf(err, "cannot export data inside DB at %s", dir)
-	return
+	return errors.Wrapf(err, "cannot export data inside DB at %s", dir)
 }