Bump version to 0.5.1

.github/workflows/tunasync.yml

@@ -28,6 +28,11 @@ jobs:
         make tunasync
         make tunasynctl
 
+    - name: Keep artifacts
+      uses: actions/upload-artifact@v1
+      with:
+        name: tunasync-bin
+        path: build/
 
   test:
     name: Test

internal/util.go

@@ -86,13 +86,32 @@ func GetJSON(url string, obj interface{}, client *http.Client) (*http.Response,
 	return resp, json.Unmarshal(body, obj)
 }
 
-func ExtractSizeFromRsyncLog(content []byte) string {
+// FindAllSubmatchInFile calls re.FindAllSubmatch to find matches in given file
-	// (?m) flag enables multi-line mode
+func FindAllSubmatchInFile(fileName string, re *regexp.Regexp) (matches [][][]byte, err error) {
-	re := regexp.MustCompile(`(?m)^Total file size: ([0-9\.]+[KMGTP]?) bytes`)
+	if fileName == "/dev/null" {
-	matches := re.FindAllSubmatch(content, -1)
+		err = errors.New("Invalid log file")
-	// fmt.Printf("%q\n", matches)
+		return
-	if len(matches) == 0 {
+	}
+	if content, err := ioutil.ReadFile(fileName); err == nil {
+		matches = re.FindAllSubmatch(content, -1)
+		// fmt.Printf("FindAllSubmatchInFile: %q\n", matches)
+	}
+	return
+}
+
+// ExtractSizeFromLog uses a regexp to extract the size from log files
+func ExtractSizeFromLog(logFile string, re *regexp.Regexp) string {
+	matches, _ := FindAllSubmatchInFile(logFile, re)
+	if matches == nil || len(matches) == 0 {
 		return ""
 	}
+	// return the first capture group of the last occurrence
 	return string(matches[len(matches)-1][1])
 }
+
+// ExtractSizeFromRsyncLog extracts the size from rsync logs
+func ExtractSizeFromRsyncLog(logFile string) string {
+	// (?m) flag enables multi-line mode
+	re := regexp.MustCompile(`(?m)^Total file size: ([0-9\.]+[KMGTP]?) bytes`)
+	return ExtractSizeFromLog(logFile, re)
+}

internal/util_test.go

@@ -1,6 +1,9 @@
 package internal
 
 import (
+	"io/ioutil"
+	"os"
+	"path/filepath"
 	"testing"
 
 	. "github.com/smartystreets/goconvey/convey"
@@ -26,7 +29,14 @@ sent 7.55M bytes  received 823.25M bytes  5.11M bytes/sec
 total size is 1.33T  speedup is 1,604.11
 `
 	Convey("Log parser should work", t, func() {
-		res := ExtractSizeFromRsyncLog([]byte(realLogContent))
+		tmpDir, err := ioutil.TempDir("", "tunasync")
+		So(err, ShouldBeNil)
+		defer os.RemoveAll(tmpDir)
+		logFile := filepath.Join(tmpDir, "rs.log")
+		err = ioutil.WriteFile(logFile, []byte(realLogContent), 0755)
+		So(err, ShouldBeNil)
+
+		res := ExtractSizeFromRsyncLog(logFile)
 		So(res, ShouldEqual, "1.33T")
 	})
 }

internal/version.go

@@ -1,3 +1,4 @@
 package internal
 
-const Version string = "0.4.2"
+// Version of the program
+const Version string = "0.5.1"

worker/cmd_provider.go

@@ -3,11 +3,11 @@ package worker
 import (
 	"errors"
 	"fmt"
-	"io/ioutil"
 	"regexp"
 	"time"
 
 	"github.com/anmitsu/go-shlex"
+	"github.com/tuna/tunasync/internal"
 )
 
 type cmdConfig struct {
@@ -18,13 +18,16 @@ type cmdConfig struct {
 	retry                       int
 	env                         map[string]string
 	failOnMatch                 string
+	sizePattern                 string
 }
 
 type cmdProvider struct {
 	baseProvider
 	cmdConfig
 	command     []string
+	dataSize    string
 	failOnMatch *regexp.Regexp
+	sizePattern *regexp.Regexp
 }
 
 func newCmdProvider(c cmdConfig) (*cmdProvider, error) {
@@ -59,6 +62,14 @@ func newCmdProvider(c cmdConfig) (*cmdProvider, error) {
 		}
 		provider.failOnMatch = failOnMatch
 	}
+	if len(c.sizePattern) > 0 {
+		var err error
+		sizePattern, err := regexp.Compile(c.sizePattern)
+		if err != nil {
+			return nil, errors.New("size-pattern regexp error: " + err.Error())
+		}
+		provider.sizePattern = sizePattern
+	}
 
 	return provider, nil
 }
@@ -71,7 +82,12 @@ func (p *cmdProvider) Upstream() string {
 	return p.upstreamURL
 }
 
+func (p *cmdProvider) DataSize() string {
+	return p.dataSize
+}
+
 func (p *cmdProvider) Run() error {
+	p.dataSize = ""
 	if err := p.Start(); err != nil {
 		return err
 	}
@@ -79,16 +95,18 @@ func (p *cmdProvider) Run() error {
 		return err
 	}
 	if p.failOnMatch != nil {
-		if logContent, err := ioutil.ReadFile(p.LogFile()); err == nil {
+		matches, err := internal.FindAllSubmatchInFile(p.LogFile(), p.failOnMatch)
-			matches := p.failOnMatch.FindAllSubmatch(logContent, -1)
+		fmt.Printf("FindAllSubmatchInFile: %q\n", matches)
-			if len(matches) != 0 {
+		if err != nil {
-				logger.Debug("Fail-on-match: %r", matches)
-				return errors.New(
-					fmt.Sprintf("Fail-on-match regexp found %d matches", len(matches)))
-			}
-		} else {
 			return err
 		}
+		if len(matches) != 0 {
+			logger.Debug("Fail-on-match: %r", matches)
+			return fmt.Errorf("Fail-on-match regexp found %d matches", len(matches))
+		}
+	}
+	if p.sizePattern != nil {
+		p.dataSize = internal.ExtractSizeFromLog(p.LogFile(), p.sizePattern)
 	}
 	return nil
 }

worker/config.go

@@ -131,6 +131,7 @@ type mirrorConfig struct {
 
 	Command       string   `toml:"command"`
 	FailOnMatch   string   `toml:"fail_on_match"`
+	SizePattern   string   `toml:"size_pattern"`
 	UseIPv6       bool     `toml:"use_ipv6"`
 	UseIPv4       bool     `toml:"use_ipv4"`
 	ExcludeFile   string   `toml:"exclude_file"`

worker/provider.go

@@ -113,6 +113,7 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 			command:     mirror.Command,
 			workingDir:  mirrorDir,
 			failOnMatch: mirror.FailOnMatch,
+			sizePattern: mirror.SizePattern,
 			logDir:      logDir,
 			logFile:     filepath.Join(logDir, "latest.log"),
 			interval:    time.Duration(mirror.Interval) * time.Minute,
@@ -135,6 +136,7 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 			excludeFile:       mirror.ExcludeFile,
 			extraOptions:      mirror.RsyncOptions,
 			overriddenOptions: mirror.RsyncOverride,
+			rsyncEnv:          mirror.Env,
 			workingDir:        mirrorDir,
 			logDir:            logDir,
 			logFile:           filepath.Join(logDir, "latest.log"),
@@ -159,6 +161,7 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 			password:      mirror.Password,
 			excludeFile:   mirror.ExcludeFile,
 			extraOptions:  mirror.RsyncOptions,
+			rsyncEnv:      mirror.Env,
 			workingDir:    mirrorDir,
 			logDir:        logDir,
 			logFile:       filepath.Join(logDir, "latest.log"),

worker/provider_test.go

@@ -5,6 +5,7 @@ import (
 	"io/ioutil"
 	"os"
 	"path/filepath"
+	"strconv"
 	"testing"
 	"time"
 
@@ -90,7 +91,7 @@ exit 0
 				fmt.Sprintf(
 					"-aHvh --no-o --no-g --stats --exclude .~tmp~/ "+
 						"--delete --delete-after --delay-updates --safe-links "+
-						"--timeout=120 --contimeout=120 -6 %s %s",
+						"--timeout=120 -6 %s %s",
 					provider.upstreamURL, provider.WorkingDir(),
 				),
 			)
@@ -114,6 +115,7 @@ func TestRsyncProviderWithAuthentication(t *testing.T) {
 		So(err, ShouldBeNil)
 		scriptFile := filepath.Join(tmpDir, "myrsync")
 		tmpFile := filepath.Join(tmpDir, "log_file")
+		proxyAddr := "127.0.0.1:1233"
 
 		c := rsyncConfig{
 			name:         "tuna",
@@ -123,6 +125,7 @@ func TestRsyncProviderWithAuthentication(t *testing.T) {
 			password:     "tunasyncpassword",
 			workingDir:   tmpDir,
 			extraOptions: []string{"--delete-excluded"},
+			rsyncEnv:     map[string]string{"RSYNC_PROXY": proxyAddr},
 			logDir:       tmpDir,
 			logFile:      tmpFile,
 			useIPv4:      true,
@@ -141,7 +144,7 @@ func TestRsyncProviderWithAuthentication(t *testing.T) {
 		Convey("Let's try a run", func() {
 			scriptContent := `#!/bin/bash
 echo "syncing to $(pwd)"
-echo $USER $RSYNC_PASSWORD $@
+echo $USER $RSYNC_PASSWORD $RSYNC_PROXY $@
 sleep 1
 echo "Done"
 exit 0
@@ -156,10 +159,11 @@ exit 0
 					"Done\n",
 				targetDir,
 				fmt.Sprintf(
-					"%s %s -aHvh --no-o --no-g --stats --exclude .~tmp~/ "+
+					"%s %s %s -aHvh --no-o --no-g --stats --exclude .~tmp~/ "+
 						"--delete --delete-after --delay-updates --safe-links "+
-						"--timeout=120 --contimeout=120 -4 --delete-excluded %s %s",
+						"--timeout=120 -4 --delete-excluded %s %s",
-					provider.username, provider.password, provider.upstreamURL, provider.WorkingDir(),
+					provider.username, provider.password, proxyAddr,
+					provider.upstreamURL, provider.WorkingDir(),
 				),
 			)
 
@@ -362,7 +366,7 @@ sleep 5
 
 		})
 	})
