added dedupe to syncer

Author: l3uddz

cmd/sync.go

@@ -115,6 +115,17 @@ func performSync(s *syncer.Syncer) error {
 		s.Log.Info("Finished syncs!")
 	}
 
+	/* Sync */
+	if len(s.Config.Remotes.Dedupe) > 0 {
+		s.Log.Info("Running dedupes...")
+
+		if err := s.Dedupe(nil); err != nil {
+			return errors.WithMessage(err, "failed performing all dedupes")
+		}
+
+		s.Log.Info("Finished dedupes!")
+	}
+
 	/* Move Server Side */
 	if len(s.Config.Remotes.MoveServerSide) > 0 {
 		s.Log.Info("Running move server-sides...")

config/syncer.go

@@ -3,12 +3,14 @@ package config
 type SyncerRemotes struct {
 	Copy           []string
 	Sync           []string
+	Dedupe         []string
 	MoveServerSide []UploaderRemotesMoveServerSide `mapstructure:"move_server_side"`
 }
 
 type SyncerRcloneParams struct {
 	Copy           []string
 	Sync           []string
+	Dedupe         []string
 	MoveServerSide []string `mapstructure:"move_server_side"`
 }
 

rclone/dedupe.go

@@ -0,0 +1,87 @@
+package rclone
+
+import (
+	"github.com/go-cmd/cmd"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+)
+
+/* Public */
+
+func Dedupe(remotePath string, additionalRcloneParams []string) (bool, int, error) {
+	// set variables
+	rLog := log.WithFields(logrus.Fields{
+		"action":      CMD_DEDUPE,
+		"remote_path": remotePath,
+	})
+	result := false
+
+	// generate required rclone parameters
+	params := []string{
+		CMD_DEDUPE,
+		remotePath,
+	}
+
+	if baseParams, err := getBaseParams(); err != nil {
+		return false, 1, errors.Wrapf(err, "failed generating baseParams to %q: %q", CMD_DEDUPE,
+			remotePath)
+	} else {
+		params = append(params, baseParams...)
+	}
+
+	if additionalParams, err := getAdditionalParams(CMD_DEDUPE, additionalRcloneParams); err != nil {
+		return false, 1, errors.Wrapf(err, "failed generating additionalParams to %s: %q",
+			CMD_DEDUPE, remotePath)
+	} else {
+		params = append(params, additionalParams...)
+	}
+
+	rLog.Debugf("Generated params: %v", params)
+
+	// setup cmd
+	cmdOptions := cmd.Options{
+		Buffered:  false,
+		Streaming: true,
+	}
+	rcloneCmd := cmd.NewCmdOptions(cmdOptions, cfg.Rclone.Path, params...)
+
+	// live stream logs
+	doneChan := make(chan struct{})
+	go func() {
+		defer close(doneChan)
+
+		for rcloneCmd.Stdout != nil || rcloneCmd.Stderr != nil {
+			select {
+			case line, open := <-rcloneCmd.Stdout:
+				if !open {
+					rcloneCmd.Stdout = nil
+					continue
+				}
+				log.Info(line)
+			case line, open := <-rcloneCmd.Stderr:
+				if !open {
+					rcloneCmd.Stderr = nil
+					continue
+				}
+				log.Info(line)
+			}
+		}
+	}()
+
+	// run command
+	rLog.Debug("Starting...")
+
+	status := <-rcloneCmd.Start()
+	<-doneChan
+
+	// check status
+	switch status.Exit {
+	case EXIT_SUCCESS:
+		result = true
+	default:
+		break
+	}
+
+	rLog.WithField("exit_code", status.Exit).Debug("Finished")
+	return result, status.Exit, status.Error
+}

rclone/param.go

@@ -58,8 +58,8 @@ func getAdditionalParams(cmd string, extraParams []string) ([]string, error) {
 		break
 	case CMD_DEDUPE:
 		params = append(params,
-			// keep newest duplicate file
-			"--dedupe-mode", "newest",
+			// tpslimit
+			"--tpslimit", "5",
 		)
 	default:
 		break

syncer/dedupe.go

@@ -0,0 +1,40 @@
+package syncer
+
+import (
+	"fmt"
+	"github.com/l3uddz/crop/rclone"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+)
+
+func (s *Syncer) Dedupe(additionalRcloneParams []string) error {
+	extraParams := s.Config.RcloneParams.Dedupe
+	if additionalRcloneParams != nil {
+		extraParams = append(extraParams, additionalRcloneParams...)
+	}
+
+	// iterate all remotes and run dedupe
+	for _, dedupeRemote := range s.Config.Remotes.Dedupe {
+		// set variables
+		rLog := s.Log.WithFields(logrus.Fields{
+			"dedupe_remote": dedupeRemote,
+		})
+
+		// dedupe remote
+		rLog.Info("Deduping...")
+		success, exitCode, err := rclone.Dedupe(dedupeRemote, extraParams)
+
+		// check result
+		if err != nil {
+			rLog.WithError(err).Errorf("Failed unexpectedly...")
+			return errors.WithMessagef(err, "dedupe failed unexpectedly with exit code: %v", exitCode)
+		} else if success {
+			// successful exit code
+			continue
+		}
+
+		return fmt.Errorf("dedupe failed with exit code: %v", exitCode)
+	}
+
+	return nil
+}

syncer/move.go

@@ -30,7 +30,7 @@ func (s *Syncer) Move(additionalRcloneParams []string) error {
 		extraParams = append(extraParams, additionalRcloneParams...)
 	}
 
-	// iterate all remotes and run copy
+	// iterate all remotes and run move
 	for _, move := range moveRemotes {
 		// set variables
 		attempts := 1