Merge branch 'wip-newlog'

cmd/tunasynctl/tunasynctl.go

@@ -32,7 +32,7 @@ const (
 	userCfgFile   = "$HOME/.config/tunasync/ctl.conf" // user-specific conf
 )
 
-var logger = logging.MustGetLogger("tunasynctl-cmd")
+var logger = logging.MustGetLogger("tunasynctl")
 
 var baseURL string
 var client *http.Client
@@ -67,7 +67,7 @@ func loadConfig(cfgFile string, cfg *config) error {
 
 func initialize(c *cli.Context) error {
 	// init logger
-	tunasync.InitLogger(c.Bool("verbose"), c.Bool("verbose"), false)
+	tunasync.InitLogger(c.Bool("verbose"), c.Bool("debug"), false)
 
 	cfg := new(config)
 
@@ -79,6 +79,7 @@ func initialize(c *cli.Context) error {
 	if _, err := os.Stat(systemCfgFile); err == nil {
 		loadConfig(systemCfgFile, cfg)
 	}
+	logger.Debug("user config file: %s", os.ExpandEnv(userCfgFile))
 	if _, err := os.Stat(os.ExpandEnv(userCfgFile)); err == nil {
 		loadConfig(os.ExpandEnv(userCfgFile), cfg)
 	}
@@ -174,15 +175,22 @@ func listJobs(c *cli.Context) error {
 			}(workerID)
 		}
 		for range args {
-			jobs = append(jobs, <-ans...)
+			job := <-ans
+			if job == nil {
+				return cli.NewExitError(
+					fmt.Sprintf("Failed to correctly get information "+
+						"of jobs from at least one manager"),
+					1)
+			}
+			jobs = append(jobs, job...)
 		}
 		genericJobs = jobs
 	}
 
 	b, err := json.MarshalIndent(genericJobs, "", "  ")
 	if err != nil {
 		return cli.NewExitError(
-			fmt.Sprintf("Error printing out informations: %s", err.Error()),
+			fmt.Sprintf("Error printing out information: %s", err.Error()),
 			1)
 	}
 	fmt.Println(string(b))
@@ -236,7 +244,7 @@ func updateMirrorSize(c *cli.Context) error {
 		)
 	}
 
-	logger.Infof("Successfully updated mirror size to %s", mirrorSize)
+	fmt.Printf("Successfully updated mirror size to %s\n", mirrorSize)
 	return nil
 }
 
@@ -279,9 +287,9 @@ func removeWorker(c *cli.Context) error {
 	res := map[string]string{}
 	err = json.NewDecoder(resp.Body).Decode(&res)
 	if res["message"] == "deleted" {
-		logger.Info("Successfully removed the worker")
+		fmt.Println("Successfully removed the worker")
 	} else {
-		logger.Info("Failed to remove the worker")
+		return cli.NewExitError("Failed to remove the worker", 1)
 	}
 	return nil
 }
@@ -314,7 +322,7 @@ func flushDisabledJobs(c *cli.Context) error {
 			1)
 	}
 
-	logger.Info("Successfully flushed disabled jobs")
+	fmt.Println("Successfully flushed disabled jobs")
 	return nil
 }
 
@@ -367,7 +375,7 @@ func cmdJob(cmd tunasync.CmdVerb) cli.ActionFunc {
 				" command: HTTP status code is not 200: %s", body),
 				1)
 		}
-		logger.Info("Succesfully send command")
+		fmt.Println("Successfully send the command")
 
 		return nil
 	}
@@ -405,7 +413,7 @@ func cmdWorker(cmd tunasync.CmdVerb) cli.ActionFunc {
 				" command: HTTP status code is not 200: %s", body),
 				1)
 		}
-		logger.Info("Succesfully send command")
+		fmt.Println("Successfully send the command")
 
 		return nil
 	}
@@ -462,6 +470,10 @@ func main() {
 			Name:  "verbose, v",
 			Usage: "Enable verbosely logging",
 		},