-	Convey("Command Provider with fail-on-match regexp should work", t, func(ctx C) {
+	Convey("Command Provider with RegExprs should work", t, func(ctx C) {
 		tmpDir, err := ioutil.TempDir("", "tunasync")
 		defer os.RemoveAll(tmpDir)
 		So(err, ShouldBeNil)
@@ -373,28 +377,73 @@ sleep 5
 			upstreamURL: "http://mirrors.tuna.moe/",
 			command:     "uptime",
 			failOnMatch: "",
+			sizePattern: "",
 			workingDir:  tmpDir,
 			logDir:      tmpDir,
 			logFile:     tmpFile,
 			interval:    600 * time.Second,
 		}
 
-		Convey("when regexp matches", func() {
+		Convey("when fail-on-match regexp matches", func() {
 			c.failOnMatch = `[a-z]+`
 			provider, err := newCmdProvider(c)
 			So(err, ShouldBeNil)
 
+			err = provider.Run()
+			So(err, ShouldNotBeNil)
+			So(provider.DataSize(), ShouldBeEmpty)
+		})
+
+		Convey("when fail-on-match regexp does not match", func() {
+			c.failOnMatch = `load average_`
+			provider, err := newCmdProvider(c)
+			So(err, ShouldBeNil)
+
+			err = provider.Run()
+			So(err, ShouldBeNil)
+		})
+
+		Convey("when fail-on-match regexp meets /dev/null", func() {
+			c.failOnMatch = `load average_`
+			c.logFile = "/dev/null"
+			provider, err := newCmdProvider(c)
+			So(err, ShouldBeNil)
+
 			err = provider.Run()
 			So(err, ShouldNotBeNil)
 		})
 
-		Convey("when regexp does not match", func() {
+		Convey("when size-pattern regexp matches", func() {
-			c.failOnMatch = `load average_`
+			c.sizePattern = `load average: ([\d\.]+)`
 			provider, err := newCmdProvider(c)
 			So(err, ShouldBeNil)
 
 			err = provider.Run()
 			So(err, ShouldBeNil)
+			So(provider.DataSize(), ShouldNotBeEmpty)
+			_, err = strconv.ParseFloat(provider.DataSize(), 32)
+			So(err, ShouldBeNil)
+		})
+
+		Convey("when size-pattern regexp does not match", func() {
+			c.sizePattern = `load ave: ([\d\.]+)`
+			provider, err := newCmdProvider(c)
+			So(err, ShouldBeNil)
+
+			err = provider.Run()
+			So(err, ShouldBeNil)
+			So(provider.DataSize(), ShouldBeEmpty)
+		})
+
+		Convey("when size-pattern regexp meets /dev/null", func() {
+			c.sizePattern = `load ave: ([\d\.]+)`
+			c.logFile = "/dev/null"
+			provider, err := newCmdProvider(c)
+			So(err, ShouldBeNil)
+
+			err = provider.Run()
+			So(err, ShouldNotBeNil)
+			So(provider.DataSize(), ShouldBeEmpty)
 		})
 	})
 }
@@ -457,7 +506,7 @@ exit 0
 				targetDir,
 				fmt.Sprintf(
 					"-aHvh --no-o --no-g --stats --exclude .~tmp~/ --safe-links "+
-						"--timeout=120 --contimeout=120 --exclude dists/ -6 "+
+						"--timeout=120 --exclude dists/ -6 "+
 						"--exclude-from %s %s %s",
 					provider.excludeFile, provider.upstreamURL, provider.WorkingDir(),
 				),
@@ -465,7 +514,7 @@ exit 0
 				fmt.Sprintf(
 					"-aHvh --no-o --no-g --stats --exclude .~tmp~/ "+
 						"--delete --delete-after --delay-updates --safe-links "+
-						"--timeout=120 --contimeout=120 --delete-excluded --cache -6 --exclude-from %s %s %s",
+						"--timeout=120 --delete-excluded --cache -6 --exclude-from %s %s %s",
 					provider.excludeFile, provider.upstreamURL, provider.WorkingDir(),
 				),
 			)
@@ -496,7 +545,7 @@ exit 0
 
 			expectedOutput := fmt.Sprintf(
 				"-aHvh --no-o --no-g --stats --exclude .~tmp~/ --safe-links "+
-					"--timeout=120 --contimeout=120 --exclude dists/ -6 "+
+					"--timeout=120 --exclude dists/ -6 "+
 					"--exclude-from %s %s %s\n",
 				provider.excludeFile, provider.upstreamURL, provider.WorkingDir(),
 			)

worker/rsync_provider.go

@@ -2,7 +2,6 @@ package worker
 
 import (
 	"errors"
-	"io/ioutil"
 	"strings"
 	"time"
 
@@ -15,6 +14,7 @@ type rsyncConfig struct {
 	upstreamURL, username, password, excludeFile string
 	extraOptions                                 []string
 	overriddenOptions                            []string
+	rsyncEnv                                     map[string]string
 	workingDir, logDir, logFile                  string
 	useIPv6, useIPv4                             bool
 	interval                                     time.Duration
@@ -50,12 +50,21 @@ func newRsyncProvider(c rsyncConfig) (*rsyncProvider, error) {
 	if c.rsyncCmd == "" {
 		provider.rsyncCmd = "rsync"
 	}
+	if c.rsyncEnv == nil {
+		provider.rsyncEnv = map[string]string{}
+	}
+	if c.username != "" {
+		provider.rsyncEnv["USER"] = c.username
+	}
+	if c.password != "" {
+		provider.rsyncEnv["RSYNC_PASSWORD"] = c.password
+	}
 
 	options := []string{
 		"-aHvh", "--no-o", "--no-g", "--stats",
 		"--exclude", ".~tmp~/",
 		"--delete", "--delete-after", "--delay-updates",
-		"--safe-links", "--timeout=120", "--contimeout=120",
+		"--safe-links", "--timeout=120",
 	}
 	if c.overriddenOptions != nil {
 		options = c.overriddenOptions
@@ -102,9 +111,7 @@ func (p *rsyncProvider) Run() error {
 	if err := p.Wait(); err != nil {
 		return err
 	}
-	if logContent, err := ioutil.ReadFile(p.LogFile()); err == nil {
+	p.dataSize = internal.ExtractSizeFromRsyncLog(p.LogFile())
-		p.dataSize = internal.ExtractSizeFromRsyncLog(logContent)
-	}
 	return nil
 }
 
@@ -116,18 +123,11 @@ func (p *rsyncProvider) Start() error {
 		return errors.New("provider is currently running")
 	}
 
-	env := map[string]string{}
-	if p.username != "" {
-		env["USER"] = p.username
-	}
-	if p.password != "" {
-		env["RSYNC_PASSWORD"] = p.password
-	}
 	command := []string{p.rsyncCmd}
 	command = append(command, p.options...)
 	command = append(command, p.upstreamURL, p.WorkingDir())
 
-	p.cmd = newCmdJob(p, command, p.WorkingDir(), env)
+	p.cmd = newCmdJob(p, command, p.WorkingDir(), p.rsyncEnv)
 	if err := p.prepareLogFile(false); err != nil {
 		return err
 	}

worker/two_stage_rsync_provider.go

@@ -3,7 +3,6 @@ package worker
 import (
 	"errors"
 	"fmt"
-	"io/ioutil"
 	"strings"
 	"time"
 
@@ -16,6 +15,7 @@ type twoStageRsyncConfig struct {
 	stage1Profile                                string
 	upstreamURL, username, password, excludeFile string
 	extraOptions                                 []string
+	rsyncEnv                                     map[string]string
 	workingDir, logDir, logFile                  string
 	useIPv6                                      bool
 	interval                                     time.Duration
@@ -59,16 +59,25 @@ func newTwoStageRsyncProvider(c twoStageRsyncConfig) (*twoStageRsyncProvider, er
 		stage1Options: []string{
 			"-aHvh", "--no-o", "--no-g", "--stats",
 			"--exclude", ".~tmp~/",
-			"--safe-links", "--timeout=120", "--contimeout=120",
+			"--safe-links", "--timeout=120",
 		},
 		stage2Options: []string{
 			"-aHvh", "--no-o", "--no-g", "--stats",
 			"--exclude", ".~tmp~/",
 			"--delete", "--delete-after", "--delay-updates",
-			"--safe-links", "--timeout=120", "--contimeout=120",
+			"--safe-links", "--timeout=120",
 		},
 	}
 
+	if c.rsyncEnv == nil {
+		provider.rsyncEnv = map[string]string{}
+	}
+	if c.username != "" {
+		provider.rsyncEnv["USER"] = c.username
+	}
+	if c.password != "" {
+		provider.rsyncEnv["RSYNC_PASSWORD"] = c.password
+	}
 	if c.rsyncCmd == "" {
 		provider.rsyncCmd = "rsync"
 	}
@@ -132,14 +141,6 @@ func (p *twoStageRsyncProvider) Run() error {
 		return errors.New("provider is currently running")
 	}
 
-	env := map[string]string{}
-	if p.username != "" {
-		env["USER"] = p.username
-	}
-	if p.password != "" {
-		env["RSYNC_PASSWORD"] = p.password
-	}
-
 	p.dataSize = ""
 	stages := []int{1, 2}
 	for _, stage := range stages {
@@ -151,7 +152,7 @@ func (p *twoStageRsyncProvider) Run() error {
 		command = append(command, options...)
 		command = append(command, p.upstreamURL, p.WorkingDir())
 
-		p.cmd = newCmdJob(p, command, p.WorkingDir(), env)
+		p.cmd = newCmdJob(p, command, p.WorkingDir(), p.rsyncEnv)
 		if err := p.prepareLogFile(stage > 1); err != nil {
 			return err
 		}
@@ -169,8 +170,6 @@ func (p *twoStageRsyncProvider) Run() error {
 			return err
 		}
 	}
-	if logContent, err := ioutil.ReadFile(p.LogFile()); err == nil {
+	p.dataSize = internal.ExtractSizeFromRsyncLog(p.LogFile())
-		p.dataSize = internal.ExtractSizeFromRsyncLog(logContent)
-	}
 	return nil
 }