Merge pull request #23 from sunoce/add_flag_for_skipping_errors

Added flag to skip file errors during copy
maorfr · Jun 15, 2022 · 326c955 · 326c955
2 parents 3d8a7af + 1650453
commit 326c955
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 16 deletions.
diff --git a/cmd/skbn.go b/cmd/skbn.go
@@ -35,10 +35,11 @@ func NewRootCmd(args []string) *cobra.Command {
 }
 
 type cpCmd struct {
-	src        string
-	dst        string
-	parallel   int
-	bufferSize float64
+	src            string
+	dst            string
+	parallel       int
+	bufferSize     float64
+	skipErrorFiles bool
 
 	out io.Writer
 }
@@ -52,7 +53,7 @@ func NewCpCmd(out io.Writer) *cobra.Command {
 		Short: "Copy files or directories Kubernetes and Cloud storage",
 		Long:  ``,
 		Run: func(cmd *cobra.Command, args []string) {
-			if err := skbn.Copy(c.src, c.dst, c.parallel, c.bufferSize); err != nil {
+			if err := skbn.Copy(c.src, c.dst, c.parallel, c.bufferSize, c.skipErrorFiles); err != nil {
 				log.Fatal(err)
 			}
 		},

diff --git a/examples/code/example.go b/examples/code/example.go
@@ -14,7 +14,7 @@ func main() {
 	bufferSize := 1.0 // 1GB of in memory buffer size
 
 	start := time.Now()
-	if err := skbn.Copy(src, dst, parallel, bufferSize); err != nil {
+	if err := skbn.Copy(src, dst, parallel, bufferSize, false); err != nil {
 		log.Fatal(err)
 	}
 	elapsed := time.Since(start)

diff --git a/pkg/skbn/skbn.go b/pkg/skbn/skbn.go
@@ -11,7 +11,6 @@ import (
 	"github.com/maorfr/skbn/pkg/utils"
 
 	"github.com/djherbis/buffer"
-	"gopkg.in/djherbis/nio.v2"
 )
 
 // FromToPair is a pair of FromPath and ToPath
@@ -21,7 +20,7 @@ type FromToPair struct {
 }
 
 // Copy copies files from src to dst
-func Copy(src, dst string, parallel int, bufferSize float64) error {
+func Copy(src, dst string, parallel int, bufferSize float64, skipErrorFiles bool) error {
 	srcPrefix, srcPath := utils.SplitInTwo(src, "://")
 	dstPrefix, dstPath := utils.SplitInTwo(dst, "://")
 
@@ -37,7 +36,7 @@ func Copy(src, dst string, parallel int, bufferSize float64) error {
 	if err != nil {
 		return err
 	}
-	err = PerformCopy(srcClient, dstClient, srcPrefix, dstPrefix, fromToPaths, parallel, bufferSize)
+	err = PerformCopy(srcClient, dstClient, srcPrefix, dstPrefix, fromToPaths, parallel, bufferSize, skipErrorFiles)
 	if err != nil {
 		return err
 	}
@@ -103,7 +102,7 @@ func GetFromToPaths(srcClient interface{}, srcPrefix, srcPath, dstPath string) (
 }
 
 // PerformCopy performs the actual copy action
-func PerformCopy(srcClient, dstClient interface{}, srcPrefix, dstPrefix string, fromToPaths []FromToPair, parallel int, bufferSize float64) error {
+func PerformCopy(srcClient, dstClient interface{}, srcPrefix, dstPrefix string, fromToPaths []FromToPair, parallel int, bufferSize float64, skipErrorFiles bool) error {
 	// Execute in parallel
 	totalFiles := len(fromToPaths)
 	if parallel == 0 {
@@ -115,7 +114,7 @@ func PerformCopy(srcClient, dstClient interface{}, srcPrefix, dstPrefix string,
 	currentLine := 0
 	for _, ftp := range fromToPaths {
 
-		if len(errc) != 0 {
+		if !skipErrorFiles && len(errc) != 0 {
 			break
 		}
 
@@ -127,39 +126,48 @@ func PerformCopy(srcClient, dstClient interface{}, srcPrefix, dstPrefix string,
 
 		go func(srcClient, dstClient interface{}, srcPrefix, fromPath, dstPrefix, toPath, currentLinePadded string, totalFiles int) {
 
-			if len(errc) != 0 {
+			if !skipErrorFiles && len(errc) != 0 {
 				return
 			}
 
 			newBufferSize := (int64)(bufferSize * 1024 * 1024) // may not be super accurate
 			buf := buffer.New(newBufferSize)
 			pr, pw := nio.Pipe(buf)
+			fileErrorChannel := make(chan error, 1)
 
 			log.Printf("[%s/%d] copy: %s://%s -> %s://%s", currentLinePadded, totalFiles, srcPrefix, fromPath, dstPrefix, toPath)
 
 			go func() {
 				defer pw.Close()
-				if len(errc) != 0 {
+				if !skipErrorFiles && len(errc) != 0 {
 					return
 				}
 				err := Download(srcClient, srcPrefix, fromPath, pw)
 				if err != nil {
 					log.Println(err, fmt.Sprintf(" src: file: %s", fromPath))
-					errc <- err
+					fileErrorChannel <- err
+					if !skipErrorFiles {
+						errc <- err
+					}
 				}
 			}()
 
 			go func() {
 				defer pr.Close()
 				defer bwg.Done()
-				if len(errc) != 0 {
+				if len(fileErrorChannel) != 0 {
+					return
+				}
+				if !skipErrorFiles && len(errc) != 0 {
 					return
 				}
 				defer log.Printf("[%s/%d] done: %s://%s -> %s://%s", currentLinePadded, totalFiles, srcPrefix, fromPath, dstPrefix, toPath)
 				err := Upload(dstClient, dstPrefix, toPath, fromPath, pr)
 				if err != nil {
 					log.Println(err, fmt.Sprintf(" dst: file: %s", toPath))
-					errc <- err
+					if !skipErrorFiles {
+						errc <- err
+					}
 				}
 			}()
 		}(srcClient, dstClient, srcPrefix, ftp.FromPath, dstPrefix, ftp.ToPath, currentLinePadded, totalFiles)