Add Exist/NotExist filters and refactor few files

Author: larrabee

cli/cli.go

@@ -71,6 +71,8 @@ type args struct {
 	FilterMtimeAfter  int64    `arg:"--filter-after-mtime" help:"Sync only files modified after given unix timestamp"`
 	FilterMtimeBefore int64    `arg:"--filter-before-mtime" help:"Sync only files modified before given unix timestamp"`
 	FilterModified    bool     `arg:"--filter-modified" help:"Sync only modified files"`
+	FilterExist       bool     `arg:"--filter-exist" help:"Sync only files, that exist in target storage'"`
+	FilterExistNot    bool     `arg:"--filter-not-exist" help:"Sync only files, that doesn't exist in target storage'"`
 	// Misc
 	Workers           uint   `arg:"-w" help:"Workers count"`
 	Debug             bool   `arg:"-d" help:"Show debug logging"`

cli/main.go

@@ -98,15 +98,15 @@ WaitLoop:
 				continue WaitLoop
 			}
 
-			if cli.ErrorHandlingMask.Has(storage.HandleErrNotExist) && isErrNotExist(err) {
+			if cli.ErrorHandlingMask.Has(storage.HandleErrNotExist) && storage.IsErrNotExist(err) {
 				var objErr *pipeline.ObjectError
 				if errors.As(err, &objErr) {
 					log.Warnf("Skip missing object: %s", *objErr.Object.Key)
 				} else {
 					log.Warnf("Skip missing object, err: %s", err)
 				}
 				continue WaitLoop
-			} else if cli.ErrorHandlingMask.Has(storage.HandleErrPermission) && isErrPermission(err) {
+			} else if cli.ErrorHandlingMask.Has(storage.HandleErrPermission) && storage.IsErrPermission(err) {
 				var objErr *pipeline.ObjectError
 				if errors.As(err, &objErr) {
 					log.Warnf("Skip permission denied object: %s", *objErr.Object.Key)

cli/setup.go

@@ -127,6 +127,20 @@ func setupPipeline(syncGroup *pipeline.Group, cli *argsParsed) {
 		})
 	}
 
+	if cli.FilterExist {
+		syncGroup.AddPipeStep(pipeline.Step{
+			Name: "FilterObjectsExist",
+			Fn:   collection.FilterObjectsExist,
+		})
+	}
+
+	if cli.FilterExistNot {
+		syncGroup.AddPipeStep(pipeline.Step{
+			Name: "FilterObjectsExistNot",
+			Fn:   collection.FilterObjectsExistNot,
+		})
+	}
+
 	syncGroup.AddPipeStep(pipeline.Step{
 		Name:       "LoadObjData",
 		Fn:         collection.LoadObjectData,

pipeline/collection/filter.go

@@ -157,7 +157,7 @@ var FilterObjectsByMtimeBefore pipeline.StepFn = func(group *pipeline.Group, ste
 }
 
 // FilterObjectsModified accepts an input object and checks if it matches the filter
-// This filter gets object meta from target storage and compare object ETags. If Etags are equal object will be skipped
+// This filter read object meta from target storage and compare object ETags. If Etags are equal object will be skipped
 // For FS storage xattr support are required for proper work.
 var FilterObjectsModified pipeline.StepFn = func(group *pipeline.Group, stepNum int, input <-chan *storage.Object, output chan<- *storage.Object, errChan chan<- error) {
 	for obj := range input {
@@ -171,3 +171,39 @@ var FilterObjectsModified pipeline.StepFn = func(group *pipeline.Group, stepNum
 		}
 	}
 }
+
+// FilterObjectsExist accepts an input object and checks if it exist in target storage
+// This filter read object meta from target storage. Object will be processed only when it exist in target storage.
+var FilterObjectsExist pipeline.StepFn = func(group *pipeline.Group, stepNum int, input <-chan *storage.Object, output chan<- *storage.Object, errChan chan<- error) {
+	for obj := range input {
+		destObj := &storage.Object{
+			Key: obj.Key,
+		}
+		err := group.Target.GetObjectMeta(destObj)
+		if err == nil {
+			output <- obj
+		} else if storage.IsErrNotExist(err) {
+			continue
+		} else {
+			errChan <- &pipeline.ObjectError{Object: obj, Err: err}
+		}
+	}
+}
+
+// FilterObjectsExist accepts an input object and checks if it exist in target storage
+// This filter read object meta from target storage. Object will be processed only when it doesn't exist in target storage.
+var FilterObjectsExistNot pipeline.StepFn = func(group *pipeline.Group, stepNum int, input <-chan *storage.Object, output chan<- *storage.Object, errChan chan<- error) {
+	for obj := range input {
+		destObj := &storage.Object{
+			Key: obj.Key,
+		}
+		err := group.Target.GetObjectMeta(destObj)
+		if err == nil {
+			continue
+		} else if storage.IsErrNotExist(err) {
+			output <- obj
+		} else {
+			errChan <- &pipeline.ObjectError{Object: obj, Err: err}
+		}
+	}
+}

cli/error.go renamed to storage/error_handling.go

@@ -1,13 +1,15 @@
-package main
+package storage
 
 import (
+	"context"
 	"errors"
 	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/aws/request"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"os"
 )
 
-func isErrNotExist(err error) bool {
+func IsErrNotExist(err error) bool {
 	var aErr awserr.Error
 	if errors.As(err, &aErr) {
 		if (aErr.Code() == s3.ErrCodeNoSuchKey) || (aErr.Code() == "NotFound") {
@@ -21,7 +23,7 @@ func isErrNotExist(err error) bool {
 	return false
 }
 
-func isErrPermission(err error) bool {
+func IsErrPermission(err error) bool {
 	var aErr awserr.Error
 	if errors.As(err, &aErr) {
 		if aErr.Code() == "AccessDenied" {
@@ -34,3 +36,22 @@ func isErrPermission(err error) bool {
 	}
 	return false
 }
+
+func IsAwsContextCanceled(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	if errors.Is(err, context.Canceled) {
+		return true
+	}
+
+	var aErr awserr.Error
+	if ok := errors.As(err, &aErr); ok && aErr.OrigErr() == context.Canceled {
+		return true
+	} else if ok && aErr.Code() == request.CanceledErrorCode {
+		return true
+	}
+
+	return false
+}

storage/utils.go

@@ -1,10 +1,6 @@
 package storage
 
 import (
-	"context"
-	"errors"
-	"github.com/aws/aws-sdk-go/aws/awserr"
-	"github.com/aws/aws-sdk-go/aws/request"
 	"strings"
 )
 
@@ -30,22 +26,3 @@ const (
 	HandleErrPermission
 	HandleErrOther = 64
 )
-
-func IsAwsContextCanceled(err error) bool {
-	if err == nil {
-		return false
-	}
-
-	if errors.Is(err, context.Canceled) {
-		return true
-	}
-
-	var aErr awserr.Error
-	if ok := errors.As(err, &aErr); ok && aErr.OrigErr() == context.Canceled {
-		return true
-	} else if ok && aErr.Code() == request.CanceledErrorCode {
-		return true
-	}
-
-	return false
-}