+		cli.BoolFlag{
+			Name:  "debug",
+			Usage: "Enable debugging logging",
+		},
 	}
 	cmdFlags := []cli.Flag{
 		cli.StringFlag{

internal/logger.go

@@ -24,12 +24,12 @@ func InitLogger(verbose, debug, withSystemd bool) {
 
 	if debug {
 		logging.SetLevel(logging.DEBUG, "tunasync")
-		logging.SetLevel(logging.DEBUG, "tunasynctl-cmd")
+		logging.SetLevel(logging.DEBUG, "tunasynctl")
 	} else if verbose {
 		logging.SetLevel(logging.INFO, "tunasync")
-		logging.SetLevel(logging.INFO, "tunasynctl-cmd")
+		logging.SetLevel(logging.INFO, "tunasynctl")
 	} else {
 		logging.SetLevel(logging.NOTICE, "tunasync")
-		logging.SetLevel(logging.NOTICE, "tunasynctl-cmd")
+		logging.SetLevel(logging.NOTICE, "tunasynctl")
 	}
 }

internal/util.go

@@ -6,12 +6,37 @@ import (
 	"crypto/x509"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io/ioutil"
 	"net/http"
+	"os/exec"
 	"regexp"
 	"time"
 )
 
+var rsyncExitValues = map[int]string{
+	0:  "Success",
+	1:  "Syntax or usage error",
+	2:  "Protocol incompatibility",
+	3:  "Errors selecting input/output files, dirs",
+	4:  "Requested action not supported: an attempt was made to manipulate 64-bit files on a platform that cannot support them; or an option was specified that is supported by the client and not by the server.",
+	5:  "Error starting client-server protocol",
+	6:  "Daemon unable to append to log-file",
+	10: "Error in socket I/O",
+	11: "Error in file I/O",
+	12: "Error in rsync protocol data stream",
+	13: "Errors with program diagnostics",
+	14: "Error in IPC code",
+	20: "Received SIGUSR1 or SIGINT",
+	21: "Some error returned by waitpid()",
+	22: "Error allocating core memory buffers",
+	23: "Partial transfer due to error",
+	24: "Partial transfer due to vanished source files",
+	25: "The --max-delete limit stopped deletions",
+	30: "Timeout in data send/receive",
+	35: "Timeout waiting for daemon connection",
+}
+
 // GetTLSConfig generate tls.Config from CAFile
 func GetTLSConfig(CAFile string) (*tls.Config, error) {
 	caCert, err := ioutil.ReadFile(CAFile)
@@ -115,3 +140,16 @@ func ExtractSizeFromRsyncLog(logFile string) string {
 	re := regexp.MustCompile(`(?m)^Total file size: ([0-9\.]+[KMGTP]?) bytes`)
 	return ExtractSizeFromLog(logFile, re)
 }
+
+// TranslateRsyncErrorCode translates the exit code of rsync to a message
+func TranslateRsyncErrorCode(cmdErr error) (exitCode int, msg string) {
+
+	if exiterr, ok := cmdErr.(*exec.ExitError); ok {
+		exitCode = exiterr.ExitCode()
+		strerr, valid := rsyncExitValues[exitCode]
+		if valid {
+			msg = fmt.Sprintf("rsync error: %s", strerr)
+		}
+	}
+	return
+}

worker/base_provider.go

@@ -19,6 +19,7 @@ type baseProvider struct {
 	isMaster bool
 
 	cmd       *cmdJob
+	logFileFd *os.File
 	isRunning atomic.Value
 
 	cgroup *cgroupHook
@@ -128,10 +129,19 @@ func (p *baseProvider) prepareLogFile(append bool) error {
 		logger.Errorf("Error opening logfile %s: %s", p.LogFile(), err.Error())
 		return err
 	}
+	p.logFileFd = logFile
 	p.cmd.SetLogFile(logFile)
 	return nil
 }
 
+func (p *baseProvider) closeLogFile() (err error) {
+	if p.logFileFd != nil {
+		err = p.logFileFd.Close()
+		p.logFileFd = nil
+	}
+	return
+}
+
 func (p *baseProvider) Run() error {
 	panic("Not Implemented")
 }

worker/cmd_provider.go

@@ -88,6 +88,7 @@ func (p *cmdProvider) DataSize() string {
 
 func (p *cmdProvider) Run() error {
 	p.dataSize = ""
+	defer p.closeLogFile()
 	if err := p.Start(); err != nil {
 		return err
 	}

worker/provider_test.go

@@ -106,6 +106,34 @@ exit 0
 		})
 
 	})
+	Convey("If the rsync program fails", t, func() {
+		tmpDir, err := ioutil.TempDir("", "tunasync")
+		defer os.RemoveAll(tmpDir)
+		So(err, ShouldBeNil)
+		tmpFile := filepath.Join(tmpDir, "log_file")
+
+		Convey("in the rsyncProvider", func() {
+
+			c := rsyncConfig{
+				name:         "tuna",
+				upstreamURL:  "rsync://rsync.tuna.moe/tuna/",
+				workingDir:   tmpDir,
+				logDir:       tmpDir,
+				logFile:      tmpFile,
+				extraOptions: []string{"--somethine-invalid"},
+				interval:     600 * time.Second,
+			}
+
+			provider, err := newRsyncProvider(c)
+			So(err, ShouldBeNil)
+
+			err = provider.Run()
+			So(err, ShouldNotBeNil)
+			loggedContent, err := ioutil.ReadFile(provider.LogFile())
+			So(err, ShouldBeNil)
+			So(string(loggedContent), ShouldContainSubstring, "Syntax or usage error")
+		})
+	})
 }
 
 func TestRsyncProviderWithAuthentication(t *testing.T) {
@@ -556,4 +584,34 @@ exit 0
 			// fmt.Println(string(loggedContent))
 		})
 	})
