feature(worker): runMirrorJob no longer controls the interval

internal/logger.go

@@ -10,9 +10,9 @@ import (
 func InitLogger(verbose, debug, withSystemd bool) {
 	var fmtString string
 	if withSystemd {
-		fmtString = "\r[%{level:.6s}] %{message}"
+		fmtString = "[%{level:.6s}] %{message}"
 	} else {
-		fmtString = "\r%{color}[%{time:06-01-02 15:04:05}][%{level:.6s}]%{color:reset} %{message}"
+		fmtString = "%{color}[%{time:06-01-02 15:04:05}][%{level:.6s}][%{shortfile}]%{color:reset} %{message}"
 	}
 	format := logging.MustStringFormatter(fmtString)
 	logging.SetFormatter(format)

worker/job.go

@@ -1,9 +1,6 @@
 package worker
 
-import (
-	"errors"
-	"time"
-)
+import "errors"
 
 // this file contains the workflow of a mirror jb
 
@@ -41,7 +38,8 @@ func runMirrorJob(provider mirrorProvider, ctrlChan <-chan ctrlAction, managerCh
 	}
 
 	runJobWrapper := func(kill <-chan empty, jobDone chan<- empty) error {
-		defer func() { jobDone <- empty{} }()
+		defer close(jobDone)
+
 		logger.Info("start syncing: %s", provider.Name())
 
 		Hooks := provider.Hooks()
@@ -89,6 +87,7 @@ func runMirrorJob(provider mirrorProvider, ctrlChan <-chan ctrlAction, managerCh
 			case syncErr = <-syncDone:
 				logger.Debug("syncing done")
 			case <-kill:
+				logger.Debug("received kill")
 				stopASAP = true
 				err := provider.Terminate()
 				if err != nil {
@@ -118,15 +117,18 @@ func runMirrorJob(provider mirrorProvider, ctrlChan <-chan ctrlAction, managerCh
 			}
 
 			// syncing failed
-			logger.Info("failed syncing %s: %s", provider.Name(), err.Error())
+			logger.Warning("failed syncing %s: %s", provider.Name(), syncErr.Error())
 			managerChan <- struct{}{}
+
 			// post-fail hooks
+			logger.Debug("post-fail hooks")
 			err = runHooks(rHooks, func(h jobHook) error { return h.postFail() }, "post-fail")
 			if err != nil {
 				return err
 			}
 			// gracefully exit
 			if stopASAP {
+				logger.Debug("No retry, exit directly")
 				return nil
 			}
 			// continue to next retry
@@ -140,6 +142,7 @@ func runMirrorJob(provider mirrorProvider, ctrlChan <-chan ctrlAction, managerCh
 			defer func() { semaphore <- empty{} }()
 			runJobWrapper(kill, jobDone)
 		case <-kill:
+			jobDone <- empty{}
 			return
 		}
 	}
@@ -160,12 +163,15 @@ func runMirrorJob(provider mirrorProvider, ctrlChan <-chan ctrlAction, managerCh
 				case jobStop:
 					enabled = false
 					close(kill)
+					<-jobDone
 				case jobDisable:
 					close(kill)
+					<-jobDone
 					return nil
 				case jobRestart:
 					enabled = true
 					close(kill)
+					<-jobDone
 					continue
 				case jobStart:
 					enabled = true
@@ -178,10 +184,7 @@ func runMirrorJob(provider mirrorProvider, ctrlChan <-chan ctrlAction, managerCh
 			}
 		}
 
-		select {
-		case <-time.After(provider.Interval()):
-			continue
-		case ctrl := <-ctrlChan:
+		ctrl := <-ctrlChan
 		switch ctrl {
 		case jobStop:
 			enabled = false
@@ -196,7 +199,4 @@ func runMirrorJob(provider mirrorProvider, ctrlChan <-chan ctrlAction, managerCh
 			return nil
 		}
 	}
-	}
-
-	return nil
 }

worker/job_test.go

@@ -9,10 +9,13 @@ import (
 	"time"
 
 	. "github.com/smartystreets/goconvey/convey"
+	. "github.com/tuna/tunasync/internal"
 )
 
 func TestMirrorJob(t *testing.T) {
 
+	InitLogger(true, true, false)
+
 	Convey("MirrorJob should work", t, func(ctx C) {
 		tmpDir, err := ioutil.TempDir("", "tunasync")
 		defer os.RemoveAll(tmpDir)
@@ -71,6 +74,7 @@ func TestMirrorJob(t *testing.T) {
 					loggedContent, err := ioutil.ReadFile(provider.LogFile())
 					So(err, ShouldBeNil)
 					So(string(loggedContent), ShouldEqual, exceptedOutput)
+					ctrlChan <- jobStart
 				}
 				select {
 				case <-managerChan:
@@ -107,7 +111,7 @@ echo $TUNASYNC_WORKING_DIR
 				go runMirrorJob(provider, ctrlChan, managerChan, semaphore)
 				time.Sleep(1 * time.Second)
 				ctrlChan <- jobStop
-				time.Sleep(1 * time.Second)
+				<-managerChan
 				exceptedOutput := fmt.Sprintf("%s\n", provider.WorkingDir())
 				loggedContent, err := ioutil.ReadFile(provider.LogFile())
 				So(err, ShouldBeNil)