+
+	Convey("If the rsync program fails", t, func(ctx C) {
+		tmpDir, err := ioutil.TempDir("", "tunasync")
+		defer os.RemoveAll(tmpDir)
+		So(err, ShouldBeNil)
+		tmpFile := filepath.Join(tmpDir, "log_file")
+
+		Convey("in the twoStageRsyncProvider", func() {
+
+			c := twoStageRsyncConfig{
+				name:          "tuna-two-stage-rsync",
+				upstreamURL:   "rsync://0.0.0.1/",
+				stage1Profile: "debian",
+				workingDir:    tmpDir,
+				logDir:        tmpDir,
+				logFile:       tmpFile,
+				excludeFile:   tmpFile,
+			}
+
+			provider, err := newTwoStageRsyncProvider(c)
+			So(err, ShouldBeNil)
+
+			err = provider.Run()
+			So(err, ShouldNotBeNil)
+			loggedContent, err := ioutil.ReadFile(provider.LogFile())
+			So(err, ShouldBeNil)
+			So(string(loggedContent), ShouldContainSubstring, "Error in socket I/O")
+
+		})
+	})
 }

worker/rsync_provider.go

@@ -105,10 +105,18 @@ func (p *rsyncProvider) DataSize() string {
 
 func (p *rsyncProvider) Run() error {
 	p.dataSize = ""
+	defer p.closeLogFile()
 	if err := p.Start(); err != nil {
 		return err
 	}
 	if err := p.Wait(); err != nil {
+		code, msg := internal.TranslateRsyncErrorCode(err)
+		if code != 0 {
+			logger.Debug("Rsync exitcode %d (%s)", code, msg)
+			if p.logFileFd != nil {
+				p.logFileFd.WriteString(msg + "\n")
+			}
+		}
 		return err
 	}
 	p.dataSize = internal.ExtractSizeFromRsyncLog(p.LogFile())

worker/runner.go

@@ -118,9 +118,6 @@ func (c *cmdJob) Wait() error {
 		return c.retErr
 	default:
 		err := c.cmd.Wait()
-		if c.cmd.Stdout != nil {
-			c.cmd.Stdout.(*os.File).Close()
-		}
 		c.retErr = err
 		close(c.finished)
 		return err

worker/two_stage_rsync_provider.go

@@ -156,6 +156,7 @@ func (p *twoStageRsyncProvider) Run() error {
 		if err := p.prepareLogFile(stage > 1); err != nil {
 			return err
 		}
+		defer p.closeLogFile()
 
 		if err = p.cmd.Start(); err != nil {
 			return err
@@ -167,6 +168,13 @@ func (p *twoStageRsyncProvider) Run() error {
 		err = p.Wait()
 		p.Lock()
 		if err != nil {
+			code, msg := internal.TranslateRsyncErrorCode(err)
+			if code != 0 {
+				logger.Debug("Rsync exitcode %d (%s)", code, msg)
+				if p.logFileFd != nil {
+					p.logFileFd.WriteString(msg + "\n")
+				}
+			}
 			return err
 		}
 	}